[14.0][IMP] base_exception: Better general Handling of Sub-Exceptions

base_exception/models/base_exception.py

@@ -1,5 +1,6 @@
 # Copyright 2011 Raphaël Valyi, Renato Lima, Guewen Baconnier, Sodexis
 # Copyright 2017 Akretion (http://www.akretion.com)
+# Copyright 2025 Raumschmiede GmbH
 # Mourad EL HADJ MIMOUNE <[email protected]>
 # Copyright 2020 Hibou Corp.
 # Copyright 2023 ACSONE SA/NV (http://acsone.eu)
@@ -91,6 +92,22 @@ def _popup_exceptions(self):
     def _get_popup_action(self):
         return self.env.ref("base_exception.action_exception_rule_confirm")
 
+    def _add_detected_exceptions_to_self(self):
+        return self != self._get_main_records()
+
+    def _detect_exceptions(self, rule):
+        records = super()._detect_exceptions(rule)
+        # If _get_main_records returns self, it adds the exceptions to itself in
+        # detect_exceptions. If _get_main_records does not return self, it adds the
+        # exceptions here
+        if not self._add_detected_exceptions_to_self():
+            return records
+
+        (self - records).exception_ids = [(3, rule.id)]
+        records.exception_ids = [(4, rule.id)]
+
+        return records
+
     def _check_exception(self):
         """Check exceptions
 

base_exception/models/base_exception_method.py

@@ -3,6 +3,7 @@
 # Mourad EL HADJ MIMOUNE <[email protected]>
 # Copyright 2020 Hibou Corp.
 # Copyright 2023 ACSONE SA/NV (http://acsone.eu)
+# Copyright 2025 Raumschmiede GmbH
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
 
 import logging
@@ -28,6 +29,13 @@ def _get_main_records(self):
         """
         return self
 
