Sébastien BEAU
4 years ago
committed by
David Beal
David Beal
11 changed files with 20 additions and 445 deletions
-
3attachment_synchronize/__manifest__.py
-
95attachment_synchronize/demo/task_demo.xml
-
4attachment_synchronize/models/attachment.py
-
20attachment_synchronize/models/task.py
-
4attachment_synchronize/tests/__init__.py
-
32attachment_synchronize/tests/common.py
-
74attachment_synchronize/tests/mock_server.py
-
50attachment_synchronize/tests/test_filestore.py
-
86attachment_synchronize/tests/test_ftp.py
-
85attachment_synchronize/tests/test_sftp.py
-
12attachment_synchronize/views/attachment_view.xml
@ -1,95 +0,0 @@ |
|||
<?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> |
|||
@ -1,4 +0,0 @@ |
|||
from . import mock_server |
|||
from . import test_ftp |
|||
from . import test_sftp |
|||
from . import test_filestore |
|||
@ -1,32 +0,0 @@ |
|||
# 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() |
|||
@ -1,74 +0,0 @@ |
|||
# 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 |
|||
@ -1,50 +0,0 @@ |
|||
# 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']) |
|||
@ -1,86 +0,0 @@ |
|||
# 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() |
|||
@ -1,85 +0,0 @@ |
|||
# 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() |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue