Techsystech
/
server-tools
mirror of https://github.com/OCA/server-tools.git
3
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Migrate and refactore external_file_location

pull/2632/head
Florian da Costa 2019-06-24 19:32:25 +02:00
parent 149de2ed96
commit f022064a47
20 changed files with 854 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
attachment_synchronize/README.rst 100644
View File

@ -0,0 +1,69 @@
.. image:: https://img.shields.io/badge/licence-AGPL--3-blue.svg
:target: http://www.gnu.org/licenses/agpl-3.0-standalone.html
:alt: License: AGPL-3
======================
Attachment Synchronize
======================
This module was written to extend the functionality of ir.attachment to support remote communication and allow you to import/export file to a remote server.
For now, FTP, SFTP and local filestore are handled by the module.
Installation
============
To install this module, you need to:
* fs python module at version 0.5.4 or under
* Paramiko python module
Usage
=====
To use this module, you need to:
* Add a location with your server infos
* Create a task with your file info and remote communication method
* A cron task will trigger each task
.. image:: https://odoo-community.org/website/image/ir.attachment/5784_f2813bd/datas
:alt: Try me on Runbot
:target: https://runbot.odoo-community.org/runbot/149/9.0
Bug Tracker
===========
Bugs are tracked on `GitHub Issues
<https://github.com/OCA/server-tools/issues>`_. In case of trouble, please
check there if your issue has already been reported. If you spotted it first,
help us smashing it by providing a detailed and welcomed feedback.
Credits
=======
Images
------
* Odoo Community Association: `Icon <https://github.com/OCA/maintainer-tools/blob/master/template/module/static/description/icon.svg>`_.
Contributors
------------
* Valentin CHEMIERE <valentin.chemiere@akretion.com>
* Mourad EL HADJ MIMOUNE <mourad.elhadj.mimoune@akretion.com>
* Florian DA COSTA <florian.dacosta@akretion.com>
Maintainer
----------
.. image:: https://odoo-community.org/logo.png
:alt: Odoo Community Association
:target: https://odoo-community.org
This module is maintained by the OCA.
OCA, or the Odoo Community Association, is a nonprofit organization whose
mission is to support the collaborative development of Odoo features and
promote its widespread use.
To contribute to this module, please visit https://odoo-community.org.

1
attachment_synchronize/__init__.py 100644
View File

@ -0,0 +1 @@
from . import models

28
attachment_synchronize/__manifest__.py 100644
View File

@ -0,0 +1,28 @@
# @ 2016 florian DA COSTA @ Akretion
# © 2016 @author Mourad EL HADJ MIMOUNE <mourad.elhadj.mimoune@akretion.com>
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
{
'name': 'Attachment Synchronize',
'version': '12.0.1.0.0',
'author': 'Akretion,Odoo Community Association (OCA)',
'website': 'https://github.com/oca/server-tools',
'license': 'AGPL-3',
'category': 'Generic Modules',
'depends': [
'base_attachment_queue',
'storage_backend',
],
'data': [
'views/attachment_view.xml',
'views/task_view.xml',
'views/storage_backend_view.xml',
'data/cron.xml',
'security/ir.model.access.csv',
],
'demo': [
# 'demo/task_demo.xml',
],
'installable': True,
'application': False,
}

16
attachment_synchronize/data/cron.xml 100644
View File

@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<odoo noupdate="1">
<record model="ir.cron" id="cronjob_run_exchange_tasks">
<field name='name'>Run attachment tasks import</field>
<field name='interval_number'>30</field>
<field name='interval_type'>minutes</field>
<field name="numbercall">-1</field>
<field name="active">False</field>
<field name="doall" eval="False" />
<field name="model_id" ref="model_storage_backend_task"/>
<field name="state">code</field>
<field name="code">model.run_task_scheduler([('method_type', '=', 'import')])</field>
</record>
</odoo>

95
attachment_synchronize/demo/task_demo.xml 100644
View File

