Browse Source

[REF] refactor code and add test

12.0-mig-module_prototyper_last
Sébastien BEAU 5 years ago
committed by David Beal
parent
commit
78179942d5
  1. 1
      attachment_synchronize/__manifest__.py
  2. 8
      attachment_synchronize/data/cron.xml
  3. 20
      attachment_synchronize/demo/attachment_synchronize_task_demo.xml
  4. 2
      attachment_synchronize/models/attachment.py
  5. 4
      attachment_synchronize/models/storage_backend.py
  6. 23
      attachment_synchronize/models/task.py
  7. 2
      attachment_synchronize/security/ir.model.access.csv
  8. 2
      attachment_synchronize/tests/__init__.py
  9. 38
      attachment_synchronize/tests/common.py
  10. 35
      attachment_synchronize/tests/test_export.py
  11. 80
      attachment_synchronize/tests/test_import.py
  12. 2
      attachment_synchronize/views/storage_backend_view.xml
  13. 4
      attachment_synchronize/views/task_view.xml

1
attachment_synchronize/__manifest__.py

@ -22,6 +22,7 @@
'security/ir.model.access.csv',
],
'demo': [
"demo/attachment_synchronize_task_demo.xml",
],
'installable': True,
'application': False,

8
attachment_synchronize/data/cron.xml

@ -1,16 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<odoo noupdate="1">
<record model="ir.cron" id="cronjob_run_exchange_tasks">
<record model="ir.cron" id="cronjob_run_attachment_synchronize_task_import">
<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="model_id" ref="model_attachment_synchronize_task"/>
<field name="state">code</field>
<field name="code">model.run_task_scheduler([('method_type', '=', 'import')])</field>
<field name="code">model.run_task_import_scheduler()</field>
</record>
</odoo>

20
attachment_synchronize/demo/attachment_synchronize_task_demo.xml

@ -0,0 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
<odoo>
<record id="import_from_filestore" model="attachment.synchronize.task">
<field name="name">TEST Import</field>
<field name="backend_id" ref="storage_backend.default_storage_backend"/>
<field name="method_type">import</field>
<field name="after_import">delete</field>
<field name="filepath">test_import</field>
<field name="emails">foo@example.org,bar@example.org</field>
</record>
<record id="export_to_filestore" model="attachment.synchronize.task">
<field name="name">TEST Export</field>
<field name="backend_id" ref="storage_backend.default_storage_backend"/>
<field name="method_type">export</field>
<field name="filepath">test_export</field>
<field name="emails">foo@example.org,bar@example.org</field>
</record>
</odoo>

2
attachment_synchronize/models/attachment.py

@ -8,7 +8,7 @@ import os
class AttachmentQueue(models.Model):
_inherit = 'attachment.queue'
task_id = fields.Many2one('storage.backend.task', string='Task')
task_id = fields.Many2one('attachment.synchronize.task', string='Task')
storage_backend_id = fields.Many2one(
'storage.backend', string='Storage Backend',
related='task_id.backend_id', store=True)

4
attachment_synchronize/models/storage_backend.py

