Maintainers
+Maintainers
This module is maintained by the OCA.
diff --git a/edi_core_oca/tests/test_backend_input.py b/edi_core_oca/tests/test_backend_input.py
index 89d7fe13f..0ca5e2e5b 100644
--- a/edi_core_oca/tests/test_backend_input.py
+++ b/edi_core_oca/tests/test_backend_input.py
@@ -3,6 +3,7 @@
# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
from odoo_test_helper import FakeModelLoader
+from psycopg2 import OperationalError
from .common import EDIBackendCommonTestCase
@@ -78,3 +79,12 @@ def test_receive_allow_empty_file_record(self):
# Check the record
self.assertEqual(self.record._get_file_content(), "")
self.assertRecordValues(self.record, [{"edi_exchange_state": "input_received"}])
+
+ def test_receive_record_with_operational_error(self):
+ self.record.edi_exchange_state = "input_pending"
+ with self.assertRaises(OperationalError):
+ self.backend.with_context(
+ test_break_receive=OperationalError("SQL error")
+ ).exchange_receive(self.record)
+ self.assertRecordValues(self.record, [{"edi_exchange_state": "input_pending"}])
+ self.assertFalse(self.record.exchange_error)
diff --git a/edi_core_oca/tests/test_backend_output.py b/edi_core_oca/tests/test_backend_output.py
index 69bd07fe8..2d5e49810 100644
--- a/edi_core_oca/tests/test_backend_output.py
+++ b/edi_core_oca/tests/test_backend_output.py
@@ -7,6 +7,7 @@
from freezegun import freeze_time
from odoo_test_helper import FakeModelLoader
+from psycopg2 import OperationalError
from odoo import fields, tools
from odoo.exceptions import UserError
@@ -122,3 +123,13 @@ 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()
+
+ def test_send_record_with_operational_error(self):
+ self.record.write({"edi_exchange_state": "output_pending"})
+ self.record._set_file_content("TEST %d" % self.record.id)
+ with self.assertRaises(OperationalError):
+ self.backend.with_context(
+ test_break_send=OperationalError("SQL error")
+ ).exchange_send(self.record)
+ self.assertRecordValues(self.record, [{"edi_exchange_state": "output_pending"}])
+ self.assertFalse(self.record.exchange_error)
diff --git a/edi_core_oca/tests/test_backend_process.py b/edi_core_oca/tests/test_backend_process.py
index 913abfb6d..de259115e 100644
--- a/edi_core_oca/tests/test_backend_process.py
+++ b/edi_core_oca/tests/test_backend_process.py
@@ -6,6 +6,7 @@
from freezegun import freeze_time
from odoo_test_helper import FakeModelLoader
+from psycopg2 import IntegrityError
from odoo import fields
from odoo.exceptions import UserError
@@ -103,4 +104,13 @@ def test_process_outbound_record(self):
with self.assertRaises(UserError):
record.action_exchange_process()
+ def test_process_record_with_integrity_error(self):
+ self.record.write({"edi_exchange_state": "input_received"})
+ with self.assertRaises(IntegrityError):
+ self.backend.with_context(
+ test_break_process=IntegrityError("SQL error")
+ ).exchange_process(self.record)
+ self.assertRecordValues(self.record, [{"edi_exchange_state": "input_received"}])
+ self.assertFalse(self.record.exchange_error)
+
# TODO: test ack file are processed
diff --git a/edi_core_oca/tests/test_edi_configuration.py b/edi_core_oca/tests/test_edi_configuration.py
index 689a8f6c1..d624b697e 100644
--- a/edi_core_oca/tests/test_edi_configuration.py
+++ b/edi_core_oca/tests/test_edi_configuration.py
@@ -152,3 +152,256 @@ def test_edi_code_snippet(self):
)
# Check the new vals after execution
self.assertEqual(vals, expected_value)
+
+
+class TestEDIConfigurationGlobalEvents(EDIBackendCommonTestCase):
+ """Test the global event dispatch via edi.configuration.
+
+ `EDIExchangeRecord._trigger_edi_event` looks up all `edi.configuration`
+ records flagged as `is_global` and matching the event trigger code,
+ then executes their `snippet_do` against the target record.
+ These tests verify the dispatch happens for all `notify_*` events
+ and that the proper target (exchange record vs related record)
+ is passed to the snippet.
+ """
+
+ # Snippet appends a marker per call so we can verify multiple invocations
+ # against different targets within the same transaction.
+ _marker_snippet = (
+ "conf.write({'description': (conf.description or '') + '|' + record._name})"
+ )
+
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ vals = {
+ "model": cls.partner._name,
+ "res_id": cls.partner.id,
+ }
+ cls.record = cls.backend.create_record("test_csv_output", vals)
+ cls.trigger_model = cls.env["edi.configuration.trigger"]
+ cls.conf_model = cls.env["edi.configuration"]
+ # Reuse existing data triggers when available, create the missing ones.
+ cls.trigger_done = cls.env.ref("edi_core_oca.edi_config_trigger_record_done")
+ cls.trigger_error = cls.env.ref("edi_core_oca.edi_config_trigger_record_error")
+ cls.trigger_ack_received = cls._get_or_create_trigger(
+ "on_edi_exchange_done_ack_received", "On ACK received"
+ )
+ cls.trigger_ack_missing = cls._get_or_create_trigger(
+ "on_edi_exchange_done_ack_missing", "On ACK missing"
+ )
+ cls.trigger_ack_received_error = cls._get_or_create_trigger(
+ "on_edi_exchange_done_ack_received_error", "On ACK received error"
+ )
+ cls.trigger_generate_complete = cls._get_or_create_trigger(
+ "on_edi_exchange_generate_complete", "On generate complete"
+ )
+
+ @classmethod
+ def _get_or_create_trigger(cls, code, name):
+ trigger = cls.trigger_model.search([("code", "=", code)], limit=1)
+ if not trigger:
+ trigger = cls.trigger_model.create({"name": name, "code": code})
+ return trigger
+
+ def _make_conf(self, trigger, name, is_global=True, snippet=None, **overrides):
+ vals = {
+ "name": name,
+ "active": True,
+ "backend_id": self.backend.id,
+ "type_id": self.exchange_type_out.id,
+ "trigger_id": trigger.id,
+ "is_global": is_global,
+ "snippet_do": snippet or self._marker_snippet,
+ }
+ vals.update(overrides)
+ return self.conf_model.create(vals)
+
+ def test_notify_done_triggers_global_conf(self):
+ conf = self._make_conf(self.trigger_done, "Global Done")
+ self.record._notify_done()
+ self.assertEqual(conf.description, f"|{self.record._name}")
+
+ def test_notify_error_triggers_global_conf(self):
+ conf = self._make_conf(self.trigger_error, "Global Error")
+ self.record._notify_error("send_ko")
+ self.assertEqual(conf.description, f"|{self.record._name}")
+
+ def test_notify_ack_received_triggers_global_conf(self):
+ conf = self._make_conf(self.trigger_ack_received, "Global ACK received")
+ self.record._notify_ack_received()
+ self.assertEqual(conf.description, f"|{self.record._name}")
+
+ def test_notify_ack_missing_triggers_global_conf(self):
+ conf = self._make_conf(self.trigger_ack_missing, "Global ACK missing")
+ self.record._notify_ack_missing()
+ self.assertEqual(conf.description, f"|{self.record._name}")
+
+ def test_notify_ack_received_error_triggers_global_conf(self):
+ conf = self._make_conf(
+ self.trigger_ack_received_error, "Global ACK received error"
+ )
+ self.record._notify_ack_received_error()
+ self.assertEqual(conf.description, f"|{self.record._name}")
+
+ def test_non_global_conf_is_ignored(self):
+ conf = self._make_conf(self.trigger_done, "Non Global Done", is_global=False)
+ self.record._notify_done()
+ self.assertFalse(conf.description)
+
+ def test_inactive_global_conf_is_ignored(self):
+ conf = self._make_conf(self.trigger_done, "Inactive Global Done")
+ conf.active = False
+ self.record._notify_done()
+ self.assertFalse(conf.description)
+
+ def test_notify_action_complete_dispatches_to_both_targets(self):
+ """`notify_action_complete` fires the event twice when the related
+ record exists: once with the exchange record as target, once with the
+ related record (partner here)."""
+ conf = self._make_conf(
+ self.trigger_generate_complete, "Global generate complete"
+ )
+ # Sanity check: the exchange record has a related record.
+ self.assertTrue(self.record.related_record_exists)
+ self.record.notify_action_complete("generate")
+ # The snippet appended one marker per call: exchange record then partner.
+ self.assertEqual(
+ conf.description,
+ f"|{self.record._name}|{self.partner._name}",
+ )
+
+ def test_notify_action_complete_no_related_record(self):
+ """When no related record exists, the event fires only on the
+ exchange record itself."""
+ conf = self._make_conf(
+ self.trigger_generate_complete, "Global generate complete - no related"
+ )
+ # Create an exchange record with no related record.
+ orphan_record = self.backend.create_record(
+ "test_csv_output", {"model": False, "res_id": False}
+ )
+ orphan_record.notify_action_complete("generate")
+ self.assertEqual(conf.description, f"|{orphan_record._name}")
+
+ def test_snippet_receives_conf_and_record(self):
+ """The snippet eval context must expose both `conf` (the configuration)
+ and `record` (the target of the event)."""
+ snippet = (
+ "conf.write({'description': 'conf=%s|record=%s' % "
+ "(conf.name, record.display_name)})"
+ )
+ conf = self._make_conf(self.trigger_done, "Context check", snippet=snippet)
+ self.record._notify_done()
+ self.assertEqual(
+ conf.description,
+ f"conf={conf.name}|record={self.record.display_name}",
+ )
+
+ def test_multiple_global_confs_all_executed(self):
+ """All global confs matching the trigger are executed."""
+ conf1 = self._make_conf(self.trigger_done, "Global Done 1")
+ conf2 = self._make_conf(self.trigger_done, "Global Done 2")
+ self.record._notify_done()
+ self.assertEqual(conf1.description, f"|{self.record._name}")
+ self.assertEqual(conf2.description, f"|{self.record._name}")
+
+ # ------------------------------------------------------------------
+ # Filtering tests for `edi_get_conf_global`
+ # ------------------------------------------------------------------
+ def test_filter_by_type_mismatch(self):
+ """A conf bound to a different exchange type must not fire."""
+ conf = self._make_conf(
+ self.trigger_done,
+ "Wrong type",
+ type_id=self.exchange_type_in.id,
+ )
+ self.record._notify_done()
+ self.assertFalse(conf.description)
+
+ def test_filter_by_type_empty_matches(self):
+ """A conf without a type matches any exchange record's type."""
+ conf = self._make_conf(self.trigger_done, "No type", type_id=False)
+ self.record._notify_done()
+ self.assertEqual(conf.description, f"|{self.record._name}")
+
+ def test_filter_by_backend_mismatch(self):
+ """A conf bound to a different backend must not fire."""
+ other_backend = self.env["edi.backend"].create(
+ {
+ "name": "Other backend",
+ "backend_type_id": self.backend.backend_type_id.id,
+ }
+ )
+ # `_constrains_backend` requires backend to be compatible with the type's
+ # backend if the type has one set. Detach the type from the conf to test
+ # only the backend filter.
+ conf = self._make_conf(
+ self.trigger_done,
+ "Wrong backend",
+ backend_id=other_backend.id,
+ type_id=False,
+ )
+ self.record._notify_done()
+ self.assertFalse(conf.description)
+
+ def test_filter_by_backend_empty_matches(self):
+ """A conf without a backend matches any exchange record's backend."""
+ conf = self._make_conf(
+ self.trigger_done,
+ "No backend",
+ backend_id=False,
+ type_id=False,
+ )
+ self.record._notify_done()
+ self.assertEqual(conf.description, f"|{self.record._name}")
+
+ def test_filter_by_model_mismatch(self):
+ """A conf bound to a different model must not fire."""
+ other_model = self.env["ir.model"]._get("res.users")
+ conf = self._make_conf(
+ self.trigger_done,
+ "Wrong model",
+ model_id=other_model.id,
+ )
+ self.record._notify_done()
+ self.assertFalse(conf.description)
+
+ def test_filter_by_model_match(self):
+ """A conf bound to the related record model fires."""
+ partner_model = self.env["ir.model"]._get(self.partner._name)
+ conf = self._make_conf(
+ self.trigger_done,
+ "Matching model",
+ model_id=partner_model.id,
+ )
+ self.record._notify_done()
+ self.assertEqual(conf.description, f"|{self.record._name}")
+
+ def test_filter_by_model_orphan_record(self):
+ """A conf with a model is skipped on records with no related model."""
+ partner_model = self.env["ir.model"]._get(self.partner._name)
+ conf_with_model = self._make_conf(
+ self.trigger_done,
+ "Model bound",
+ model_id=partner_model.id,
+ )
+ conf_no_model = self._make_conf(self.trigger_done, "Model-less")
+ orphan_record = self.backend.create_record(
+ "test_csv_output", {"model": False, "res_id": False}
+ )
+ orphan_record._notify_done()
+ self.assertFalse(conf_with_model.description)
+ self.assertEqual(conf_no_model.description, f"|{orphan_record._name}")
+
+ def test_edi_get_conf_global_returns_only_matching(self):
+ """Direct check on the new helper method."""
+ matching = self._make_conf(self.trigger_done, "Matching")
+ wrong_trigger = self._make_conf(self.trigger_error, "Wrong trigger")
+ non_global = self._make_conf(self.trigger_done, "Non global", is_global=False)
+ result = self.env["edi.configuration"].edi_get_conf_global(
+ self.record, self.trigger_done.code
+ )
+ self.assertIn(matching, result)
+ self.assertNotIn(wrong_trigger, result)
+ self.assertNotIn(non_global, result)
diff --git a/edi_core_oca/views/edi_configuration_views.xml b/edi_core_oca/views/edi_configuration_views.xml
index 053db7764..674e88f8b 100644
--- a/edi_core_oca/views/edi_configuration_views.xml
+++ b/edi_core_oca/views/edi_configuration_views.xml
@@ -74,6 +74,7 @@
name="model_id"
options="{'no_create': True, 'no_create_edit': True}"
/>
+
+
+
+
+
+
+
+
+
diff --git a/edi_purchase_oca/tests/__init__.py b/edi_purchase_oca/tests/__init__.py
new file mode 100644
index 000000000..6bdd2b970
--- /dev/null
+++ b/edi_purchase_oca/tests/__init__.py
@@ -0,0 +1,2 @@
+from . import test_generate
+from . import test_order
diff --git a/edi_purchase_oca/tests/common.py b/edi_purchase_oca/tests/common.py
new file mode 100644
index 000000000..80d99c031
--- /dev/null
+++ b/edi_purchase_oca/tests/common.py
@@ -0,0 +1,82 @@
+# Copyright 2026 Camptocamp SA
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
+
+from odoo import fields
+from odoo.fields import Command
+
+from odoo.addons.edi_core_oca.tests.common import EDIBackendTestMixin
+
+
+class PurchaseEDIBackendTestMixin(EDIBackendTestMixin):
+ @classmethod
+ def _get_backend_type(cls):
+ backend_type = cls.env["edi.backend.type"].search(
+ [("code", "=", "purchase_demo")], limit=1
+ )
+ if backend_type:
+ return backend_type
+ return cls.env["edi.backend.type"].create(
+ {
+ "name": "Purchase DEMO",
+ "code": "purchase_demo",
+ }
+ )
+
+ @classmethod
+ def _get_backend(cls):
+ backend_type = cls._get_backend_type()
+ backend = cls.env["edi.backend"].search(
+ [("backend_type_id", "=", backend_type.id)], limit=1
+ )
+ if backend:
+ return backend
+ return cls.env["edi.backend"].create(
+ {
+ "name": "purchase DEMO",
+ "backend_type_id": backend_type.id,
+ }
+ )
+
+ @classmethod
+ def _create_exchange_type(cls, **kw):
+ model = cls.env["edi.exchange.type"]
+ code = kw.get("code")
+ if code:
+ exchange_type = model.search(
+ [("code", "=", code), ("backend_id", "=", cls.backend.id)], limit=1
+ )
+ if exchange_type:
+ return exchange_type
+ return super()._create_exchange_type(**kw)
+
+
+class OrderMixin:
+ @classmethod
+ def _create_purchase_order(cls, **kw):
+ model = cls.env["purchase.order"]
+ vals = {
+ "partner_id": cls.vendor.id,
+ "user_id": cls.env.ref("base.user_admin").id,
+ "date_planned": fields.Datetime.now(),
+ }
+ vals.update(kw)
+ if hasattr(model, "play_onchanges"):
+ po_vals = model.play_onchanges(vals, [])
+ else:
+ po_vals = vals.copy()
+ if "order_line" in vals:
+ po_vals["order_line"] = [Command.create(x) for x in vals["order_line"]]
+ return model.create(po_vals)
+
+ @classmethod
+ def _setup_order_records(cls):
+ cls.vendor = cls.env["res.partner"].create(
+ {"name": "ACME inc", "country_id": cls.env.company.country_id.id}
+ )
+ cls.product = cls.env["product.product"].create(
+ {
+ "name": "Product 1",
+ "default_code": "1234567",
+ "purchase_ok": True,
+ }
+ )
diff --git a/edi_purchase_oca/tests/test_generate.py b/edi_purchase_oca/tests/test_generate.py
new file mode 100644
index 000000000..0f814b911
--- /dev/null
+++ b/edi_purchase_oca/tests/test_generate.py
@@ -0,0 +1,96 @@
+# Copyright 2026 Camptocamp SA
+# @author Simone Orsi
+
+EDI Purchase
+ +
Handle purchase orders via EDI.
+This is a base module to plug purchase processes with the EDI framework.
+To handle inbound/outbound purchase orders, you need to create your own +integration modules on top of this base module.
+Table of contents
+
+
+-
+
- Bug Tracker +
- Credits + +
+
+Bug Tracker
+Bugs are tracked on GitHub Issues. +In case of trouble, please check there if your issue has already been reported. +If you spotted it first, help us to smash it by providing a detailed and welcomed +feedback.
+Do not contact contributors directly about support or help with technical issues.
+
+
+Credits
+
+
+Authors
+-
+
- ForgeFlow +
- Camptocamp +
+
+Contributors
+-
+
- Lois Rilo lois.rilo@forgeflow.com +
- Simone Orsi simone.orsi@camptocamp.com +
- Phan Hong Phuc <phucph@trobz.com> +
- Maksym Yankin maksym.yankin@camptocamp.com +
+
+
+
+Maintainers
+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.
+This module is part of the OCA/edi-framework project on GitHub.
+You are welcome to contribute. To learn how please visit https://odoo-community.org/page/Contribute.
+-
+
+
+
+
+
+
+
+
+
diff --git a/edi_purchase_ubl_output_oca/tests/__init__.py b/edi_purchase_ubl_output_oca/tests/__init__.py
new file mode 100644
index 000000000..427d09a22
--- /dev/null
+++ b/edi_purchase_ubl_output_oca/tests/__init__.py
@@ -0,0 +1 @@
+from . import test_generate
diff --git a/edi_purchase_ubl_output_oca/tests/test_generate.py b/edi_purchase_ubl_output_oca/tests/test_generate.py
new file mode 100644
index 000000000..9cddc6dc5
--- /dev/null
+++ b/edi_purchase_ubl_output_oca/tests/test_generate.py
@@ -0,0 +1,98 @@
+# Copyright 2026 Camptocamp SA
+# @author Simone Orsi
+
+EDI UBL Purchase
+ +
Handle purchase exchanges with the EDI framework.
+This module is mostly a glue module for purchase_order_ubl with edi_oca.
+Allows you to generate and send purchase orders as UBL XML files.
+
+
+Important
+This is an alpha version, the data model and design can change at any time without warning. +Only for development or testing purpose, do not use in production. +More details on development status
+Table of contents
+
+
+-
+
- Bug Tracker +
- Credits + +
+
+Bug Tracker
+Bugs are tracked on GitHub Issues. +In case of trouble, please check there if your issue has already been reported. +If you spotted it first, help us to smash it by providing a detailed and welcomed +feedback.
+Do not contact contributors directly about support or help with technical issues.
+
+
+Credits
+
+
+Authors
+-
+
- Camptocamp +
+
+Contributors
+-
+
- Simone Orsi <simone.orsi@camptocamp.com> +
+
+
+
+Maintainers
+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.
+Current maintainer:
+
This module is part of the OCA/edi project on GitHub.
+You are welcome to contribute. To learn how please visit https://odoo-community.org/page/Contribute.
+