@ -0,0 +1,95 @@
<?xml version="1.0" encoding="UTF-8"?>
<odoo noupdate="1">
<record id="location_ftp" model="external.file.location">
<field name="name">TEST FTP</field>
<field name="protocol">ftp</field>
<field name="address">my-ftp-address</field>
<field name="login">my-ftp-user</field>
<field name="password">my-ftp-password</field>
<field name="port">21</field>
</record>
<record id="location_sftp" model="external.file.location">
<field name="name">TEST SFTP</field>
<field name="protocol">sftp</field>
<field name="address">my-sftp-address</field>
<field name="login">my-sftp-user</field>
<field name="password">my-sftp-password</field>
<field name="port">22</field>
</record>
<record id="location_filestore" model="external.file.location">
<field name="name">TEST File Store</field>
<field name="protocol">file_store</field>
<field name="filestore_rootpath">/</field>
</record>
<record id="ftp_import_task" model="external.file.task">
<field name="method_type">import</field>
<field name="location_id" eval="ref('external_file_location.location_ftp')"/>
<field name="filename">test-import-ftp.txt</field>
<field name="filepath">/home/user/test</field>
<field name="name">Import FTP Task</field>
</record>
<record id="ftp_export_task" model="external.file.task">
<field name="method_type">export</field>
<field name="location_id" eval="ref('external_file_location.location_ftp')"/>
<field name="filepath">/home/user/test</field>
<field name="name">Export FTP Task</field>
</record>
<record id="sftp_import_task" model="external.file.task">
<field name="method_type">import</field>
<field name="location_id" eval="ref('external_file_location.location_sftp')"/>
<field name="filename">test-import-sftp.txt</field>
<field name="filepath">/home/user/test</field>
<field name="name">Import SFTP Task</field>
</record>
<record id="sftp_export_task" model="external.file.task">
<field name="method_type">export</field>
<field name="location_id" eval="ref('external_file_location.location_sftp')"/>
<field name="filepath">/home/user/test</field>
<field name="name">Export SFTP Task</field>
</record>
<record id="filestore_import_task" model="external.file.task">
<field name="method_type">import</field>
<field name="location_id" eval="ref('external_file_location.location_filestore')"/>
<field name="filename">test-import-filestore.txt</field>
<field name="filepath">/home/user/test</field>
<field name="name">Import filestore Task</field>
</record>
<record id="filestore_export_task" model="external.file.task">
<field name="method_type">export</field>
<field name="location_id" eval="ref('external_file_location.location_filestore')"/>
<field name="filepath">/home/user/test</field>
<field name="name">Export filestore Task</field>
</record>
<record id="ir_attachment_export_file_sftp" model="ir.attachment.metadata">
<field name="name">Sftp text export file</field>
<field name="datas">dGVzdCBzZnRwIGZpbGUgZXhwb3J0</field>
<field name="datas_fname">sftp_test_export.txt</field>
<field name="task_id" eval="ref('external_file_location.sftp_export_task')"/>
<field name="file_type">export_external_location</field>
</record>
<record id="ir_attachment_export_file_ftp" model="ir.attachment.metadata">
<field name="name">ftp text export file</field>
<field name="datas">dGVzdCBmdHAgZmlsZSBleHBvcnQ=</field>
<field name="datas_fname">ftp_test_export.txt</field>
<field name="task_id" eval="ref('external_file_location.ftp_export_task')"/>
<field name="file_type">export_external_location</field>
</record>
<record id="ir_attachment_export_file_filestore" model="ir.attachment.metadata">
<field name="name">filestore text export file</field>
<field name="datas">dGVzdCBmaWxlc3RvcmUgZmlsZSBleHBvcnQ=</field>
<field name="datas_fname">filestore_test_export.txt</field>
<field name="task_id" eval="ref('external_file_location.filestore_export_task')"/>
<field name="file_type">export_external_location</field>
</record>
</odoo>

3
attachment_synchronize/models/__init__.py 100644
View File

@ -0,0 +1,3 @@
from . import attachment
from . import task
from . import storage_backend

26
attachment_synchronize/models/attachment.py 100644
View File

