Rework push routing rules, with reclassification

stock_routing_operation/__manifest__.py

@@ -13,11 +13,7 @@
         "demo/stock_picking_type_demo.xml",
         "demo/stock_routing_demo.xml",
     ],
-    "data": [
-        "views/stock_location_views.xml",
-        "views/stock_routing_views.xml",
-        "security/ir.model.access.csv",
-    ],
+    "data": ["views/stock_routing_views.xml", "security/ir.model.access.csv"],
     "installable": True,
     "development_status": "Alpha",
 }

stock_routing_operation/models/stock_move.py

@@ -202,7 +202,10 @@ def _apply_routing_rule_pull(self):
             # original_destination = move.move_line_ids[0].location_dest_id
             original_destination = move.location_dest_id
 
+            current_source_location = move.location_id
             current_picking_type = move.picking_id.picking_type_id
+            move.location_id = routing_rule.location_src_id
+            move.picking_type_id = routing_rule.picking_type_id
             if self.env["stock.location"].search(
                 [
                     ("id", "=", routing_rule.location_dest_id.id),
@@ -212,35 +215,24 @@ def _apply_routing_rule_pull(self):
                 # The destination of the move, as a parent of the destination
                 # of the routing, goes to the correct place, but is not precise
                 # enough: set the new destination to match the rule's one
-                move.location_id = routing_rule.location_src_id
                 move.location_dest_id = routing_rule.location_dest_id
-                move.picking_type_id = routing_rule.picking_type_id
 
-            elif self.env["stock.location"].search(
+            elif not self.env["stock.location"].search(
                 [
                     ("id", "=", routing_rule.location_dest_id.id),
                     ("id", "parent_of", move.location_dest_id.id),
                 ]
             ):
-                # The destination of the move is already more precise than the
-                # expected destination of the routing: keep it, but we still
-                # want to change the picking type
-                move.location_id = routing_rule.location_src_id
-                move.picking_type_id = routing_rule.picking_type_id
-            else:
                 # The destination of the move is unrelated (nor identical, nor
                 # a parent or a child) to the routing destination: in this case
                 # we have to add a routing move before the current move to
                 # route the goods in the correct place
-                source_location = move.location_id
-                move.location_id = routing_rule.location_src_id
                 move.location_dest_id = routing_rule.location_dest_id
-                move.picking_type_id = routing_rule.picking_type_id
                 # create a copy of the move with the current picking type and
                 # going to its original destination: it will be assigned to the
                 # same picking as the original picking of our move
                 move._insert_routing_moves(
-                    current_picking_type, source_location, original_destination
+                    current_picking_type, current_source_location, original_destination
                 )
 
             pickings_to_check_for_emptiness |= move.picking_id
@@ -315,6 +307,7 @@ def _split_and_set_rule_for_routing_push(self):
                     move.push_routing_rule_id = rule
                 continue
 
+            # TODO split when we split pull
             for rule, move_lines in routing_rules.items():
                 if not rule:
                     # No routing operation is required for these moves,
@@ -345,6 +338,7 @@ def _apply_routing_rule_push(self):
         type with the routing operation ones and creates a new chained move
         after it.
         """
+        pickings_to_check_for_emptiness = self.env["stock.picking"]
         for move in self:
             if move.state not in ("assigned", "partially_available"):
                 continue
@@ -361,28 +355,39 @@ def _apply_routing_rule_push(self):
             if not routing_rule:
                 continue
             if move.picking_id.picking_type_id == routing_rule.picking_type_id:
-                # already correct
+                # the routing rule has already been applied and re-classified
+                # the move in the picking type
+                continue
+            if move.location_dest_id == routing_rule.location_src_id:
+                # the routing rule has already been applied and added a new
+                # routing move after this one
                 continue
 
             if self.env["stock.location"].search(
                 [
+                    # the source is already correct (more precise than the routing),
+                    # but we still want to classify the move in the routing's picking
+                    # type
                     ("id", "=", routing_rule.location_src_id.id),
+                    # if the source location of the move is a child of the routing's
+                    # source location, we don't need to change it
                     ("id", "parent_of", move.location_id.id),
                 ]
             ):
-                # This move has been created for the routing operation,
-                # or was already created with the correct locations anyway,
-                # exit or it would indefinitely add a next move
-                continue
-
-            # Move the goods in the "routing" location instead.
-            # In this use case, we want to keep the move lines so we don't
-            # change the reservation.
-            move.write({"location_dest_id": routing_rule.location_src_id.id})
-            move.move_line_ids.write(
-                {"location_dest_id": routing_rule.location_src_id.id}
-            )
+                move.picking_type_id = routing_rule.picking_type_id
+                pickings_to_check_for_emptiness |= move.picking_id
+                move._assign_picking()
+            else:
+                # Fall here when the source location is unrelated to the
+                # routing's one. Redirect the move and move line to go through
+                # the routing and add a new move after it to reach the
+                # destination of the routing.
+                move.location_dest_id = routing_rule.location_src_id
+                move.move_line_ids.location_dest_id = routing_rule.location_src_id
+                move._insert_routing_moves(
+                    routing_rule.picking_type_id,
+                    routing_rule.location_src_id,
+                    destination,
+                )
 
-            move._insert_routing_moves(
-                routing_rule.picking_type_id, routing_rule.location_src_id, destination
-            )
+        pickings_to_check_for_emptiness._routing_operation_handle_empty()

stock_routing_operation/tests/test_routing_operation_dest.py

@@ -11,7 +11,7 @@ def setUpClass(cls):
         cls.wh = cls.env["stock.warehouse"].create(
             {
                 "name": "Base Warehouse",
-                "reception_steps": "one_step",
+                "reception_steps": "two_steps",
                 "delivery_steps": "pick_ship",
                 "code": "WHTEST",
             }
@@ -198,7 +198,6 @@ def test_change_location_to_routing_operation(self):
         # | Stock/Handover → Highbay                                          |
         # | Product1 Stock/Highbay/Handover → Highbay1-2 (waiting) added_move |
         # +-------------------------------------------------------------------+
-
         self.assert_src_supplier(move_a)
         self.assert_dest_input(move_a)
         self.assert_src_input(move_b)
@@ -551,3 +550,93 @@ def test_several_move_lines(self):
         self.assertEqual(move_b_shelf.state, "done")
         self.assertEqual(move_b_handover.state, "done")
         self.assertEqual(routing_move.state, "done")
+
+    def test_classify_picking_type_sub_location(self):
+        # When a move already comes from a location within the source location
+        # of the routing's picking type, we don't need a new routing move, but
+        # we want to re-classify the move in a stock.picking of the routing's
+        # picking type.
+        # For this test, we create a handover inside Input, and we change the
+        # routing to be Input -> Highbay. Then we change the moves to go
+        # through Input Handover, to match the picking type.
+        # The source location of the move stays "Input Handover" because it is already
+        # more precise as the "Input" of the picking type.
+        input_ho_location = self.env["stock.location"].create(
+            {"location_id": self.wh.wh_input_stock_loc_id.id, "name": "Input Handover"}
+        )
+        # any move from input (and sub-locations) to highbay has to be classified in
+        # our picking type
+        self.pick_type_routing_op.default_location_src_id = (
+            self.wh.wh_input_stock_loc_id
+        )
+
+        in_picking, internal_picking = self._create_supplier_input_highbay(
+            self.wh, [(self.product1, 10, self.location_hb_1_2)]
+        )
+        move_a = in_picking.move_lines
+        move_b = internal_picking.move_lines
+        # go through our Input Handover location, as it is under the source location
+        # of the routing's picking type, we should not have an additional move,
+        # but move_b must be classified in the routing's picking type
+        move_a.location_dest_id = input_ho_location
+        move_a.move_line_ids.location_dest_id = input_ho_location
+        move_b.location_id = input_ho_location
+
+        self.process_operations(move_a)
+
+        self.assertEqual(move_a.state, "done")
+
+        # move B is classified in a new picking
+        self.assertEqual(move_b.push_routing_rule_id, self.routing.rule_ids)
+        self.assertEqual(move_b.state, "assigned")
+        self.assertEqual(move_b.location_id, input_ho_location)
+        self.assertEqual(move_b.move_line_ids.location_id, input_ho_location)
+        self.assertEqual(move_b.picking_id.location_id, input_ho_location)
+        self.assert_dest_highbay_1_2(move_b)
+        self.assert_dest_highbay_1_2(move_b.move_line_ids)
+        self.assert_dest_highbay_1_2(move_b.picking_id)
+        self.assertEqual(move_b.picking_id.picking_type_id, self.pick_type_routing_op)
+        self.assertFalse(move_b.move_dest_ids)
+
+    def test_picking_type_super_location_extra_move(self):
+        # When a move comes from a location above the source location of the
+        # routing's picking type, we need an extra move to reach the particular
+        # space in the location (example: the goods were brought to Input, but the
+        # picking type is "Input/Handover -> Highbay"), we'll need an extra move to
+        # move goods from Input to Input/Handover).
+        # For this test, we create a handover inside Input, and we change the
+        # routing to be "Input Handover" -> Highbay. And we change the routing source
+        # location to "Input Handover".
+        input_ho_location = self.env["stock.location"].create(
+            {"location_id": self.wh.wh_input_stock_loc_id.id, "name": "Input Handover"}
+        )
+        # any move from input (and sub-locations) to highbay has to be classified in
+        # our picking type
+        self.pick_type_routing_op.default_location_src_id = input_ho_location
+
+        in_picking, internal_picking = self._create_supplier_input_highbay(
+            self.wh, [(self.product1, 10, self.location_hb_1_2)]
+        )
+        move_a = in_picking.move_lines
+        move_b = internal_picking.move_lines
+
+        self.process_operations(move_a)
+
+        self.assertEqual(move_a.state, "done")
+
+        self.assertEqual(move_b.push_routing_rule_id, self.routing.rule_ids)
+        self.assertEqual(move_b.state, "assigned")
+        self.assert_src_input(move_b)
+        self.assertEqual(move_b.location_dest_id, input_ho_location)
+        self.assertEqual(move_b.move_line_ids.location_dest_id, input_ho_location)
+
+        # we have an extra move to reach the Highbay from Input/Handover
+        extra_move = move_b.move_dest_ids
+        self.assert_dest_highbay_1_2(extra_move)
+        self.assert_dest_highbay_1_2(extra_move.picking_id)
+        self.assertEqual(
+            extra_move.picking_id.picking_type_id, self.pick_type_routing_op
+        )
+        self.assertFalse(extra_move.move_dest_ids)
+
+        # TODO tests for domains