[14.0] edi: generate/send chain jobs + add identity exact match based on checksum

edi_oca/models/edi_backend.py

@@ -298,18 +298,23 @@ def exchange_send(self, exchange_record):
         res = ""
         try:
             self._exchange_send(exchange_record)
+            _logger.debug("%s sent", exchange_record.identifier)
+        except self._send_retryable_exceptions() as err:
+            error = _get_exception_msg()
+            _logger.debug("%s send failed. To be retried.", exchange_record.identifier)
+            raise RetryableJobError(
+                error, **exchange_record._job_retry_params()
+            ) from err
         except self._swallable_exceptions():
             if self.env.context.get("_edi_send_break_on_error"):
                 raise
             error = _get_exception_msg()
             state = "output_error_on_send"
             message = exchange_record._exchange_status_message("send_ko")
             res = f"Error: {error}"
-        except self._send_retryable_exceptions() as err:
-            error = _get_exception_msg()
-            raise RetryableJobError(
-                error, **exchange_record._job_retry_params()
-            ) from err
+            _logger.debug(
+                "%s send failed. Marked as errored.", exchange_record.identifier
+            )
         else:
             # TODO: maybe the send handler should return desired message and state
             message = exchange_record._exchange_status_message("send_ok")
@@ -344,7 +349,9 @@ def _swallable_exceptions(self):
 
     def _send_retryable_exceptions(self):
         # IOError is a base class for all connection errors
-        return (IOError,)
+        # OSError is a base class for all errors
+        # when dealing w/ internal or external systems or filesystems
+        return (IOError, OSError)
 
     def _output_check_send(self, exchange_record):
         if exchange_record.direction != "output":
@@ -370,7 +377,6 @@ def _cron_check_output_exchange_sync(self, **kw):
         for backend in self:
             backend._check_output_exchange_sync(**kw)
 
-    # TODO: consider splitting cron in 2 (1 for receiving, 1 for processing)
     def _check_output_exchange_sync(
         self, skip_send=False, skip_sent=True, record_ids=None
     ):
@@ -391,7 +397,12 @@ def _check_output_exchange_sync(
             len(new_records),
         )
         for rec in new_records:
-            rec.with_delay().action_exchange_generate()
+            job1 = rec.delayable().action_exchange_generate()
+            if not skip_send:
+                # Chain send job.
+                # Raise prio to max to send the record out as fast as possible.
+                job1.on_done(rec.delayable(priority=0).action_exchange_send())
+            job1.delay()
 
         if skip_send:
             return

edi_oca/models/edi_exchange_record.py

@@ -9,6 +9,8 @@
 
 from odoo import _, api, exceptions, fields, models
 
+from ..utils import exchange_record_job_identity_exact, get_checksum
+
 _logger = logging.getLogger(__name__)
 
 
@@ -50,6 +52,9 @@ class EDIExchangeRecord(models.Model):
     exchange_filename = fields.Char(
         compute="_compute_exchange_filename", readonly=False, store=True
     )