@ -0,0 +1,26 @@
# @ 2016 Florian DA COSTA @ Akretion
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
from odoo import models, fields, api
import os
class IrAttachmentMetadata(models.Model):
_inherit = 'ir.attachment.metadata'
task_id = fields.Many2one('storage.backend.task', string='Task')
storage_backend_id = fields.Many2one(
'storage.backend', string='Storage Backend',
related='task_id.backend_id', store=True)
file_type = fields.Selection(
selection_add=[
('export',
'Export File (External location)')
])
@api.multi
def _run(self):
super(IrAttachmentMetadata, self)._run()
if self.file_type == 'export':
path = os.path.join(self.task_id.filepath, self.datas_fname)
self.storage_backend_id._add_b64_data(path, self.datas)

11
attachment_synchronize/models/storage_backend.py 100644
View File

@ -0,0 +1,11 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
from odoo import fields, models
class StorageBackend(models.Model):
_inherit = "storage.backend"
task_ids = fields.One2many(
"storage.backend.task", "backend_id",
string="Tasks")

169
attachment_synchronize/models/task.py 100644
View File

@ -0,0 +1,169 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
from odoo import models, fields, api
import odoo
from odoo import tools
from base64 import b64encode
import os
import datetime
import logging
_logger = logging.getLogger(__name__)
try:
# We use a jinja2 sandboxed environment to render mako templates.
# Note that the rendering does not cover all the mako syntax, in particular
# arbitrary Python statements are not accepted, and not all expressions are
# allowed: only "public" attributes (not starting with '_') of objects may
# be accessed.
# This is done on purpose: it prevents incidental or malicious execution of
# Python code that may break the security of the server.
from jinja2.sandbox import SandboxedEnvironment
mako_template_env = SandboxedEnvironment(
variable_start_string="${",
variable_end_string="}",
line_statement_prefix="%",
trim_blocks=True, # do not output newline after blocks
)
mako_template_env.globals.update({
'str': str,
'datetime': datetime,
'len': len,
'abs': abs,
'min': min,
'max': max,
'sum': sum,
'filter': filter,
'map': map,
'round': round,
})
except ImportError:
_logger.warning("jinja2 not available, templating features will not work!")
class StorageTask(models.Model):
_name = 'storage.backend.task'
_description = 'Storage Backend task'
name = fields.Char(required=True)
method_type = fields.Selection(
[('import', 'Import'), ('export', 'Export')],
required=True)
filename = fields.Char(help='File name which is imported.'
'The system will check if the remote file at '
'least contains the pattern in its name. '
'Leave it empty to import all files')
filepath = fields.Char(help='Path to imported/exported file')
backend_id = fields.Many2one(
'storage.backend', string='Backend', required=True)
attachment_ids = fields.One2many('ir.attachment.metadata', 'task_id',
string='Attachment')
move_path = fields.Char(string='Move Path',
help='Imported File will be moved to this path')
new_name = fields.Char(string='New Name',
help='Imported File will be renamed to this name\n'
'Name can use mako template where obj is an '
'ir_attachement. template exemple : '
' ${obj.name}-${obj.create_date}.csv')
after_import = fields.Selection(
selection=[
('rename', 'Rename'),
('move', 'Move'),
('move_rename', 'Move & Rename'),
('delete', 'Delete'),
], help='Action after import a file')
file_type = fields.Selection(
selection=[],
string="File Type",
help="The file type determines an import method to be used "
"to parse and transform data before their import in ERP")
active = fields.Boolean(default=True)
@api.multi
def _prepare_attachment_vals(self, datas, filename):
self.ensure_one()
vals = {
'name': filename,
'datas': datas,
'datas_fname': filename,
'task_id': self.id,
'file_type': self.file_type or False,
}
return vals
@api.model
def _template_render(self, template, record):
try:
template = mako_template_env.from_string(tools.ustr(template))
except Exception:
_logger.exception("Failed to load template %r", template)
variables = {'obj': record}
try:
render_result = template.render(variables)
except Exception:
_logger.exception(
"Failed to render template %r using values %r" %
(template, variables))
render_result = u""
if render_result == u"False":
render_result = u""
return render_result
@api.model
def run_task_scheduler(self, domain=None):
if domain is None:
domain = []
domain.append([('method_type', '=', 'import')])
tasks = self.env['storage.backend'].search(domain)
for task in tasks:
task.run_import()
@api.multi
def run_import(self):
self.ensure_one()
attach_obj = self.env['ir.attachment.metadata']
backend = self.backend_id
all_filenames = backend._list(relative_path=self.filepath)
if self.filename:
filenames = [x for x in all_filenames if self.filename in x]
for file_name in filenames:
with api.Environment.manage():
with odoo.registry(
self.env.cr.dbname).cursor() as new_cr:
new_env = api.Environment(new_cr, self.env.uid,
self.env.context)
try:
full_absolute_path = os.path.join(
self.filepath, file_name)
datas = backend._get_b64_data(full_absolute_path)
attach_vals = self._prepare_attachment_vals(
datas, file_name)
attachment = attach_obj.with_env(new_env).create(
attach_vals)
new_full_path = False
if self.after_import == 'rename':
new_name = self._template_render(
self.new_name, attachment)
new_full_path = os.path.join(
self.filepath, new_name)
elif self.after_import == 'move':
new_full_path = os.path.join(
self.move_path, file_name)
elif self.after_import == 'move_rename':
new_name = self._template_render(
self.new_name, attachment)
new_full_path = os.path.join(
self.move_path, new_name)
if new_full_path:
backend._add_b64_data(new_full_path, datas)
if self.after_import in (
'delete', 'rename', 'move', 'move_rename'
):
backend._delete(full_absolute_path)
except Exception as e:
new_env.cr.rollback()
raise e
else:
new_env.cr.commit()