+    def _get_sub_exception_field_names(self):
+        """
+        Use this to check exceptions on underlying records and the
+        detected exceptions need to be added to the current record
+        """
+        return []
+
     def _rule_domain(self):
         """Filter exception.rules.
         By default, only the rules with the correct model
@@ -52,7 +60,9 @@ def detect_exceptions(self):
             records_with_rule_in_exceptions = main_records.filtered(
                 lambda r, rule_id=rule_info.id: rule_id in r.exception_ids.ids
             )
-            records_with_exception = self._detect_exceptions(rule_info)
+            records_with_exception = self._detect_exceptions(
+                rule_info
+            )._get_main_records()
             to_remove = records_with_rule_in_exceptions - records_with_exception
             to_add = records_with_exception - records_with_rule_in_exceptions
             # we expect to always work on the same model type
@@ -64,23 +74,18 @@ def detect_exceptions(self):
             rules_to_add[rule_info.id] |= to_add
             if records_with_exception:
                 all_exception_ids.append(rule_info.id)
-        # Cumulate all the records to attach to the rule
-        # before linking. We don't want to call "rule.write()"
-        # which would:
-        # * write on write_date so lock the exception.rule
-        # * trigger the recomputation of "main_exception_id" on
-        #   all the sale orders related to the rule, locking them all
-        #   and preventing concurrent writes
-        # Reversing the write by writing on SaleOrder instead of
-        # ExceptionRule fixes the 2 kinds of unexpected locks.
-        # It should not result in more queries than writing on ExceptionRule:
-        # the "to remove" part generates one DELETE per rule on the relation
-        # table
-        # and the "to add" part generates one INSERT (with unnest) per rule.
+
         for rule_id, records in rules_to_remove.items():
             records.write({"exception_ids": [(3, rule_id)]})
         for rule_id, records in rules_to_add.items():
             records.write({"exception_ids": [(4, rule_id)]})
+
+        # Detect exceptions on underlying sub-records if needed. If the sub-records'
+        # model defines the current model in _get_main_records,
+        # the exceptions detected are added to the current record
+        for field in self._get_sub_exception_field_names():
+            all_exception_ids += self.mapped(field).detect_exceptions()
+
         return all_exception_ids
 
     @api.model

base_exception/readme/CONTRIBUTORS.rst

@@ -14,3 +14,4 @@
 
 * Kevin Khao <[email protected]>
 * Laurent Mignon <[email protected]>
+* Joshua Lauer <[email protected]>

base_exception/tests/common.py

@@ -16,17 +16,24 @@ def setUpClass(cls):
 
         cls.loader = FakeModelLoader(cls.env, cls.__module__)
         cls.loader.backup_registry()
-        from .purchase_test import ExceptionRule, LineTest, PurchaseTest
+        from .purchase_test import ExceptionRule, LineTest, LineTestMethod, PurchaseTest
 
-        cls.loader.update_registry((ExceptionRule, LineTest, PurchaseTest))
+        cls.loader.update_registry(
+            (ExceptionRule, LineTest, LineTestMethod, PurchaseTest)
+        )
 
         cls.partner = cls.env["res.partner"].create({"name": "Foo"})
         cls.po = cls.env["base.exception.test.purchase"].create(
             {
                 "name": "Test base exception to basic purchase",
                 "partner_id": cls.partner.id,
                 "line_ids": [
-                    (0, 0, {"name": "line test", "amount": 120.0, "qty": 1.5})
+                    (0, 0, {"name": "line test", "amount": 120.0, "qty": 1.5}),
+                    (0, 0, {"name": "line test 2", "amount": 220.0, "qty": 1.5}),
+                ],
+                "line_method_ids": [
+                    (0, 0, {"name": "line test", "amount": 120.0, "qty": 1.5}),
+                    (0, 0, {"name": "line test 2", "amount": 220.0, "qty": 1.5}),
                 ],
             }
         )
@@ -40,6 +47,26 @@ def setUpClass(cls):
                 "exception_type": "by_py_code",
             }
         )
+        cls.sub_exception_rule = cls.env["exception.rule"].create(
+            {
+                "name": "Amount less than 100",
+                "description": "Line must have price greater than 100",
+                "sequence": 9,
+                "model": "base.exception.test.purchase.line",
+                "code": "failed = self.amount < 100",
+                "exception_type": "by_py_code",
+            }
+        )
+        cls.sub_exception_rule_method = cls.env["exception.rule"].create(
+            {
+                "name": "Amount less than 100",
+                "description": "Method Line must have price greater than 100",
+                "sequence": 9,
+                "model": "base.exception.method.test.purchase.line",
+                "code": "failed = self.amount < 100",
+                "exception_type": "by_py_code",
+            }
+        )
 
     @classmethod
     def tearDownClass(cls):

base_exception/tests/purchase_test.py

@@ -13,8 +13,16 @@ class ExceptionRule(models.Model):
         selection_add=[("exception_method_no_zip", "Purchase exception no zip")]
     )
     model = fields.Selection(
-        selection_add=[("base.exception.test.purchase", "Purchase Test")],
-        ondelete={"base.exception.test.purchase": "cascade"},
+        selection_add=[
+            ("base.exception.test.purchase", "Purchase Test"),
+            ("base.exception.test.purchase.line", "Purchase Test Line"),
+            ("base.exception.method.test.purchase.line", "Purchase Test Method Line"),
+        ],
+        ondelete={
+            "base.exception.test.purchase": "cascade",
+            "base.exception.test.purchase.line": "cascade",
+            "base.exception.method.test.purchase.line": "cascade",
+        },
     )
 
 
@@ -40,6 +48,9 @@ class PurchaseTest(models.Model):
     active = fields.Boolean(default=True)
     partner_id = fields.Many2one("res.partner", string="Partner")
     line_ids = fields.One2many("base.exception.test.purchase.line", "lead_id")
+    line_method_ids = fields.One2many(
+        "base.exception.method.test.purchase.line", "lead_id"
+    )
     amount_total = fields.Float(compute="_compute_amount_total", store=True)
 
     @api.depends("line_ids")
@@ -48,7 +59,7 @@ def _compute_amount_total(self):
             for line in record.line_ids:
                 record.amount_total += line.amount * line.qty
 
-    @api.constrains("ignore_exception", "line_ids", "state")
+    @api.constrains("ignore_exception", "line_ids", "line_method_ids", "state")
     def test_purchase_check_exception(self):
         orders = self.filtered(lambda s: s.state == "purchase")
         if orders:
@@ -69,6 +80,9 @@ def button_confirm(self):
     def button_cancel(self):
         self.write({"state": "cancel"})
 
+    def _get_sub_exception_field_names(self):
+        return ["line_ids", "line_method_ids"]
+
     def exception_method_no_zip(self):
         records_fail = self.env["base.exception.test.purchase"]
         for rec in self:
@@ -78,10 +92,33 @@ def exception_method_no_zip(self):
 
 
 class LineTest(models.Model):
+    _inherit = "base.exception"
     _name = "base.exception.test.purchase.line"
     _description = "Base Exception Test Model Line"
 
     name = fields.Char()
     lead_id = fields.Many2one("base.exception.test.purchase", ondelete="cascade")
     qty = fields.Float()
     amount = fields.Float()
+
+    def _get_main_records(self):
+        return self.lead_id
+
+
+class LineTestMethod(models.Model):
+    _inherit = "base.exception.method"
+    _name = "base.exception.method.test.purchase.line"
+    _description = "Base Exception Test Model Line"
+
+    name = fields.Char()
+    lead_id = fields.Many2one("base.exception.test.purchase", ondelete="cascade")
+    qty = fields.Float()
+    amount = fields.Float()
+
+    # Models inheriting from .method must implement this field as their records
+    # are filtered based on this field
+    ignore_exception = fields.Boolean("Ignore Exceptions", copy=False)
+
+    # This model here must override _get_main_records as it has no exception_ids
+    def _get_main_records(self):
+        return self.lead_id

base_exception/tests/test_base_exception.py

@@ -72,3 +72,71 @@ def test_blocking_exception(self):
             self.po.button_confirm()
         self.assertEqual(self.po.exception_ids, self.exception_rule)
         self.assertTrue(self.po.exceptions_summary)
+
+    def test_exception_in_sub_records(self):
+        self.exception_rule.active = False
+        line = self.po.line_ids[0]
+        line.amount = 90
+
+        self.po.detect_exceptions()
+
+        self.assertEqual(self.po.exception_ids, self.sub_exception_rule)
+        # Even if the exception was not raised by self.po, its description must still
+        # be in the summary of it
+        self.assertIn(self.sub_exception_rule.description, self.po.exceptions_summary)
+
+        self.assertEqual(line.main_exception_id, self.sub_exception_rule)
+        self.assertEqual(line.exception_ids, self.sub_exception_rule)
+        self.assertIn(
+            self.sub_exception_rule.description,
+            line.exceptions_summary,
+        )
+        # self.po has 2 lines, the exception is detected only on the 1st line.
+        # The 2nd line must not have any exception assigned or summary set
+        self.assertFalse(self.po.line_ids[1].exception_ids)
+        self.assertFalse(self.po.line_ids[1].exceptions_summary)
+
+        self.exception_rule.active = True
+
+        self.po.detect_exceptions()
+
+        # Now both exceptions must be assigned to the record, in the right order
+        self.assertEqual(self.po.exception_ids[0], self.sub_exception_rule)
+        self.assertEqual(self.po.exception_ids[1], self.exception_rule)
+
+        self.po.line_ids[1].amount = 80
+        self.po.detect_exceptions()
+
+        self.assertEqual(line.exception_ids, self.sub_exception_rule)
+        self.assertEqual(self.po.line_ids[1].exception_ids, self.sub_exception_rule)
+        self.assertIn(
+            self.sub_exception_rule.description,
+            self.po.line_ids[1].exceptions_summary,
+        )
+
+        line.amount = 200
+        line.detect_exceptions()
+
+        # Updating sub-exceptions must update exceptions on parent
+        self.assertFalse(line.exception_ids)
+        self.assertEqual(self.po.exception_ids, self.exception_rule)
+
+    def test_exception_in_sub_record_method(self):
+        self.exception_rule.active = False
+
+        self.po.line_method_ids.amount = 90
+        # Model has no exception_ids and returns self.lead_id in _get_main_records,
+        # that's why the exceptions are added to the parent
+        self.po.line_method_ids.detect_exceptions()
+
+        self.assertEqual(self.po.exception_ids, self.sub_exception_rule_method)
+        self.assertIn(
+            self.sub_exception_rule_method.description,
+            self.po.exceptions_summary,
+        )
+
+        self.po.line_method_ids.amount = 200
+        self.po.detect_exceptions()
+        # Parent implemented line_method_ids in _get_sub_exception_field_names,
+        # that's why the exceptions were detected on child records but none was found
+        self.assertFalse(self.po.exception_ids)