diff --git a/shopfloor_single_product_transfer/README.rst b/shopfloor_single_product_transfer/README.rst index e74ee3aa40..c82cbc550a 100644 --- a/shopfloor_single_product_transfer/README.rst +++ b/shopfloor_single_product_transfer/README.rst @@ -91,6 +91,7 @@ Contributors - Matthieu Méquignon - Michael Tietz (MT Software) +- Laurent Mignon (ACSONE SA/NV) Design ~~~~~~ diff --git a/shopfloor_single_product_transfer/__manifest__.py b/shopfloor_single_product_transfer/__manifest__.py index bba234fba1..93d92b729f 100644 --- a/shopfloor_single_product_transfer/__manifest__.py +++ b/shopfloor_single_product_transfer/__manifest__.py @@ -1,7 +1,7 @@ { "name": "Shopfloor Single Product Transfer", "summary": "Move an item from one location to another.", - "version": "16.0.1.0.0", + "version": "16.0.2.0.0", "category": "Inventory", "website": "https://github.com/OCA/wms", "author": "Camptocamp, BCIM, Odoo Community Association (OCA)", diff --git a/shopfloor_single_product_transfer/data/shopfloor_scenario_data.xml b/shopfloor_single_product_transfer/data/shopfloor_scenario_data.xml index 01cfdd9585..cbedce7140 100644 --- a/shopfloor_single_product_transfer/data/shopfloor_scenario_data.xml +++ b/shopfloor_single_product_transfer/data/shopfloor_scenario_data.xml @@ -8,6 +8,10 @@ { "allow_create_moves": true, + "allow_get_work": true, + "allow_move_line_search_sort_order": true, + "allow_move_line_search_additional_domain": true, + "scan_location_or_pack_first": true, "allow_unreserve_other_moves": true, "allow_ignore_no_putaway_available": true, "allow_alternative_destination": true, diff --git a/shopfloor_single_product_transfer/migrations/16.0.2.0.0/post-migrate.py b/shopfloor_single_product_transfer/migrations/16.0.2.0.0/post-migrate.py new file mode 100644 index 0000000000..5d9deb2d89 --- /dev/null +++ b/shopfloor_single_product_transfer/migrations/16.0.2.0.0/post-migrate.py @@ -0,0 +1,41 @@ +# Copyright 2026 ACSONE SA/NV +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html). + +import json +import logging + +from odoo import SUPERUSER_ID, api + +_logger = logging.getLogger(__name__) + + +def migrate(cr, version): + _logger.info("Updating scenario options for shopfloor_single_product_transfer") + if not version: + return + env = api.Environment(cr, SUPERUSER_ID, {}) + single_product_transfer_scenario = env.ref( + "shopfloor_single_product_transfer.scenario_single_product_transfer" + ) + _update_scenario_options(single_product_transfer_scenario) + + +def _update_scenario_options(scenario): + options = scenario.options + if "allow_get_work" not in options: + options["allow_get_work"] = True + _logger.info("Option allow_get_work added to scenario %s", scenario.name) + if "allow_move_line_search_sort_order" not in options: + options["allow_move_line_search_sort_order"] = True + options["allow_move_line_search_additional_domain"] = True + _logger.info( + "Option allow_alternative_destination_package added to scenario %s", + scenario.name, + ) + if "scan_location_or_pack_first" not in options: + options["scan_location_or_pack_first"] = True + _logger.info( + "Option scan_location_or_pack_first added to scenario %s", scenario.name + ) + options_edit = json.dumps(options or {}, indent=4, sort_keys=True) + scenario.write({"options_edit": options_edit}) diff --git a/shopfloor_single_product_transfer/readme/CONTRIBUTORS.md b/shopfloor_single_product_transfer/readme/CONTRIBUTORS.md index 8375deed59..415333a0df 100644 --- a/shopfloor_single_product_transfer/readme/CONTRIBUTORS.md +++ b/shopfloor_single_product_transfer/readme/CONTRIBUTORS.md @@ -1,5 +1,6 @@ - Matthieu Méquignon \<\> - Michael Tietz (MT Software) \<\> +- Laurent Mignon (ACSONE SA/NV) \<\> ## Design diff --git a/shopfloor_single_product_transfer/services/single_product_transfer.py b/shopfloor_single_product_transfer/services/single_product_transfer.py index aa676f6358..3bdb93e7f2 100644 --- a/shopfloor_single_product_transfer/services/single_product_transfer.py +++ b/shopfloor_single_product_transfer/services/single_product_transfer.py @@ -69,7 +69,40 @@ class ShopfloorSingleProductTransfer(Component): _usage = "single_product_transfer" _description = __doc__ + _advisory_lock_find_work = "single_product_transfer_find_work" + # Responses + def _response_for_start(self, message=None, popup=None): + """Transition to the 'start' or 'get_work' state + + The switch to 'get_work' is done if the option is enabled on the scenario + """ + if self.work.menu.allow_get_work: + return self._response( + next_state="get_work", data={}, message=message, popup=popup + ) + return self._response_for_select_location_or_package(message=message) + + def _response_for_start_line( + self, + move_line, + message=None, + selected_location_id=None, + selected_package_id=None, + ): + """Transition to the 'start_line' state + + This is used to confirm the processing of a move line + by the user. The user will be requested to select the + product or the package to process the move line. + """ + data = { + "move_line": self.data.move_line(move_line), + "selected_location_id": selected_location_id, + "selected_package_id": selected_package_id, + "scan_location_or_pack_first": self.work.menu.scan_location_or_pack_first, + } + return self._response(next_state="start_line", data=data, message=message) def _response_for_select_location_or_package(self, message=None, popup=None): return self._response( @@ -117,6 +150,16 @@ def _response_for_set_location(self, move_line, package, message=None): } return self._response(next_state="set_location", data=data, message=message) + def _message_for_invalid_product(self, exc): + product = None + if exc.recordset._name == "product.product": + product = exc.recordset[:1] + elif exc.recordset._name == "product.packaging": + product = exc.recordset[:1].product_id + if product: + return self.msg_store.product_not_found_in_current_picking(product) + return self.msg_store.wrong_record(exc.recordset) + # Handlers def _scan_location__quant_domain(self, location): @@ -679,6 +722,10 @@ def _set_quantity__post_move(self, move_line, location, confirmation=None): message = self.msg_store.transfer_done_success(move_line.picking_id) completion_info = self._actions_for("completion.info") completion_info_popup = completion_info.popup(move_line) + if self.work.menu.allow_get_work: + return self._response_for_start( + message=message, popup=completion_info_popup + ) if ( not self.is_allow_move_create() and not self._has_pending_operations_at_same_location(move_line) @@ -686,6 +733,10 @@ def _set_quantity__post_move(self, move_line, location, confirmation=None): return self._response_for_select_location_or_package( message=message, popup=completion_info_popup ) + if self.work.menu.allow_get_work: + return self._response_for_start( + message=message, popup=completion_info_popup + ) return self._response_for_select_product( location=move_line.location_id, package=move_line.package_id, @@ -814,14 +865,290 @@ def _scan_location_or_package__by_location(self, location): return response return self._response_for_select_product(location=location) - # Endpoints + def _recover_previous_session(self): + """When a user starts a transfer, then leaves the session and comes back later, + we want to be able to restore the previous session so they can continue where + they left off. + This method looks for any move line in progress for the user and returns the + corresponding response to restore the session. - def start(self): + :return: A response to restore the previous session, or False if no session to + recover + """ + + response = False move_line = self._find_user_move_line() if move_line: message = self.msg_store.recovered_previous_session() - return self._response_for_set_quantity(move_line, message=message) - return self._response_for_select_location_or_package() + response = self._response_for_set_quantity(move_line, message=message) + return response + + def _scan_line_scan_loc__check_product_tracking( + self, + move_line, + selected_location_id=None, + selected_package_id=None, + ): + product = move_line.product_id + if product.tracking == "lot": + return self._response_for_start_line( + move_line, + message=self.msg_store.scan_lot_on_product_tracked_by_lot(), + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + + def _scan_line__by_location( + self, location, move_line, selected_location_id=None, selected_package_id=None + ): + if location == move_line.location_id: + message = self._check_first_scan_location_or_pack_first( + move_line, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + scanned_location=location, + ) + if message: + return message + response = self._scan_line_scan_loc__check_product_tracking( + move_line, + selected_location_id=location.id, + selected_package_id=selected_package_id, + ) + if response: + return response + return self._response_for_set_quantity(move_line) + + def _scan_line__by_package( + self, package, move_line, selected_location_id=None, selected_package_id=None + ): + if move_line.package_id == package: + message = self._check_first_scan_location_or_pack_first( + move_line, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + scanned_package=package, + ) + if message: + return message + return self._response_for_set_quantity(move_line) + + def _scan_line__by_product( + self, product, move_line, selected_location_id=None, selected_package_id=None + ): + if product == move_line.product_id: + message = self._check_first_scan_location_or_pack_first( + move_line, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + if message: + return message + + response = self._scan_line_scan_loc__check_product_tracking( + move_line, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + if response: + return response + else: + return self._response_for_set_quantity(move_line) + + def _scan_line__by_packaging( + self, packaging, move_line, selected_location_id=None, selected_package_id=None + ): + response = self._scan_line_scan_loc__check_product_tracking( + move_line, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + if response: + return response + return self._scan_line__by_product( + packaging.product_id, + move_line, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + + def _scan_line__by_lot( + self, lot, move_line, selected_location_id=None, selected_package_id=None + ): + if lot == move_line.lot_id: + message = self._check_first_scan_location_or_pack_first( + move_line, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + if message: + return message + return self._response_for_set_quantity(move_line) + + def _scan_line__fallback( + self, record, move_line, selected_location_id=None, selected_package_id=None + ): + # Nothing matches what is expected from the move line. + if record: + return self._response_for_start_line( + move_line, + message=self.msg_store.wrong_record(record), + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + return self._response_for_start_line( + move_line, + message=self.msg_store.barcode_not_found(), + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + + def _check_first_scan_location_or_pack_first( + self, + move_line, + selected_location_id=None, + selected_package_id=None, + scanned_location=None, + scanned_package=None, + ): + """Restrict scanning product or lot first with option on. + + When the option first scan location or pack first is on. + When the line being worked on has a package, ask to scan the package first, + otherwise ask to scan the location first. + """ + if not self.work.menu.scan_location_or_pack_first: + return None + message = None + if move_line.package_id: + if not selected_package_id and not scanned_package: + message = self.msg_store.line_has_package_scan_package() + elif not selected_location_id and not scanned_location: + message = self.msg_store.scan_the_location_first() + if message: + return self._response_for_start_line( + move_line, + message=message, + selected_location_id=selected_location_id or scanned_location.id + if scanned_location + else None, + selected_package_id=selected_package_id or scanned_package.id + if scanned_package + else None, + ) + return None + + def _try_select_move_line(self, move_line): + """Check if the move line can be worked on by the user. + + This is a method hookable to apply specific rules on which move lines can be + selected/skipped when looking for the next move line to work on. + + By default, it checks if the move line has no putaway available when the option + 'ignore_no_putaway_available' is enabled, and if so, it will skip the move line. + """ + if self.work.menu.ignore_no_putaway_available and self._actions_for( + "stock" + ).no_putaway_available(self.picking_types, move_line): + return None + return move_line + + def _get_next_move_line_to_work(self): + """Get the next move line to work on for the user.""" + move_lines = self.search_move_line.search_move_lines(match_user=True) + for line in move_lines: + if line := self._try_select_move_line(line): + return line + return None + + # Endpoints + + def start(self): + response = self._recover_previous_session() + return response or self._response_for_start() + + def find_work(self): + """Find the new location to work from, for a user. + + First recover any started pickings. + The find the first move line from the oldest transfer that can be worked on. + Mark the first move lines as picked. + And ask the user to confirm. + + Transitions: + * start: no work found + * start: a move line has been found but no putaway location is available + * select_line: a move line has been found and marked as picked, + ask the user to confirm + """ + response = self._recover_previous_session() + if response: + return response + self._actions_for("lock").advisory(self._advisory_lock_find_work) + move_line = self._get_next_move_line_to_work() + if not move_line: + return self._response_for_start(message=self.msg_store.no_work_found()) + stock = self._actions_for("stock") + if ( + not self.work.menu.ignore_no_putaway_available + and stock.no_putaway_available(self.picking_types, move_line) + ): + message = self.msg_store.no_putaway_destination_available() + return self._response_for_start(message=message) + if self.work.menu.no_prefill_qty: + stock.mark_move_line_as_picked(move_line, quantity=0) + else: + stock.mark_move_line_as_picked(move_line) + return self._response_for_start_line(move_line) + + def confirm_start_line( + self, + selected_line_id, + barcode, + selected_location_id=None, + selected_package_id=None, + ): + """Validate the selected line by scanning the location, product, lot + or package.""" + move_line = self.env["stock.move.line"].browse(selected_line_id) + if not move_line.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + + search = self._actions_for("search").for_products(move_line.product_id) + handlers = { + "location": self._scan_line__by_location, + "package": self._scan_line__by_package, + "lot": self._scan_line__by_lot, + "product": self._scan_line__by_product, + "packaging": self._scan_line__by_packaging, + "none": self._scan_line__fallback, + } + try: + search_result = search.find( + barcode, + types=handlers.keys(), + ) + except SearchInvalidProduct as e: + message = self._message_for_invalid_product(e) + return self._response_for_start_line( + move_line, + message=message, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + handler = handlers.get(search_result.type, self._scan_line__fallback) + response = handler( + search_result.record, + move_line, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) + return response or self._scan_line__fallback( + search_result.record, + move_line, + selected_location_id=selected_location_id, + selected_package_id=selected_package_id, + ) def scan_location_or_package(self, barcode): """Scan a source location or a source package. @@ -892,15 +1219,7 @@ def scan_product(self, barcode, location_id=None, package_id=None): package=package, ) except SearchInvalidProduct as e: - product = None - if e.recordset._name == "product.product": - product = e.recordset[:1] - elif e.recordset._name == "product.packaging": - product = e.recordset[:1].product_id - if product: - message = self.msg_store.product_not_found_in_current_picking(product) - else: - message = self.msg_store.wrong_record(e.recordset) + message = self._message_for_invalid_product(e) else: message = self.msg_store.barcode_not_found() return self._response_for_select_product( @@ -908,7 +1227,7 @@ def scan_product(self, barcode, location_id=None, package_id=None): ) def scan_product__action_cancel(self): - return self._response_for_select_location_or_package() + return self._response_for_start() def set_quantity(self, selected_line_id, barcode, quantity, confirmation=None): """Sets quantity done if a product is scanned, @@ -937,13 +1256,21 @@ def set_quantity(self, selected_line_id, barcode, quantity, confirmation=None): # Puts the product in a new or an existing pack "package": self._set_quantity__by_package, } - search = self._actions_for("search") - search_result = search.find(barcode, types=handlers_by_type.keys()) - handler = handlers_by_type.get(search_result.type) - if handler: - confirmed = confirmation == barcode - return handler(move_line, search_result.record, confirmation=confirmed) - message = self.msg_store.barcode_not_found() + search = self._actions_for("search").for_products(move_line.product_id) + try: + search_result = search.find( + barcode, + types=handlers_by_type.keys(), + ) + handler = handlers_by_type.get(search_result.type) + if handler: + confirmed = confirmation == barcode + return handler(move_line, search_result.record, confirmation=confirmed) + + except SearchInvalidProduct as e: + message = self._message_for_invalid_product(e) + else: + message = self.msg_store.barcode_not_found() return self._response_for_set_quantity(move_line, message=message) def set_quantity__action_cancel(self, selected_line_id): @@ -954,7 +1281,7 @@ def set_quantity__action_cancel(self, selected_line_id): else: stock = self._actions_for("stock") stock.unmark_move_line_as_picked(move_line) - return self._response_for_select_location_or_package() + return self._response_for_start() def set_location(self, selected_line_id, package_id, barcode): """Sets the destination location @@ -983,6 +1310,9 @@ class ShopfloorSingleProductTransferValidator(Component): def start(self): return {} + def get_work(self): + return {} + def scan_location_or_package(self): return {"barcode": {"required": True, "type": "string"}} @@ -1016,6 +1346,22 @@ def set_location(self): "barcode": {"required": True, "type": "string"}, } + def confirm_start_line(self): + return { + "selected_line_id": {"coerce": to_int, "required": True, "type": "integer"}, + "barcode": {"required": True, "type": "string"}, + "selected_location_id": { + "coerce": to_int, + "required": False, + "type": "integer", + }, + "selected_package_id": { + "coerce": to_int, + "required": False, + "type": "integer", + }, + } + class ShopfloorSingleProductTransferValidatorResponse(Component): _inherit = "base.shopfloor.validator.response" @@ -1030,6 +1376,8 @@ def _states(self): "select_product": self._schema_select_product, "set_quantity": self._schema_set_quantity, "set_location": self._schema_set_location, + "start_line": self._schema_start_line, + "get_work": {}, } def start(self): @@ -1054,11 +1402,17 @@ def set_quantity__action_cancel(self): next_states=self._set_quantity__action_cancel_next_states() ) + def confirm_start_line(self): + return self._response_schema(next_states=self._confirm_start_line_next_states()) + def set_location(self): return self._response_schema(next_states=self._set_location_next_states()) + def find_work(self): + return self._response_schema(next_states=self._find_work_next_states()) + def _start_next_states(self): - return {"select_location_or_package", "set_quantity"} + return {"select_location_or_package", "set_quantity", "get_work"} def _scan_location_next_states(self): return {"select_location_or_package", "select_product"} @@ -1067,17 +1421,23 @@ def _scan_product_next_states(self): return {"select_product", "set_quantity"} def _scan_product__action_cancel_next_states(self): - return {"select_location_or_package"} + return {"select_location_or_package", "get_work"} def _set_quantity_next_states(self): - return {"set_quantity", "select_product", "set_location"} + return {"set_quantity", "select_product", "set_location", "get_work"} def _set_quantity__action_cancel_next_states(self): - return {"select_location_or_package"} + return {"select_location_or_package", "get_work"} def _set_location_next_states(self): return {"set_quantity", "select_product", "set_location"} + def _find_work_next_states(self): + return {"start_line", "get_work"} + + def _confirm_start_line_next_states(self): + return {"start_line", "set_quantity", "get_work"} + @property def _schema_select_location_or_package(self): return {} @@ -1110,3 +1470,16 @@ def _schema_set_location(self): "move_line": {"type": "dict", "schema": self.schemas.move_line()}, "package": {"type": "dict", "schema": self.schemas.package()}, } + + @property + def _schema_start_line(self): + return { + "move_line": {"type": "dict", "schema": self.schemas.move_line()}, + "selected_location_id": {"type": "integer", "nullable": True}, + "selected_package_id": {"type": "integer", "nullable": True}, + "scan_location_or_pack_first": { + "type": "boolean", + "nullable": False, + "required": False, + }, + } diff --git a/shopfloor_single_product_transfer/tests/__init__.py b/shopfloor_single_product_transfer/tests/__init__.py index 7ec1261ca0..eb7eb1e237 100644 --- a/shopfloor_single_product_transfer/tests/__init__.py +++ b/shopfloor_single_product_transfer/tests/__init__.py @@ -1,3 +1,4 @@ +from . import test_find_work from . import test_start from . import test_scan_location_or_package from . import test_scan_product diff --git a/shopfloor_single_product_transfer/tests/test_find_work.py b/shopfloor_single_product_transfer/tests/test_find_work.py new file mode 100644 index 0000000000..a29bdb4498 --- /dev/null +++ b/shopfloor_single_product_transfer/tests/test_find_work.py @@ -0,0 +1,617 @@ +# Copyright 2026 ACSONE SA/NV (https://www.acsone.eu) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +from odoo import fields + +from .common import CommonCase + + +class TestFindWork(CommonCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.menu.sudo().allow_get_work = True + cls.location_src_a = ( + cls.env["stock.location"] + .sudo() + .create( + { + "name": "Source A", + "location_id": cls.location_src.id, + } + ) + ) + cls.location_src_b = ( + cls.env["stock.location"] + .sudo() + .create( + { + "name": "Source B", + "location_id": cls.location_src.id, + } + ) + ) + cls.product = cls.product_a + cls._add_stock_to_product(cls.product_a, cls.location_src_a, 10) + cls._add_stock_to_product(cls.product_b, cls.location_src_b, 10) + cls.picking_1 = cls._create_picking(lines=[(cls.product_a, 10)]) + cls.picking_2 = cls._create_picking(lines=[(cls.product_b, 10)]) + # Simulate putaway rules having run so that no_putaway_available returns + # False for the class-level pickings. Without this, find_work would + # return no_putaway_destination_available for every test. + cls.picking_1.move_line_ids.sudo().location_dest_id = cls.dispatch_location.id + cls.picking_2.move_line_ids.sudo().location_dest_id = cls.dispatch_location.id + + def _data_for_start_line( + self, move_line, selected_location_id=None, selected_package_id=None + ): + return { + "move_line": self._data_for_move_line(move_line), + "selected_location_id": selected_location_id, + "selected_package_id": selected_package_id, + "scan_location_or_pack_first": self.menu.scan_location_or_pack_first, + } + + def _setup_lot_move_line(self, location=None): + location = location or self.location_src + self._set_product_tracking_by_lot(self.product_a) + lot = self._create_lot_for_product(self.product_a, "LOT001") + self._add_stock_to_product(self.product_a, location, 5, lot=lot) + picking = self._create_picking(lines=[(self.product_a, 5)]) + move_line = fields.first(picking.move_line_ids) + return move_line, lot + + def _assert_start_line_lot_required( + self, response, move_line, selected_location_id=None + ): + self.assert_response( + response, + next_state="start_line", + data=self._data_for_start_line( + move_line, selected_location_id=selected_location_id + ), + message=self.msg_store.scan_lot_on_product_tracked_by_lot(), + ) + + def _assert_set_quantity(self, response, move_line): + self.assert_response( + response, + next_state="set_quantity", + data={ + "move_line": self._data_for_move_line(move_line), + "asking_confirmation": None, + }, + ) + + def test_find_work(self): + response = self.service.dispatch("find_work") + data = self._data_for_start_line(fields.first(self.picking_1.move_line_ids)) + self.assert_response( + response, + next_state="start_line", + data=data, + ) + + # cancel select product to go back to find_work + response = self.service.dispatch("scan_product__action_cancel") + self.assert_response( + response, + next_state="get_work", + ) + + # cancel the first picking + self.picking_1.action_cancel() + response = self.service.dispatch("find_work") + data = self._data_for_start_line(fields.first(self.picking_2.move_line_ids)) + self.assert_response( + response, + next_state="start_line", + data=data, + ) + + def test_find_work_with_prefilled_quantity(self): + self.menu.sudo().no_prefill_qty = False + response = self.service.dispatch("find_work") + move_line = fields.first(self.picking_1.move_line_ids) + data = self._data_for_start_line(move_line) + self.assert_response( + response, + next_state="start_line", + data=data, + ) + self.assertEqual(move_line.qty_done, move_line.reserved_uom_qty) + + def test_find_work_with_not_prefilled_quantity(self): + self._enable_no_prefill_qty() + response = self.service.dispatch("find_work") + move_line = fields.first(self.picking_1.move_line_ids) + data = self._data_for_start_line(move_line) + self.assert_response( + response, + next_state="start_line", + data=data, + ) + self.assertEqual(move_line.qty_done, 0) + + def test_confirm_start_line_line_not_found(self): + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": 0, "barcode": "whatever"}, + ) + self.assert_response( + response, + next_state="get_work", + message=self.msg_store.record_not_found(), + ) + + def test_confirm_start_line_barcode_not_found(self): + move_line = fields.first(self.picking_1.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": move_line.id, "barcode": "NOPE"}, + ) + data = self._data_for_start_line(move_line) + self.assert_response( + response, + next_state="start_line", + data=data, + message=self.msg_store.barcode_not_found(), + ) + + def test_confirm_start_line_scan_product(self): + move_line = fields.first(self.picking_1.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a.barcode, + }, + ) + self._assert_set_quantity(response, move_line) + + def test_confirm_start_line_scan_wrong_product(self): + move_line = fields.first(self.picking_1.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_b.barcode, + }, + ) + data = self._data_for_start_line(move_line) + self.assert_response( + response, + next_state="start_line", + data=data, + message=self.msg_store.product_not_found_in_current_picking(self.product_b), + ) + + def test_confirm_start_line_scan_product_tracked_by_lot(self): + self._set_product_tracking_by_lot(self.product_a) + lot = self._create_lot_for_product(self.product_a, "LOT001") + self._add_stock_to_product(self.product_a, self.location_src, 5, lot=lot) + picking = self._create_picking(lines=[(self.product_a, 5)]) + move_line = fields.first(picking.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a.barcode, + }, + ) + self._assert_start_line_lot_required(response, move_line) + + def test_confirm_start_line_scan_packaging(self): + move_line = fields.first(self.picking_1.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a_packaging.barcode, + }, + ) + self._assert_set_quantity(response, move_line) + + def test_confirm_start_line_scan_lot(self): + self._set_product_tracking_by_lot(self.product_a) + lot = self._create_lot_for_product(self.product_a, "LOT001") + self._add_stock_to_product(self.product_a, self.location_src, 5, lot=lot) + picking = self._create_picking(lines=[(self.product_a, 5)]) + move_line = fields.first(picking.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": move_line.id, "barcode": lot.name}, + ) + self._assert_set_quantity(response, move_line) + + def test_confirm_start_line_scan_wrong_lot(self): + self._set_product_tracking_by_lot(self.product_a) + lot = self._create_lot_for_product(self.product_a, "LOT001") + wrong_lot = self._create_lot_for_product(self.product_a, "LOT_WRONG") + self._add_stock_to_product(self.product_a, self.location_src, 5, lot=lot) + picking = self._create_picking(lines=[(self.product_a, 5)]) + move_line = fields.first(picking.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": move_line.id, "barcode": wrong_lot.name}, + ) + data = self._data_for_start_line(move_line) + self.assert_response( + response, + next_state="start_line", + data=data, + message=self.msg_store.wrong_record(wrong_lot), + ) + + def test_confirm_start_line_scan_package(self): + package = self._create_empty_package("PKG001") + self._add_stock_to_product( + self.product_a, self.location_src_a, 5, package=package + ) + picking = self._create_picking(lines=[(self.product_a, 5)]) + move_line = fields.first(picking.move_line_ids) + self.assertEqual(move_line.package_id, package) + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": move_line.id, "barcode": package.name}, + ) + self._assert_set_quantity(response, move_line) + + def test_confirm_start_line_scan_wrong_package(self): + package = self._create_empty_package("PKG001") + wrong_package = self._create_empty_package("PKG_WRONG") + self._add_stock_to_product( + self.product_a, self.location_src_a, 5, package=package + ) + picking = self._create_picking(lines=[(self.product_a, 5)]) + move_line = fields.first(picking.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": move_line.id, "barcode": wrong_package.name}, + ) + data = self._data_for_start_line(move_line) + self.assert_response( + response, + next_state="start_line", + data=data, + message=self.msg_store.wrong_record(wrong_package), + ) + + def _enable_scan_location_or_pack_first(self): + self.menu.sudo().scan_location_or_pack_first = True + + def _setup_packaged_move_line(self): + package = self._create_empty_package("PKG001") + self._add_stock_to_product( + self.product_a, self.child_location, 5, package=package + ) + picking = self._create_picking(lines=[(self.product_a, 5)]) + move_line = fields.first(picking.move_line_ids) + return move_line, package + + def test_confirm_start_line_slpf_scan_product_requires_location(self): + self._enable_scan_location_or_pack_first() + move_line = fields.first(self.picking_1.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a.barcode, + }, + ) + data = self._data_for_start_line(move_line) + self.assert_response( + response, + next_state="start_line", + data=data, + message=self.msg_store.scan_the_location_first(), + ) + + def test_confirm_start_line_scan_slpf_scan_location(self): + self._enable_scan_location_or_pack_first() + self._add_stock_to_product(self.product_a, self.child_location, 5) + picking = self._create_picking(lines=[(self.product_a, 5)]) + move_line = fields.first(picking.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.child_location.barcode, + }, + ) + data = { + "move_line": self._data_for_move_line(move_line), + "asking_confirmation": None, + } + self.assert_response(response, next_state="set_quantity", data=data) + + def test_confirm_start_line_scan_slpf_lot_tracked_scan_location(self): + # With scan_location_or_pack_first, scanning the location on a + # lot-tracked line is sufficient to confirm: goes to set_quantity + # directly, bypassing the lot scan step. + self._enable_scan_location_or_pack_first() + self._set_product_tracking_by_lot(self.product_a) + lot = self._create_lot_for_product(self.product_a, "LOT001") + self._add_stock_to_product(self.product_a, self.child_location, 5, lot=lot) + picking = self._create_picking(lines=[(self.product_a, 5)]) + move_line = fields.first(picking.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.child_location.barcode, + }, + ) + self._assert_start_line_lot_required( + response, move_line, selected_location_id=self.child_location.id + ) + + def test_confirm_start_line_scan_slpf_scan_product_with_location( + self, + ): + self._enable_scan_location_or_pack_first() + move_line = fields.first(self.picking_1.move_line_ids) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a.barcode, + "selected_location_id": move_line.location_id.id, + }, + ) + self._assert_set_quantity(response, move_line) + + def test_confirm_start_line_slpf_package_scan_product_requires_package(self): + self._enable_scan_location_or_pack_first() + move_line, _package = self._setup_packaged_move_line() + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a.barcode, + }, + ) + data = self._data_for_start_line(move_line) + self.assert_response( + response, + next_state="start_line", + data=data, + message=self.msg_store.line_has_package_scan_package(), + ) + + def test_confirm_start_line_slpf_package_scan_location_requires_package(self): + self._enable_scan_location_or_pack_first() + move_line, _package = self._setup_packaged_move_line() + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.child_location.barcode, + }, + ) + data = self._data_for_start_line( + move_line, selected_location_id=self.child_location.id + ) + self.assert_response( + response, + next_state="start_line", + data=data, + message=self.msg_store.line_has_package_scan_package(), + ) + + def test_confirm_start_line_scan_slpf_package_scan_package( + self, + ): + self._enable_scan_location_or_pack_first() + move_line, package = self._setup_packaged_move_line() + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": move_line.id, "barcode": package.name}, + ) + self._assert_set_quantity(response, move_line) + + # ------------------------------------------------------------------------- + # Lot tracking: all paths that require a lot must stay on start_line, + # and scanning the lot must reach set_quantity. + # ------------------------------------------------------------------------- + + # -- Without scan_location_or_pack_first -- + + def test_confirm_start_line_lot_tracked_scan_location_requires_lot(self): + move_line, _lot = self._setup_lot_move_line(self.child_location) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.child_location.barcode, + }, + ) + self._assert_start_line_lot_required( + response, move_line, selected_location_id=self.child_location.id + ) + + def test_confirm_start_line_lot_tracked_scan_packaging_requires_lot(self): + move_line, _lot = self._setup_lot_move_line() + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a_packaging.barcode, + }, + ) + self._assert_start_line_lot_required(response, move_line) + + def test_confirm_start_line_lot_tracked_scan_product_then_lot(self): + move_line, lot = self._setup_lot_move_line() + # Step 1: product scan -> lot required + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a.barcode, + }, + ) + self._assert_start_line_lot_required(response, move_line) + # Step 2: lot scan -> set_quantity + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": move_line.id, "barcode": lot.name}, + ) + self._assert_set_quantity(response, move_line) + + def test_confirm_start_line_lot_tracked_scan_location_then_lot(self): + move_line, lot = self._setup_lot_move_line(self.child_location) + # Step 1: location scan -> lot required + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.child_location.barcode, + }, + ) + self._assert_start_line_lot_required( + response, move_line, selected_location_id=self.child_location.id + ) + # Step 2: lot scan -> set_quantity (no slpf, no location check needed) + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": move_line.id, "barcode": lot.name}, + ) + self._assert_set_quantity(response, move_line) + + # -- With scan_location_or_pack_first -- + + def test_confirm_start_line_slpf_lot_tracked_scan_lot_no_location(self): + self._enable_scan_location_or_pack_first() + move_line, lot = self._setup_lot_move_line() + response = self.service.dispatch( + "confirm_start_line", + params={"selected_line_id": move_line.id, "barcode": lot.name}, + ) + self.assert_response( + response, + next_state="start_line", + data=self._data_for_start_line(move_line), + message=self.msg_store.scan_the_location_first(), + ) + + def test_confirm_start_line_slpf_lot_tracked_product_with_location_requires_lot( + self, + ): + self._enable_scan_location_or_pack_first() + move_line, _lot = self._setup_lot_move_line() + location_id = move_line.location_id.id + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a.barcode, + "selected_location_id": location_id, + }, + ) + # slpf check passes (location provided); lot still required + self._assert_start_line_lot_required( + response, move_line, selected_location_id=location_id + ) + + def test_confirm_start_line_slpf_lot_tracked_product_with_location_then_lot(self): + self._enable_scan_location_or_pack_first() + move_line, lot = self._setup_lot_move_line() + location_id = move_line.location_id.id + # Step 1: product + location -> lot required (location preserved in response) + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.product_a.barcode, + "selected_location_id": location_id, + }, + ) + self._assert_start_line_lot_required( + response, move_line, selected_location_id=location_id + ) + # Step 2: lot + location -> set_quantity + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": lot.name, + "selected_location_id": location_id, + }, + ) + self._assert_set_quantity(response, move_line) + + def test_confirm_start_line_slpf_lot_tracked_scan_location_then_lot(self): + self._enable_scan_location_or_pack_first() + move_line, lot = self._setup_lot_move_line(self.child_location) + # Step 1: location scan -> lot required + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": self.child_location.barcode, + }, + ) + self._assert_start_line_lot_required( + response, move_line, selected_location_id=self.child_location.id + ) + # Step 2: lot + location (frontend passes the confirmed location) + # -> set_quantity + response = self.service.dispatch( + "confirm_start_line", + params={ + "selected_line_id": move_line.id, + "barcode": lot.name, + "selected_location_id": self.child_location.id, + }, + ) + self._assert_set_quantity(response, move_line) + + # ------------------------------------------------------------------------- + # ignore_no_putaway_available flag behaviour in find_work + # ------------------------------------------------------------------------- + + def test_find_work_no_putaway_destination(self): + # With ignore_no_putaway_available=False (default), find_work returns + # an error and stays at get_work when the candidate line has no + # putaway destination (location_dest_id == picking type default). + self.picking_1.action_cancel() + self.picking_2.action_cancel() + self._add_stock_to_product(self.product_a, self.location_src_a, 3) + self._create_picking(lines=[(self.product_a, 3)]) + response = self.service.dispatch("find_work") + self.assert_response( + response, + next_state="get_work", + message=self.msg_store.no_putaway_destination_available(), + ) + + def test_find_work_ignore_no_putaway_skips_to_next(self): + # With ignore_no_putaway_available=True, lines without a specific + # putaway destination are skipped; the next eligible line is returned. + self._enable_ignore_no_putaway_available() + default_dest = self.picking_1.picking_type_id.default_location_dest_id + self.picking_1.move_line_ids.sudo().location_dest_id = default_dest.id + # picking_2 still has dispatch_location as destination (set in setUpClass) + response = self.service.dispatch("find_work") + move_line = fields.first(self.picking_2.move_line_ids) + self.assert_response( + response, + next_state="start_line", + data=self._data_for_start_line(move_line), + ) + + def test_find_work_ignore_no_putaway_no_work_found(self): + # With ignore_no_putaway_available=True, if every candidate line has no + # putaway destination, find_work returns no_work_found. + self._enable_ignore_no_putaway_available() + self.picking_1.action_cancel() + self.picking_2.action_cancel() + self._add_stock_to_product(self.product_a, self.location_src_a, 3) + self._create_picking(lines=[(self.product_a, 3)]) + response = self.service.dispatch("find_work") + self.assert_response( + response, + next_state="get_work", + message=self.msg_store.no_work_found(), + ) diff --git a/shopfloor_single_product_transfer/tests/test_scan_product.py b/shopfloor_single_product_transfer/tests/test_scan_product.py index 64c5a05516..22a8a8ddde 100644 --- a/shopfloor_single_product_transfer/tests/test_scan_product.py +++ b/shopfloor_single_product_transfer/tests/test_scan_product.py @@ -526,6 +526,11 @@ def test_action_cancel(self): response = self.service.dispatch("scan_product__action_cancel") self.assert_response(response, next_state="select_location_or_package", data={}) + def test_action_cancel_with_get_work(self): + self.menu.sudo().allow_get_work = True + response = self.service.dispatch("scan_product__action_cancel") + self.assert_response(response, next_state="get_work", data={}) + def test_scan_product_packaging(self): location = self.location_src packaging = self.product_a_packaging diff --git a/shopfloor_single_product_transfer/tests/test_set_quantity.py b/shopfloor_single_product_transfer/tests/test_set_quantity.py index 4d498c0b2f..f030ffc941 100644 --- a/shopfloor_single_product_transfer/tests/test_set_quantity.py +++ b/shopfloor_single_product_transfer/tests/test_set_quantity.py @@ -357,6 +357,36 @@ def test_set_quantity_scan_lot_prefill_qty_enabled(self): response, next_state="set_quantity", message=expected_message, data=data ) + def test_set_quantity_scan_lot_not_unique(self): + """Even if the lot is not unique, we should be able to process the line by + scanning the lot, if the scanned lot is the one on the move line.""" + self._set_product_tracking_by_lot(self.product_b) + duplicate_lot = self._create_lot_for_product(self.product_b, "LOT_BARCODE") + self._add_stock_to_product(self.product_b, self.location, 10, lot=duplicate_lot) + # First, select a picking + self._set_product_tracking_by_lot(self.product) + lot = self._create_lot_for_product(self.product, duplicate_lot.name) + self._add_stock_to_product(self.product, self.location, 5, lot=lot) + picking = self._setup_picking(lot=lot) + move_line = picking.move_line_ids + self.service.dispatch( + "scan_product", + params={"location_id": self.location.id, "barcode": lot.name}, + ) + response = self.service.dispatch( + "set_quantity", + params={ + "selected_line_id": move_line.id, + "quantity": 1, + "barcode": lot.name, + }, + ) + data = { + "move_line": self._data_for_move_line(move_line), + "asking_confirmation": None, + } + self.assert_response(response, next_state="set_quantity", data=data) + def test_set_quantity_scan_packaging(self): """Scan a packaging to process an existing line.""" # First, select a picking @@ -689,6 +719,30 @@ def test_action_cancel(self): # Ensure the picking is not cancelled if allow_move_create is not enabled self.assertTrue(move_line.picking_id.state == "assigned") + def test_action_cancel_with_get_work(self): + self.menu.sudo().allow_get_work = True + picking = self._setup_picking() + self.service.dispatch( + "scan_product", + params={ + "location_id": self.location.id, + "barcode": self.product.barcode, + }, + ) + move_line = picking.move_line_ids + move_line.qty_done = 10.0 + response = self.service.dispatch( + "set_quantity__action_cancel", + params={"selected_line_id": move_line.id}, + ) + data = {} + self.assert_response(response, next_state="get_work", data=data) + # Ensure qty_picked and user has been reset. + self.assertFalse(move_line.picking_id.user_id) + self.assertEqual(move_line.qty_done, 0.0) + # Ensure the picking is not cancelled if allow_move_create is not enabled + self.assertTrue(move_line.picking_id.state == "assigned") + def test_action_cancel_allow_move_create(self): # We perform the same actions as in test_action_cancel, # but with the allow_move_create option enabled @@ -790,6 +844,33 @@ def test_set_quantity_scan_location(self): self.picking_type.default_location_src_id, ) + def test_set_quantity_scan_location_with_get_work(self): + self.menu.sudo().allow_get_work = True + picking = self._setup_picking() + location = self.location + self.service.dispatch( + "scan_product", + params={"location_id": location.id, "barcode": self.product.barcode}, + ) + move_line = picking.move_line_ids + response = self.service.dispatch( + "set_quantity", + params={ + "selected_line_id": move_line.id, + "quantity": 6, + "barcode": self.dispatch_location.name, + }, + ) + completion_info = self.service._actions_for("completion.info") + expected_popup = completion_info.popup(move_line) + expected_message = self.msg_store.transfer_done_success(move_line.picking_id) + self.assert_response( + response, + next_state="get_work", + message=expected_message, + popup=expected_popup, + ) + def test_set_quantity_scan_location_allow_move_create(self): self.menu.sudo().allow_move_create = True picking = self._setup_picking() diff --git a/shopfloor_single_product_transfer/tests/test_start.py b/shopfloor_single_product_transfer/tests/test_start.py index b3e33547da..63b4b1469a 100644 --- a/shopfloor_single_product_transfer/tests/test_start.py +++ b/shopfloor_single_product_transfer/tests/test_start.py @@ -9,7 +9,13 @@ def test_start(self): response = self.service.dispatch("start") self.assert_response(response, next_state="select_location_or_package", data={}) - def test_recover(self): + def test_start_with_work(self): + self.menu.sudo().allow_get_work = True + response = self.service.dispatch("start") + + self.assert_response(response, next_state="get_work") + + def _check_recover(self): product = self.product_a location = self.location_src self._add_stock_to_product(product, location, 10) @@ -30,3 +36,10 @@ def test_recover(self): self.assert_response( response, next_state="set_quantity", data=data, message=message ) + + def test_recover(self): + self._check_recover() + + def test_recover_with_work(self): + self.menu.sudo().allow_get_work = True + self._check_recover()