2
attachment_synchronize/security/ir.model.access.csv 100644
View File

@ -0,0 +1,2 @@
id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
access_storage_backend_task_manager,storage.backend.task.manager,model_storage_backend_task,base.group_system,1,1,1,1
1 id name model_id:id group_id:id perm_read perm_write perm_create perm_unlink
2 access_storage_backend_task_manager storage.backend.task.manager model_storage_backend_task base.group_system 1 1 1 1

BIN
attachment_synchronize/static/description/icon.png 100644
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 9.2 KiB

4
attachment_synchronize/tests/__init__.py 100644
View File

@ -0,0 +1,4 @@
from . import mock_server
from . import test_ftp
from . import test_sftp
from . import test_filestore

32
attachment_synchronize/tests/common.py 100644
View File

@ -0,0 +1,32 @@
# coding: utf-8
# @ 2016 Florian da Costa @ Akretion
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import openerp.tests.common as common
from openerp import api
from StringIO import StringIO
class ContextualStringIO(StringIO):
"""
snippet from http://bit.ly/1HfH6uW (stackoverflow)
"""
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
return False
class TestConnection(common.TransactionCase):
def setUp(self):
super(TestConnection, self).setUp()
self.registry.enter_test_mode()
self.env = api.Environment(self.registry.test_cr, self.env.uid,
self.env.context)
def tearDown(self):
self.registry.leave_test_mode()
super(TestConnection, self).tearDown()

74
attachment_synchronize/tests/mock_server.py 100644
View File

@ -0,0 +1,74 @@
# coding: utf-8
# Copyright (C) 2014 initOS GmbH & Co. KG (<http://www.initos.com>).
# @ 2015 Valentin CHEMIERE @ Akretion
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import mock
from contextlib import contextmanager
from collections import defaultdict
class MultiResponse(dict):
pass
class ConnMock(object):
def __init__(self, response):
self.response = response
self._calls = []
self.call_count = defaultdict(int)
def __getattribute__(self, method):
if method not in ('_calls', 'response', 'call_count'):
def callable(*args, **kwargs):
self._calls.append({
'method': method,
'args': args,
'kwargs': kwargs,
})
call = self.response[method]
if isinstance(call, MultiResponse):
call = call[self.call_count[method]]
self.call_count[method] += 1
return call
return callable
else:
return super(ConnMock, self).__getattribute__(method)
def __call__(self, *args, **kwargs):
return self
def __enter__(self, *args, **kwargs):
return self
def __exit__(self, *args, **kwargs):
pass
def __repr__(self, *args, **kwargs):
return self
def __getitem__(self, key):
return
@contextmanager
def server_mock_sftp(response):
with mock.patch('openerp.addons.external_file_location.tasks.sftp.'
'SftpTask', ConnMock(response)) as SFTPFS:
yield SFTPFS._calls
@contextmanager
def server_mock_ftp(response):
with mock.patch('openerp.addons.external_file_location.tasks.ftp.'
'FtpTask', ConnMock(response)) as FTPFS:
yield FTPFS._calls
@contextmanager
def server_mock_filestore(response):
with mock.patch('openerp.addons.external_file_location.tasks.filestore.'
'FileStoreTask', ConnMock(response)) as FTPFS:
yield FTPFS._calls