@ -6,6 +6,6 @@ from odoo import fields, models
class StorageBackend(models.Model):
_inherit = "storage.backend"
task_ids = fields.One2many(
"storage.backend.task", "backend_id",
synchronize_task_ids = fields.One2many(
"attachment.synchronize.task", "backend_id",
string="Tasks")

23
attachment_synchronize/models/task.py

@ -6,6 +6,7 @@ import os
import odoo
from odoo import api, fields, models, tools
from odoo.osv import expression
_logger = logging.getLogger(__name__)
@ -41,9 +42,9 @@ except ImportError:
_logger.warning("jinja2 not available, templating features will not work!")
class StorageBackendTask(models.Model):
_name = 'storage.backend.task'
_description = 'Storage Backend task'
class AttachmentSynchronizeTask(models.Model):
_name = 'attachment.synchronize.task'
_description = 'Attachment synchronize task'
name = fields.Char(required=True)
method_type = fields.Selection(
@ -118,13 +119,13 @@ class StorageBackendTask(models.Model):
return render_result
@api.model
def run_task_scheduler(self, domain=None):
def run_task_import_scheduler(self, domain=None):
if domain is None:
domain = []
domain = expression.AND(domain, [
domain = expression.AND([domain, [
('method_type', '=', 'import'),
('enabled', '=', True),
])
]])
for task in self.search(domain):
task.run_import()
@ -133,8 +134,8 @@ class StorageBackendTask(models.Model):
self.ensure_one()
attach_obj = self.env['attachment.queue']
backend = self.backend_id
filenames = backend._list(
relative_path=self.filepath, pattern=self.pattern)
filepath = self.filepath or ""
filenames = backend._list(relative_path=filepath, pattern=self.pattern)
if self.check_duplicated_files:
filenames = self._file_to_import(filenames)
total_import = 0
@ -145,8 +146,7 @@ class StorageBackendTask(models.Model):
new_env = api.Environment(new_cr, self.env.uid,
self.env.context)
try:
full_absolute_path = os.path.join(
self.filepath, file_name)
full_absolute_path = os.path.join(filepath, file_name)
datas = backend._get_b64_data(full_absolute_path)
attach_vals = self._prepare_attachment_vals(
datas, file_name)
@ -156,8 +156,7 @@ class StorageBackendTask(models.Model):
if self.after_import == 'rename':
new_name = self._template_render(
self.new_name, attachment)
new_full_path = os.path.join(
self.filepath, new_name)
new_full_path = os.path.join(filepath, new_name)
elif self.after_import == 'move':
new_full_path = os.path.join(
self.move_path, file_name)

2
attachment_synchronize/security/ir.model.access.csv

@ -1,2 +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
access_attachment_synchronize_task_manager,attachment.synchronize.task.manager,model_attachment_synchronize_task,base.group_system,1,1,1,1

2
attachment_synchronize/tests/__init__.py

@ -0,0 +1,2 @@
from . import test_import
from . import test_export

38
attachment_synchronize/tests/common.py

@ -0,0 +1,38 @@
# Copyright 2020 Akretion (http://www.akretion.com).
# @author Sébastien BEAU <sebastien.beau@akretion.com>
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import mock
import os
from odoo.addons.storage_backend.tests.common import Common
class SyncCommon(Common):
def _clean_testing_directory(self):
for test_dir in [
self.directory_input, self.directory_output, self.directory_archived]:
for filename in self.backend._list(test_dir):
self.backend._delete(os.path.join(test_dir, filename))
def _create_test_file(self):
self.backend._add_b64_data(
os.path.join(self.directory_input, "bar.txt"),
self.filedata,
mimetype=u"text/plain")
def setUp(self):
super().setUp()
self.env.cr.commit = mock.Mock()
self.registry.enter_test_mode(self.env.cr)
self.directory_input = "test_import"
self.directory_output = "test_output"
self.directory_archived = "test_archived"
self._clean_testing_directory()
self._create_test_file()
self.task = self.env.ref("attachment_synchronize.import_from_filestore")
def tearDown(self):
self.registry.leave_test_mode()
self._clean_testing_directory()
super().tearDown()

35
attachment_synchronize/tests/test_export.py

@ -0,0 +1,35 @@
# Copyright 2020 Akretion (http://www.akretion.com).
# @author Sébastien BEAU <sebastien.beau@akretion.com>
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import os
import mock
from .common import SyncCommon
def raising_side_effect(*args, **kwargs):
raise Exception("Boom")
class TestExport(SyncCommon):
def setUp(self):
super().setUp()
self.task = self.env.ref("attachment_synchronize.export_to_filestore")
self.attachment = self.env["attachment.queue"].create({
"name": "foo.txt",
"datas_fname": "foo.txt",
"task_id": self.task.id,
"file_type": "export",
"datas": self.filedata,
})
def test_export(self):
self.attachment.run()
result = self.backend._list("test_export")
self.assertEqual(result, ["foo.txt"])
def test_failing_export(self):
with mock.patch.object(type(self.backend), "_add_b64_data", side_effect=raising_side_effect):
self.attachment.run()
self.assertEqual(self.attachment.state, "failed")
self.assertEqual(self.attachment.state_message, "Boom")

80
attachment_synchronize/tests/test_import.py

@ -0,0 +1,80 @@
# Copyright 2020 Akretion (http://www.akretion.com).
# @author Sébastien BEAU <sebastien.beau@akretion.com>
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from .common import SyncCommon
class TestImport(SyncCommon):
@property
def archived_files(self):
return self.backend._list(self.directory_archived)
@property
def input_files(self):
return self.backend._list(self.directory_input)
def _check_attachment_created(self, count=1):
attachment = self.env["attachment.queue"].search([("name", "=", "bar.txt")])
self.assertEqual(len(attachment), count)
def test_import_with_rename(self):
self.task.write({"after_import": "rename", "new_name": "foo.txt"})
self.task.run_import()
self._check_attachment_created()
self.assertEqual(self.input_files, ["foo.txt"])
self.assertEqual(self.archived_files, [])
def test_import_with_move(self):
self.task.write({"after_import": "move", "move_path": self.directory_archived})
self.task.run_import()
self._check_attachment_created()
self.assertEqual(self.input_files, [])
self.assertEqual(self.archived_files, ["bar.txt"])
def test_import_with_move_and_rename(self):
self.task.write({
"after_import": "move_rename",
"new_name": "foo.txt",
"move_path": self.directory_archived,
})
self.task.run_import()
self._check_attachment_created()
self.assertEqual(self.input_files, [])
self.assertEqual(self.archived_files, ["foo.txt"])
def test_import_with_delete(self):
self.task.write({"after_import": "delete"})
self.task.run_import()
self._check_attachment_created()
self.assertEqual(self.input_files, [])
self.assertEqual(self.archived_files, [])
def test_import_twice(self):
self.task.write({"after_import": "delete"})
self.task.run_import()
self._check_attachment_created(count=1)
self._create_test_file()
self.task.run_import()
self._check_attachment_created(count=2)
def test_import_twice_no_duplicate(self):
self.task.write({"after_import": "delete", "check_duplicated_files": True})
self.task.run_import()
self._check_attachment_created(count=1)
self._create_test_file()
self.task.run_import()
self._check_attachment_created(count=1)
def test_running_cron(self):
self.task.write({"after_import": "delete"})
self.env["attachment.synchronize.task"].run_task_import_scheduler()
self._check_attachment_created(count=1)
def test_running_cron_disable_task(self):
self.task.write({"after_import": "delete", "enabled": False})
self.env["attachment.synchronize.task"].run_task_import_scheduler()
self._check_attachment_created(count=0)

2
attachment_synchronize/views/storage_backend_view.xml

@ -9,7 +9,7 @@
<xpath expr="//group[@name='config']" position="after">
<notebook>
<page string="Tasks">
<field name="task_ids"/>
<field name="synchronize_task_ids"/>
</page>
</notebook>
</xpath>

4
attachment_synchronize/views/task_view.xml

@ -1,7 +1,7 @@
<?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="model">attachment.synchronize.task</field>
<field name="arch" type="xml">
<form>
<header>
@ -37,7 +37,7 @@
</record>
<record id="view_task_tree" model="ir.ui.view">
<field name="model">storage.backend.task</field>
<field name="model">attachment.synchronize.task</field>
<field name="arch" type="xml">
<tree string="Tasks">
<field name="name" select="1"/>

Loading…
Cancel
Save