[16.0][BWP][FIX] shopfloor_reception: Add session recover at scan_line

shopfloor_reception/services/reception.py

@@ -8,7 +8,7 @@
 from decorator import contextmanager
 
 from odoo import fields
-from odoo.tools import float_compare
+from odoo.tools import float_compare, float_is_zero
 
 from odoo.addons.base_rest.components.service import to_int
 from odoo.addons.component.core import Component
@@ -285,17 +285,14 @@ def _scan_line__find_or_create_line(self, picking, move, qty_done=1):
         If none are found create a new line.
 
         """
-        line = None
         unassigned_lines = self.env["stock.move.line"]
-        for move_line in move.move_line_ids:
-            if move_line.result_package_id:
-                continue
-            if move_line.shopfloor_user_id.id == self.env.uid:
-                line = move_line
-                break
-            elif not move_line.shopfloor_user_id:
-                unassigned_lines |= move_line
-        if not line and unassigned_lines:
+        for line in move.move_line_ids:
+            if line.shopfloor_user_id.id == self.env.uid:
+                return self._scan_line__recover(picking, line, qty_done)
+            elif not line.shopfloor_user_id:
+                unassigned_lines |= line
+        line = None
+        if unassigned_lines:
             lock = self._actions_for("lock")
             for move_line in unassigned_lines:
                 if lock.for_update(move_line, skip_locked=True):
@@ -306,6 +303,24 @@ def _scan_line__find_or_create_line(self, picking, move, qty_done=1):
             line = self.env["stock.move.line"].create(values)
         return self._scan_line__assign_user(picking, line, qty_done)
 
+    def _scan_line__recover(self, picking, line, default_qty):
+        product = line.product_id
+        message = self.msg_store.recovered_previous_session()
+        # Do not restore further than set_destination, because a destination location
+        # might be set by default, and we want the user to be allowed to change it.
+        if line.result_package_id:
+            # Destination package is set, go to set_destination
+            return self._response_for_set_destination(picking, line, message=message)
+        if product.tracking not in ("lot", "serial") or (line.lot_id or line.lot_name):
+            # If lot already set, go to set_quantity
+            rounding = line.product_uom_id.rounding
+            if float_is_zero(line.qty_done, precision_rounding=rounding):
+                # If no qty_done, set default qty_done
+                line.qty_done = default_qty
+            return self._before_state__set_quantity(picking, line, message=message)
+        # Otherwise go to select_lot
+        return self._response_for_set_lot(picking, line, message=message)
+
     def _scan_line__assign_user(self, picking, line, qty_done):
         product = line.product_id
         stock = self._actions_for("stock")
@@ -1706,7 +1721,7 @@ def _list_stock_pickings_next_states(self):
         }
 
     def _scan_line_next_states(self):
-        return {"select_move", "set_lot", "set_quantity"}
+        return {"select_move", "set_lot", "set_quantity", "set_destination"}
 
     def _set_lot_next_states(self):
         return {"select_move", "set_lot", "set_quantity"}

shopfloor_reception/tests/__init__.py

@@ -13,3 +13,4 @@
 from . import test_return_scan_line
 from . import test_return_set_quantity
 from . import test_return_reception_done
+from . import test_recover

shopfloor_reception/tests/common.py

@@ -144,6 +144,9 @@ def assertMessage(self, response, expected_message):
         for key, value in expected_message.items():
             self.assertEqual(message.get(key), value)
 
+    def assertNotMessage(self, response, expected_message):
+        self.assertMessage(response, expected_message)
+
     @classmethod
     def _get_move_ids_from_response(cls, response):
         state = response.get("next_state")

shopfloor_reception/tests/test_recover.py

@@ -0,0 +1,176 @@
+# Copyright 2025 Camptocamp SA
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
+
+# Recover happens at line selection.
+# If a line exists for current user
+
+from .common import CommonCase
+
+
+class TestRecover(CommonCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.recover_msg = {
+            "message_type": "info",
+            "body": "Recovered previous session.",
+        }
+
+    def test_recover(self):
+        # here, product isn't tracked by lot, but the move has a move
+        # line already assigned to the user.
+        # No quantity done, we should be redirected to set_quantity,
+        # with the default qty set
+        picking = self._create_picking()
+        # First time we select the line, no recover
+        response = self.service.dispatch(
+            "scan_line",
+            params={"picking_id": picking.id, "barcode": self.product_a.default_code},
+        )
+        selected_move_line = picking.move_line_ids.filtered(
+            lambda li: li.product_id == self.product_a
+        )
+        self.assertEqual(selected_move_line.qty_done, 1)
+        self.assertEqual(selected_move_line.shopfloor_user_id.id, self.env.uid)
+        picking_data = self.data.picking(picking)
+        move_line_data = self.data.move_lines(selected_move_line)
+        self.assert_response(
+            response,
+            next_state="set_quantity",
+            data={
+                "picking": picking_data,
+                "selected_move_line": move_line_data,
+                "confirmation_required": None,
+            },
+        )
+        # Now that there's a shopfloor_user_id, we should recover the session
+        # but since didn't change anything, nothing should change.
+        # Most importantly, the existing move_line should be reused throughout
+        # the whole process
+        response = self.service.dispatch(
+            "scan_line",
+            params={"picking_id": picking.id, "barcode": self.product_a.default_code},
+        )
+        # qty done is the same
+        self.assertEqual(selected_move_line.qty_done, 1)
+        self.assert_response(
+            response,
+            next_state="set_quantity",
+            data={
+                "picking": picking_data,
+                "selected_move_line": move_line_data,
+                "confirmation_required": None,
+            },
+            message=self.recover_msg,
+        )
+        # Set qty_done to 5/10 on the move line, we should recover it
+        selected_move_line.qty_done = 5
+        response = self.service.dispatch(
+            "scan_line",
+            params={"picking_id": picking.id, "barcode": self.product_a.default_code},
+        )
+        self.assertEqual(selected_move_line.qty_done, 5)
+        move_line_data = self.data.move_lines(selected_move_line)
+        self.assert_response(
+            response,
+            next_state="set_quantity",
+            data={
+                "picking": picking_data,
+                "selected_move_line": move_line_data,
+                "confirmation_required": None,
+            },
+            message=self.recover_msg,
+        )
+        # If the goods were put in a pack, we move to set destination
+        response = self.service.dispatch(
+            "process_with_new_pack",
+            params={
+                "picking_id": picking.id,
+                "selected_line_id": selected_move_line.id,
+                "quantity": selected_move_line.qty_done,
+            },
+        )
+        package = selected_move_line.result_package_id
+        self.assertTrue(package)
+        self.assertEqual(selected_move_line.qty_done, 5)
+        self.assertEqual(selected_move_line.reserved_uom_qty, 5)
+        picking_data = self.data.picking(picking)
+        move_line_data = self.data.move_lines(selected_move_line)
+        self.assert_response(
+            response,
+            next_state="set_destination",
+            data={
+                "picking": picking_data,
+                "selected_move_line": move_line_data,
+            },
+        )
+        # Scan the line again, we should end up with the exact same result
+        # with the additionnal recover message
+        response = self.service.dispatch(
+            "scan_line",
+            params={"picking_id": picking.id, "barcode": self.product_a.default_code},
+        )
+        self.assert_response(
+            response,
+            next_state="set_destination",
+            data={
+                "picking": picking_data,
+                "selected_move_line": move_line_data,
+            },
+            message=self.recover_msg,
+        )
+
+    def test_recover_tracking_by_lot(self):
+        # exact same test, just showing that we skip the set lot when recovering
+        picking = self._create_picking()
+        self.product_a.tracking = "lot"
+        # First time we select the line, no recover
+        response = self.service.dispatch(
+            "scan_line",
+            params={"picking_id": picking.id, "barcode": self.product_a.default_code},
+        )
+        selected_move_line = picking.move_line_ids.filtered(
+            lambda li: li.product_id == self.product_a
+        )
+        picking_data = self.data.picking(picking)
+        move_line_data = self.data.move_lines(selected_move_line)
+        self.assert_response(
+            response,
+            next_state="set_lot",
+            data={
+                "picking": picking_data,
+                "selected_move_line": move_line_data,
+            },
+        )
+        # Scan the same line, we end up on the same screen, but with a recover msg
+        response = self.service.dispatch(
+            "scan_line",
+            params={"picking_id": picking.id, "barcode": self.product_a.default_code},
+        )
+        self.assert_response(
+            response,
+            next_state="set_lot",
+            data={
+                "picking": picking_data,
+                "selected_move_line": move_line_data,
+            },
+            message=self.recover_msg,
+        )
+        # Set a lot to the move line, we recover again, but straight to set quantity.
+        selected_move_line.lot_id = self._create_lot()
+        response = self.service.dispatch(
+            "scan_line",
+            params={"picking_id": picking.id, "barcode": self.product_a.default_code},
+        )
+        move_line_data = self.data.move_lines(selected_move_line)
+        self.assert_response(
+            response,
+            next_state="set_quantity",
+            data={
+                "picking": picking_data,
+                "selected_move_line": move_line_data,
+                "confirmation_required": None,
+            },
+            message=self.recover_msg,
+        )
+        # The rest is all the same as test_recover