50
attachment_synchronize/tests/test_filestore.py 100644
View File

@ -0,0 +1,50 @@
# coding: utf-8
# @ 2015 Valentin CHEMIERE @ Akretion
# ©2016 @author Mourad EL HADJ MIMOUNE <mourad.elhadj.mimoune@akretion.com>
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import logging
from base64 import b64decode
from .common import TestConnection, ContextualStringIO
from .mock_server import server_mock_filestore
_logger = logging.getLogger(__name__)
class TestfilestoreConnection(TestConnection):
def setUp(self):
super(TestfilestoreConnection, self).setUp()
self.test_file_filestore = ContextualStringIO()
self.test_file_filestore.write('import filestore')
self.test_file_filestore.seek(0)
def test_00_filestore_import(self):
self.task = self.env.ref(
'external_file_location.filestore_import_task')
with server_mock_filestore(
{'open': self.test_file_filestore,
'listdir': ['test-import-filestore.txt']}):
self.task.run_import()
search_file = self.env['ir.attachment.metadata'].search(
[('name', '=', 'test-import-filestore.txt')])
self.assertEqual(len(search_file), 1)
self.assertEqual(b64decode(search_file[0].datas), 'import filestore')
def test_01_filestore_export(self):
self.task = self.env.ref(
'external_file_location.filestore_export_task')
self.filestore_attachment = self.env.ref(
'external_file_location.ir_attachment_export_file_filestore')
with server_mock_filestore(
{'setcontents': ''}) as Fakefilestore:
self.task.run_export()
if Fakefilestore:
self.assertEqual('setcontents', Fakefilestore[-1]['method'])
self.assertEqual('done', self.filestore_attachment.state)
self.assertEqual(
'/home/user/test/filestore_test_export.txt',
Fakefilestore[-1]['args'][0])
self.assertEqual(
'test filestore file export',
Fakefilestore[-1]['kwargs']['data'])

86
attachment_synchronize/tests/test_ftp.py 100644
View File

