[IMP] stock_owner_restriction: increase method coverage

Extend following methods for the scenario where owner restriction should be forced:

- _update_available_quantity() (stock.quant)
- _update_reserved_quantity() (stock.quant)
- Split lines to identify the owner_restriction setting into separate method for
  better extensibility.
- Fix read_group() (stock.quant) which was incorrectly assigning the owner record
  in the domain instead of the id.
- Add compute for restrict_partner_id to control owner assignment based on picking type restriction.
- Unlink stock move lines depending on owner_restriction if owner_id is updated.
- Update test cases

Author: AungKoKoLin1997

stock_owner_restriction/models/product.py

@@ -14,12 +14,12 @@ def _compute_quantities_dict(
             return super()._compute_quantities_dict(
                 lot_id, owner_id, package_id, from_date=from_date, to_date=to_date
             )
-        restricted_owner_id = self.env.context.get("force_restricted_owner_id", None)
-        if owner_id is None and restricted_owner_id is None:
+        restricted_owner = self.env.context.get("force_restricted_owner_id", None)
+        if owner_id is None and restricted_owner is None:
             # Force owner to False if is None
             owner_id = False
-        elif restricted_owner_id:
-            owner_id = restricted_owner_id
+        elif restricted_owner:
+            owner_id = restricted_owner.id
         return super()._compute_quantities_dict(
             lot_id, owner_id, package_id, from_date=from_date, to_date=to_date
         )

stock_owner_restriction/models/stock_move.py

@@ -1,14 +1,30 @@
 # Copyright 2020 Carlos Dauden - Tecnativa
 # Copyright 2020 Sergio Teruel - Tecnativa
+# Copyright 2023-2024 Quartile
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
 from collections import defaultdict
 
-from odoo import models
+from odoo import api, fields, models
 
 
 class StockMove(models.Model):
     _inherit = "stock.move"
 
+    # Add compute method to the standard field
+    restrict_partner_id = fields.Many2one(
+        compute="_compute_restrict_partner_id",
+        store=True,
+        readonly=False,
+    )
+
+    @api.depends("picking_type_id", "picking_id.owner_id", "move_dest_ids")
+    def _compute_restrict_partner_id(self):
+        for move in self:
+            if move.picking_type_id.owner_restriction == "picking_partner":
+                move.restrict_partner_id = move._get_owner_for_assign()
+            else:
+                move.restrict_partner_id = False
+
     def _get_moves_to_assign_with_standard_behavior(self):
         """This method is expected to be extended as necessary. e.g. you may not want to
         handle subcontracting receipts (whose picking type is normal incoming receipt
@@ -20,9 +36,16 @@ def _get_moves_to_assign_with_standard_behavior(self):
             or m.picking_type_id.owner_restriction == "standard_behavior"
         )
 
+    def _get_owner_restriction(self):
+        """This method is expected to be extended as necessary. e.g. different logic
+        needs to be applied to moves in unbuild orders.
+        """
+        self.ensure_one()
+        return self.picking_type_id.owner_restriction
+
     def _get_owner_for_assign(self):
         """This method is expected to be extended as necessary. e.g. different logic
-        needs to be applied for moves in manufacturing orders.
+        needs to be applied to moves in manufacturing orders.
         """
         self.ensure_one()
         partner = self.move_dest_ids.picking_id.owner_id
@@ -37,7 +60,8 @@ def _action_assign(self, force_qty=False):
         res = super(StockMove, moves)._action_assign(force_qty=force_qty)
         dict_key = defaultdict(lambda: self.env["stock.move"])
         for move in self - moves:
-            if move.picking_type_id.owner_restriction == "unassigned_owner":
+            owner_restriction = move._get_owner_restriction()
+            if owner_restriction == "unassigned_owner":
                 dict_key[False] |= move
             else:
                 partner = move._get_owner_for_assign()

stock_owner_restriction/models/stock_picking.py

@@ -8,3 +8,10 @@ class StockPicking(models.Model):
     _inherit = "stock.picking"
 
     owner_restriction = fields.Selection(related="picking_type_id.owner_restriction")
+
+    def write(self, vals):
+        if "owner_id" in vals:
+            for pick in self:
+                if pick.owner_restriction in ("unassigned_owner", "picking_partner"):
+                    pick.move_line_ids.unlink()
+        return super().write(vals)

stock_owner_restriction/models/stock_quant.py

@@ -1,5 +1,6 @@
 # Copyright 2020 Carlos Dauden - Tecnativa
 # Copyright 2020 Sergio Teruel - Tecnativa
+# Copyright 2023 Quartile Limited
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
 from odoo import api, models
 from odoo.osv import expression
@@ -33,20 +34,29 @@ def _gather(
             owner_id=self._get_restriction_owner_id(location_id, owner_id),
             strict=strict,
         )
-        restricted_owner_id = self.env.context.get("force_restricted_owner_id", None)
-        if owner_id is None or restricted_owner_id is None:
+        restricted_owner = self.env.context.get("force_restricted_owner_id", None)
+        if owner_id is None or restricted_owner is None:
             return records
         return records.filtered(
-            lambda q: q.owner_id == (restricted_owner_id or self.env["res.partner"])
+            lambda q: q.owner_id == (restricted_owner or self.env["res.partner"])
         )
 
     @api.model
     def read_group(
         self, domain, fields, groupby, offset=0, limit=None, orderby=False, lazy=True
     ):
-        restricted_owner_id = self.env.context.get("force_restricted_owner_id", None)
-        if restricted_owner_id is not None:
-            domain = expression.AND([domain, [("owner_id", "=", restricted_owner_id)]])
+        owner_in_domain = any(t[0] == "owner_id" for t in domain)
+        if not owner_in_domain:
+            restricted_owner = self.env.context.get("force_restricted_owner_id", None)
+            if restricted_owner is not None:
+                owner_domain = [
+                    (
+                        "owner_id",
+                        "=",
+                        restricted_owner and restricted_owner.id or False,
+                    ),
+                ]
+                domain = expression.AND([domain, owner_domain])
         return super(StockQuant, self).read_group(
             domain,
             fields,
@@ -68,9 +78,9 @@ def _get_available_quantity(
         strict=False,
         allow_negative=False,
     ):
-        restricted_owner_id = self.env.context.get("force_restricted_owner_id", None)
-        if not owner_id and restricted_owner_id is not None:
-            owner_id = restricted_owner_id
+        restricted_owner = self.env.context.get("force_restricted_owner_id", None)
+        if not owner_id and restricted_owner is not None:
+            owner_id = restricted_owner
         return super()._get_available_quantity(
             product_id,
             location_id,
@@ -80,3 +90,51 @@ def _get_available_quantity(
             strict=strict,
             allow_negative=allow_negative,
         )
+
+    @api.model
+    def _update_available_quantity(
+        self,
+        product_id,
+        location_id,
+        quantity,
+        lot_id=None,
+        package_id=None,
+        owner_id=None,
+        in_date=None,
+    ):
+        restricted_owner = self.env.context.get("force_restricted_owner_id", None)
+        if not owner_id and restricted_owner is not None:
+            owner_id = restricted_owner
+        return super()._update_available_quantity(
+            product_id,
+            location_id,
+            quantity,
+            lot_id=lot_id,
+            package_id=package_id,
+            owner_id=owner_id,
+            in_date=in_date,
+        )
+
+    @api.model
+    def _update_reserved_quantity(
+        self,
+        product_id,
+        location_id,
+        quantity,
+        lot_id=None,
+        package_id=None,
+        owner_id=None,
+        strict=False,
+    ):
+        restricted_owner = self.env.context.get("force_restricted_owner_id", None)
+        if not owner_id and restricted_owner is not None:
+            owner_id = restricted_owner
+        return super()._update_reserved_quantity(
+            product_id,
+            location_id,
+            quantity,
+            lot_id=lot_id,
+            package_id=package_id,
+            owner_id=owner_id,
+            strict=strict,
+        )

stock_owner_restriction/tests/test_stock_owner_restriction.py

@@ -70,7 +70,7 @@ def test_product_qty_available(self):
         # context depends of product qty_available
         self.assertEqual(
             self.product.with_context(
-                force_restricted_owner_id=self.owner.id
+                force_restricted_owner_id=self.owner
             ).qty_available,
             500.00,
         )
@@ -166,3 +166,29 @@ def test_search_qty(self):
             [("id", "=", self.product.id), ("qty_available", ">", 499.00)]
         )
         self.assertTrue(products)
+
+    def test_update_available_quantity(self):
+        self.env["stock.quant"].with_context(
+            force_restricted_owner_id=self.owner
+        )._update_available_quantity(
+            self.product, self.picking_type_out.default_location_src_id, 100
+        )
+        self.assertEqual(self.product.qty_available, 500.00)
+        # Invalidate the cache of the product record to ensure qty_available is recalculated
+        self.product.invalidate_recordset()
+        self.assertEqual(
+            self.product.with_context(skip_restricted_owner=True).qty_available, 1100.00
+        )
+
+    def test_update_reserved_quantity(self):
+        self.env["stock.quant"].with_context(
+            force_restricted_owner_id=self.owner
+        )._update_reserved_quantity(
+            self.product, self.picking_type_out.default_location_src_id, 100
+        )
+        self.assertEqual(self.product.qty_available, 500.00)
+        # Invalidate the cache of the product record to ensure qty_available is recalculated
+        self.product.invalidate_recordset()
+        self.assertEqual(
+            self.product.with_context(skip_restricted_owner=True).qty_available, 1000.00
+        )