+    exchange_filechecksum = fields.Char(
+        compute="_compute_exchange_filechecksum", store=True
+    )
     exchanged_on = fields.Datetime(
         string="Exchanged on",
         help="Sent or received on this date.",
@@ -133,6 +138,14 @@ def _compute_exchange_filename(self):
             if not rec.exchange_filename:
                 rec.exchange_filename = rec.type_id._make_exchange_filename(rec)
 
+    @api.depends("exchange_file")
+    def _compute_exchange_filechecksum(self):
+        for rec in self:
+            content = rec.exchange_file or ""
+            if not isinstance(content, bytes):
+                content = content.encode()
+            rec.exchange_filechecksum = get_checksum(content)
+
     @api.depends("edi_exchange_state")
     def _compute_exchanged_on(self):
         for rec in self:
@@ -566,6 +579,8 @@ def _job_delay_params(self):
         channel = self.type_id.sudo().job_channel_id
         if channel:
             params["channel"] = channel.complete_name
+        # Avoid generating the same job for the same record if existing
+        params["identity_key"] = exchange_record_job_identity_exact
         return params
 
     def with_delay(self, **kw):

edi_oca/tests/test_backend_output.py

@@ -1,4 +1,5 @@
 # Copyright 2020 ACSONE
+# Copyright 2021 Camptocamp
 # @author: Simone Orsi <[email protected]>
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
 
@@ -8,6 +9,8 @@
 from odoo import fields, tools
 from odoo.exceptions import UserError
 
+from odoo.addons.queue_job.tests.common import trap_jobs
+
 from .common import EDIBackendCommonComponentRegistryTestCase
 from .fake_components import FakeOutputChecker, FakeOutputGenerator, FakeOutputSender
 
@@ -17,7 +20,6 @@ class EDIBackendTestOutputCase(EDIBackendCommonComponentRegistryTestCase):
     def setUpClass(cls):
         super().setUpClass()
         cls._build_components(
-            # TODO: test all components lookup
             cls,
             FakeOutputGenerator,
             FakeOutputSender,
@@ -36,14 +38,14 @@ def setUp(self):
         FakeOutputChecker.reset_faked()
 
     def test_generate_record_output(self):
-        self.backend.with_context(fake_output="yeah!").exchange_generate(self.record)
+        self.record.with_context(fake_output="yeah!").action_exchange_generate()
         self.assertEqual(self.record._get_file_content(), "yeah!")
 
     def test_generate_record_output_pdf(self):
-        result = tools.file_open(
+        pdf_content = tools.file_open(
             "result.pdf", subdir="addons/edi_oca/tests", mode="rb"
         ).read()
-        self.backend.with_context(fake_output=result).exchange_generate(self.record)
+        self.record.with_context(fake_output=pdf_content).action_exchange_generate()
 
     def test_send_record(self):
         self.record.write({"edi_exchange_state": "output_pending"})
@@ -105,3 +107,53 @@ def test_send_not_generated_record(self):
                 err.exception.args[0], "Record ID=%d has no file to send!" % record.id
             )
             mocked.assert_not_called()
+
+
+class EDIBackendTestOutputJobsCase(EDIBackendCommonComponentRegistryTestCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls._build_components(
+            cls,
+            FakeOutputGenerator,
+            FakeOutputSender,
+            FakeOutputChecker,
+        )
+        vals = {
+            "model": cls.partner._name,
+            "res_id": cls.partner.id,
+        }
+        cls.record = cls.backend.create_record("test_csv_output", vals)
+        cls.record.type_id.exchange_file_auto_generate = True
+
+    @classmethod
+    def _setup_context(cls):
+        # Re-enable jobs
+        return dict(super()._setup_context(), test_queue_job_no_delay=False)
+
+    def test_job(self):
+        with trap_jobs() as trap:
+            self.backend._check_output_exchange_sync(record_ids=self.record.ids)
+            trap.assert_jobs_count(2)
+            trap.assert_enqueued_job(
+                self.record.action_exchange_generate,
+            )
+            trap.assert_enqueued_job(
+                self.record.action_exchange_send, properties=dict(priority=0)
+            )
+            # No matter how many times we schedule jobs
+            self.record.with_delay().action_exchange_generate()
+            self.record.with_delay().action_exchange_generate()
+            self.record.with_delay().action_exchange_generate()
+            # identity key should prevent having new jobs for same record same file
+            trap.assert_jobs_count(2)
+            # but if we change the content
+            self.record._set_file_content("something different")
+            # 1st call will schedule another job
+            self.record.with_delay().action_exchange_generate()
+            # the 2nd one not
+            self.record.with_delay().action_exchange_generate()
+            trap.assert_jobs_count(3)
+            self.record.with_delay().action_exchange_send()
+            trap.assert_jobs_count(4)
+        # TODO: test input in the same way

edi_oca/tests/test_record.py

@@ -3,12 +3,15 @@
 # @author: Simone Orsi <[email protected]>
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
 
+import base64
+
 import mock
 from freezegun import freeze_time
 
 from odoo import exceptions, fields
 from odoo.tools import mute_logger
 
+from odoo.addons.edi_oca.utils import get_checksum
 from odoo.addons.queue_job.delay import DelayableRecordset
 
 from .common import EDIBackendCommonTestCase
@@ -216,3 +219,19 @@ def test_retry(self):
             mocked.assert_not_called()
         self.assertEqual(record0.edi_exchange_state, "output_pending")
         self.assertFalse(record0.retryable)
+
+    def test_checksum(self):
+        filecontent = base64.b64encode(b"ABC")
+        checksum1 = get_checksum(filecontent)
+        vals = {
+            "model": self.partner._name,
+            "res_id": self.partner.id,
+            "exchange_file": filecontent,
+        }
+        record0 = self.backend.create_record("test_csv_output", vals)
+        self.assertEqual(record0.exchange_filechecksum, checksum1)
+        filecontent = base64.b64encode(b"DEF")
+        checksum2 = get_checksum(filecontent)
+        record0.exchange_file = filecontent
+        self.assertEqual(record0.exchange_filechecksum, checksum2)
+        self.assertNotEqual(record0.exchange_filechecksum, checksum1)

edi_oca/utils.py

@@ -2,9 +2,25 @@
 # @author Simone Orsi <[email protected]>
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
 
+import hashlib
+
 from odoo.addons.http_routing.models.ir_http import slugify
+from odoo.addons.queue_job.job import identity_exact_hasher
 
 
 def normalize_string(a_string, sep="_"):
     """Normalize given string, replace dashes with given separator."""
     return slugify(a_string).replace("-", sep)
+
+
+def get_checksum(filecontent):
+    return hashlib.md5(filecontent).hexdigest()
+
+
+def exchange_record_job_identity_exact(job_):
+    hasher = identity_exact_hasher(job_)
+    # Include files checksum
+    hasher.update(
+        str(sorted(job_.recordset.mapped("exchange_filechecksum"))).encode("utf-8")
+    )
+    return hasher.hexdigest()

edi_oca/views/edi_exchange_record_views.xml

@@ -105,6 +105,10 @@
                                 name="exchange_filename"
                                 attrs="{'invisible': [('exchange_file', '!=', False)]}"
                             />
+                            <field
+                                name="exchange_filechecksum"
+                                attrs="{'invisible': [('exchange_file', '!=', False)]}"
+                            />
                         </group>
                         <group
                             name="ack"