@ -0,0 +1,86 @@
# coding: utf-8
# @ 2015 Valentin CHEMIERE @ Akretion
# ©2016 @author Mourad EL HADJ MIMOUNE <mourad.elhadj.mimoune@akretion.com>
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import logging
from base64 import b64decode
import hashlib
from .common import TestConnection, ContextualStringIO
from .mock_server import server_mock_ftp
from .mock_server import MultiResponse
from openerp.exceptions import UserError
_logger = logging.getLogger(__name__)
class TestFtpConnection(TestConnection):
def setUp(self):
super(TestFtpConnection, self).setUp()
self.test_file_ftp = ContextualStringIO()
self.test_file_ftp.write('import ftp')
self.test_file_ftp.seek(0)
def test_00_ftp_import(self):
self.task = self.env.ref('external_file_location.ftp_import_task')
with server_mock_ftp(
{'open': self.test_file_ftp,
'listdir': ['test-import-ftp.txt']}):
self.task.run_import()
search_file = self.env['ir.attachment.metadata'].search(
[('name', '=', 'test-import-ftp.txt')])
self.assertEqual(len(search_file), 1)
self.assertEqual(b64decode(search_file[0].datas), 'import ftp')
def test_01_ftp_export(self):
self.task = self.env.ref('external_file_location.ftp_export_task')
self.ftp_attachment = self.env.ref(
'external_file_location.ir_attachment_export_file_ftp')
with server_mock_ftp(
{'setcontents': ''}) as FakeFTP:
self.task.run_export()
if FakeFTP:
self.assertEqual('setcontents', FakeFTP[-1]['method'])
self.assertEqual('done', self.ftp_attachment.state)
self.assertEqual(
'/home/user/test/ftp_test_export.txt',
FakeFTP[-1]['args'][0])
self.assertEqual(
'test ftp file export',
FakeFTP[-1]['kwargs']['data'])
def test_02_ftp_import_md5(self):
md5_file = ContextualStringIO()
md5_file.write(hashlib.md5('import ftp').hexdigest())
md5_file.seek(0)
task = self.env.ref('external_file_location.ftp_import_task')
task.md5_check = True
with server_mock_ftp(
{'open': MultiResponse({
1: md5_file,
0: self.test_file_ftp}),
'listdir': [task.filename]}) as Fakeftp:
task.run_import()
search_file = self.env['ir.attachment.metadata'].search(
(('name', '=', task.filename),))
self.assertEqual(len(search_file), 1)
self.assertEqual(b64decode(search_file[0].datas),
'import ftp')
self.assertEqual('open', Fakeftp[-1]['method'])
self.assertEqual(hashlib.md5('import ftp').hexdigest(),
search_file.external_hash)
def test_03_ftp_import_md5_corrupt_file(self):
md5_file = ContextualStringIO()
md5_file.write(hashlib.md5('import test ftp corrupted').hexdigest())
md5_file.seek(0)
task = self.env.ref('external_file_location.ftp_import_task')
task.md5_check = True
with server_mock_ftp(
{'open': MultiResponse({
1: md5_file,
0: self.test_file_ftp}),
'listdir': [task.filename]}):
with self.assertRaises(UserError):
task.run_import()

85
attachment_synchronize/tests/test_sftp.py 100644
View File

@ -0,0 +1,85 @@
# coding: utf-8
# @ 2015 Valentin CHEMIERE @ Akretion
# ©2016 @author Mourad EL HADJ MIMOUNE <mourad.elhadj.mimoune@akretion.com>
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import logging
from base64 import b64decode
import hashlib
from .common import TestConnection, ContextualStringIO
from .mock_server import server_mock_sftp
from .mock_server import MultiResponse
from openerp.exceptions import UserError
_logger = logging.getLogger(__name__)
class TestSftpConnection(TestConnection):
def setUp(self):
super(TestSftpConnection, self).setUp()
self.test_file_sftp = ContextualStringIO()
self.test_file_sftp.write('import sftp')
self.test_file_sftp.seek(0)
def test_00_sftp_import(self):
task = self.env.ref('external_file_location.sftp_import_task')
with server_mock_sftp(
{'open': self.test_file_sftp,
'listdir': [task.filename]}):
task.run_import()
search_file = self.env['ir.attachment.metadata'].search(
[('name', '=', task.filename)])
self.assertEqual(len(search_file), 1)
self.assertEqual(b64decode(search_file[0].datas), 'import sftp')
def test_01_sftp_export(self):
self.task = self.env.ref('external_file_location.sftp_export_task')
self.sftp_attachment = self.env.ref(
'external_file_location.ir_attachment_export_file_sftp')
with server_mock_sftp(
{'setcontents': ''}) as FakeSFTP:
self.task.run_export()
if FakeSFTP:
self.assertEqual('setcontents', FakeSFTP[-1]['method'])
self.assertEqual(
'/home/user/test/sftp_test_export.txt',
FakeSFTP[-1]['args'][0])
self.assertEqual(
'test sftp file export',
FakeSFTP[-1]['kwargs']['data'])
def test_02_sftp_import_md5(self):
md5_file = ContextualStringIO()
md5_file.write(hashlib.md5('import sftp').hexdigest())
md5_file.seek(0)
task = self.env.ref('external_file_location.sftp_import_task')
task.md5_check = True
with server_mock_sftp(
{'open': MultiResponse({
1: md5_file,
0: self.test_file_sftp}),
'listdir': [task.filename]}) as FakeSFTP:
task.run_import()
search_file = self.env['ir.attachment.metadata'].search(
(('name', '=', task.filename),))
self.assertEqual(len(search_file), 1)
self.assertEqual(b64decode(search_file[0].datas),
'import sftp')
self.assertEqual('open', FakeSFTP[-1]['method'])
self.assertEqual(hashlib.md5('import sftp').hexdigest(),
search_file.external_hash)
def test_03_sftp_import_md5_corrupt_file(self):
md5_file = ContextualStringIO()
md5_file.write(hashlib.md5('import test sftp corrupted').hexdigest())
md5_file.seek(0)
task = self.env.ref('external_file_location.sftp_import_task')
task.md5_check = True
with server_mock_sftp(
{'open': MultiResponse({
1: md5_file,
0: self.test_file_sftp}),
'listdir': [task.filename]}):
with self.assertRaises(UserError):
task.run_import()

