diff --git a/edi_queue_oca/models/__init__.py b/edi_queue_oca/models/__init__.py index e1a8e889b..e6c8d6714 100644 --- a/edi_queue_oca/models/__init__.py +++ b/edi_queue_oca/models/__init__.py @@ -1,3 +1,4 @@ from . import edi_exchange_record from . import edi_exchange_type from . import edi_backend +from . import queue_job diff --git a/edi_queue_oca/models/edi_exchange_record.py b/edi_queue_oca/models/edi_exchange_record.py index c8ddeff7d..aa2769a7e 100644 --- a/edi_queue_oca/models/edi_exchange_record.py +++ b/edi_queue_oca/models/edi_exchange_record.py @@ -66,6 +66,33 @@ def delayable(self, **kw): def _job_retry_params(self): return {} + def _mark_failed_from_queue_job(self, job): + """Set the EDI error matching a terminal queue job failure. + + :param job: failed ``queue.job`` record + """ + failure_mapping = { + "action_exchange_process": ("input_processed_error", "process_ko"), + "action_exchange_receive": ("input_receive_error", "receive_ko"), + "action_exchange_send": ("output_error_on_send", "send_ko"), + } + failure = failure_mapping.get(job.method_name) + if not failure: + return + state, message_key = failure + for record in self: + state_changed = record.edi_exchange_state != state + record.write( + { + "edi_exchange_state": state, + "exchange_error": job.exc_message, + "exchange_error_traceback": job.exc_info, + "exchanged_on": fields.Datetime.now(), + } + ) + if state_changed: + record._notify_error(message_key) + def _compute_related_queue_jobs_count(self): for rec in self: # TODO: We should refactor the object field on queue_job to use jsonb field diff --git a/edi_queue_oca/models/queue_job.py b/edi_queue_oca/models/queue_job.py new file mode 100644 index 000000000..d8bfe8da9 --- /dev/null +++ b/edi_queue_oca/models/queue_job.py @@ -0,0 +1,28 @@ +# Copyright 2026 Camptocamp SA +# License LGPL-3.0 or later (https://www.gnu.org/licenses/lgpl). + +from odoo import models + + +class QueueJob(models.Model): + _inherit = "queue.job" + + def write(self, vals): + result = super().write(vals) + if vals.get("state") == "failed": + self._mark_related_edi_exchanges_failed() + return result + + def _mark_related_edi_exchanges_failed(self): + """Propagate terminal EDI job failures to their exchange records.""" + supported_methods = { + "action_exchange_process", + "action_exchange_receive", + "action_exchange_send", + } + jobs = self.filtered( + lambda job: job.model_name == "edi.exchange.record" + and job.method_name in supported_methods + ) + for job in jobs: + job.records.sudo()._mark_failed_from_queue_job(job) diff --git a/edi_queue_oca/tests/test_backend_jobs.py b/edi_queue_oca/tests/test_backend_jobs.py index 61ac106bd..f8b149a6e 100644 --- a/edi_queue_oca/tests/test_backend_jobs.py +++ b/edi_queue_oca/tests/test_backend_jobs.py @@ -94,6 +94,7 @@ def test_output(self): self.assertEqual(created, self._get_related_jobs(record)) def test_output_fail_retry(self): + """Test a retryable send failure keeps the exchange pending.""" job_counter = self.job_counter() vals = { "model": self.partner._name, @@ -106,8 +107,113 @@ def test_output_fail_retry(self): job_counter.search_created() with mock.patch.object(type(self.backend), "_exchange_send") as mocked: mocked.side_effect = ReqConnectionError("Connection broken") - with self.assertRaises(RetryableJobError): + with self.assertRaisesRegex(RetryableJobError, "Connection broken"): job.perform() + self.assertEqual(record.edi_exchange_state, "output_pending") + + def test_failed_send_job_marks_exchange_as_error(self): + """Test a terminal send job failure marks its exchange as failed.""" + record = self.backend.create_record( + "test_csv_output", + { + "model": self.partner._name, + "res_id": self.partner.id, + "edi_exchange_state": "output_pending", + }, + ) + record._set_file_content("ABC") + job = record.with_delay().action_exchange_send() + + job.db_record().write( + { + "state": "failed", + "exc_message": "Connection broken", + "exc_info": "Traceback of the connection failure", + } + ) + + self.assertEqual(record.edi_exchange_state, "output_error_on_send") + self.assertEqual(record.exchange_error, "Connection broken") + self.assertEqual( + record.exchange_error_traceback, "Traceback of the connection failure" + ) + self.assertTrue(record.exchanged_on) + + def test_failed_receive_job_marks_exchange_as_error(self): + """Test a terminal receive job failure marks its exchange as failed.""" + record = self.backend.create_record( + "test_csv_input", + { + "model": self.partner._name, + "res_id": self.partner.id, + "edi_exchange_state": "input_pending", + }, + ) + job = record.with_delay().action_exchange_receive() + + job.db_record().write( + { + "state": "failed", + "exc_message": "Receive failed", + "exc_info": "Traceback for receive", + } + ) + + self.assertEqual(record.edi_exchange_state, "input_receive_error") + self.assertEqual(record.exchange_error, "Receive failed") + self.assertEqual(record.exchange_error_traceback, "Traceback for receive") + + def test_failed_process_job_marks_exchange_as_error(self): + """Test a terminal process job failure marks its exchange as failed.""" + record = self.backend.create_record( + "test_csv_input", + { + "model": self.partner._name, + "res_id": self.partner.id, + "edi_exchange_state": "input_received", + }, + ) + job = record.with_delay().action_exchange_process() + + job.db_record().write( + { + "state": "failed", + "exc_message": "Process failed", + "exc_info": "Traceback for process", + } + ) + + self.assertEqual(record.edi_exchange_state, "input_processed_error") + self.assertEqual(record.exchange_error, "Process failed") + self.assertEqual(record.exchange_error_traceback, "Traceback for process") + + def test_unsupported_failed_jobs_do_not_mark_exchange_as_error(self): + """Test generate and non-exchange jobs do not alter the exchange state.""" + record = self.backend.create_record( + "test_csv_output", + { + "model": self.partner._name, + "res_id": self.partner.id, + "edi_exchange_state": "output_pending", + }, + ) + jobs = ( + record.with_delay().action_exchange_generate(), + self.backend.with_delay().exchange_send(record), + ) + + for job in jobs: + job.db_record().write( + { + "state": "failed", + "exc_message": "Unsupported job failed", + "exc_info": "Unsupported job traceback", + } + ) + + self.assertEqual(record.edi_exchange_state, "output_pending") + self.assertFalse(record.exchange_error) + self.assertFalse(record.exchange_error_traceback) def test_input(self): job_counter = self.job_counter()