26
attachment_synchronize/views/attachment_view.xml 100644
View File

@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<odoo>
<record id="view_attachment_improved_form" model="ir.ui.view">
<field name="model">ir.attachment.metadata</field>
<field name="inherit_id" ref="base_attachment_queue.view_attachment_improved_form" />
<field name="arch" type="xml">
<field name="url" position="after">
<field name="task_id" attrs="{'required': [('file_type', '=', 'export')]}"/>
<field name="storage_backend_id"/>
</field>
</field>
</record>
<record id="view_external_attachment_tree" model="ir.ui.view">
<field name="model">ir.attachment.metadata</field>
<field name="inherit_id" ref="base_attachment_queue.view_external_attachment_tree" />
<field name="arch" type="xml">
<field name="file_type" position="after">
<field name="task_id"/>
<field name="storage_backend_id"/>
</field>
</field>
</record>
</odoo>

18
attachment_synchronize/views/storage_backend_view.xml 100644
View File

@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<odoo>
<record id="view_storage_backend_form" model="ir.ui.view">
<field name="model">storage.backend</field>
<field name="inherit_id" ref="storage_backend.storage_backend_view_form" />
<field name="priority" eval="250"/>
<field name="arch" type="xml">
<form>
<xpath expr="/form/sheet" position="inside">
<separator string="Tasks"/>
<field name="task_ids" nolabel="1"/>
</xpath>
</form>
</field>
</record>
</odoo>

59
attachment_synchronize/views/task_view.xml 100644
View File

@ -0,0 +1,59 @@
<?xml version="1.0" encoding="UTF-8"?>
<odoo>
<record id="view_task_form" model="ir.ui.view">
<field name="model">storage.backend.task</field>
<field name="arch" type="xml">
<form>
<header>
<button name="run_import" type="object" string="Run" icon="gtk-execute" attrs="{'invisible': [('method_type', '!=', 'import')]}"/>
</header>
<sheet>
<field name="method_type" invisible="1"/>
<group col="4">
<div class="oe_title" style="width: 390px;" colspan="4">
<label class="oe_edit_only" for="name" string="Name"/>
<h1><field name="name" class="oe_inline"/></h1>
</div>
<field name="method_type" colspan="2"/>
<span colspan="2"/>
<field name="filename" colspan="4" attrs="{'invisible':[('method_type','!=','import')]}"/>
<field name="filepath" colspan="4" />
</group>
<group col="6">
<field name="after_import" attrs="{'invisible':[('method_type','!=','import')]}"/>
<group col="4" colspan="4" >
<field name="move_path" colspan="4"
attrs="{'invisible':['|', '&amp;',
('after_import','!=','move'),
('after_import','!=','move_rename'),
('method_type','!=','import')]}"/>
<field name="new_name" colspan="4"
attrs="{'invisible': ['|', '&amp;',
('after_import','!=','rename'),
('after_import','!=','move_rename'),
('method_type','!=','import')]}"/>
</group>
<field name="active" colspan="2"/>
</group>
<group string="Data importation setting">
<field name="file_type" attrs="{'invisible':[('method_type','!=','import')]}"/>
</group>
</sheet>
</form>
</field>
</record>
<record id="view_task_tree" model="ir.ui.view">
<field name="model">storage.backend.task</field>
<field name="arch" type="xml">
<tree string="Tasks" >
<field name="name" select="1"/>
<field name="method_type"/>
<field name="filename"/>
<field name="filepath"/>
</tree>
</field>
</record>
</odoo>