diff --git a/shopfloor/actions/__init__.py b/shopfloor/actions/__init__.py index ac8c5da74d..42f45d7eee 100644 --- a/shopfloor/actions/__init__.py +++ b/shopfloor/actions/__init__.py @@ -20,6 +20,7 @@ from . import data from . import data_detail from . import completion_info +from . import location_content_transfer_sorter from . import message from . import search from . import inventory diff --git a/shopfloor/actions/data.py b/shopfloor/actions/data.py index 326e19ff91..09ead326eb 100644 --- a/shopfloor/actions/data.py +++ b/shopfloor/actions/data.py @@ -148,6 +148,35 @@ def _move_line_parser(self): ("location_dest_id:location_dest", self._location_parser), ] + def package_level(self, record, **kw): + return self._jsonify(record, self._package_level_parser) + + def package_levels(self, records, **kw): + return [self.package_level(rec, **kw) for rec in records] + + @property + def _package_level_parser(self): + return [ + "id", + "is_done", + ("package_id:package_src", self._package_parser), + ("location_dest_id:location_dest", self._location_parser), + ( + "location_id:location_src", + lambda rec, fname: self.location(rec.picking_id.location_id), + ), + # tnx to stock_quant_package_product_packaging + ( + "package_id:product", + lambda rec, fname: self.product(rec.package_id.single_product_id), + ), + # TODO: allow to pass mapped path to base_jsonify + ( + "package_id:quantity", + lambda rec, fname: rec.package_id.single_product_qty, + ), + ] + def product(self, record, **kw): return self._jsonify(record, self._product_parser, **kw) diff --git a/shopfloor/actions/location_content_transfer_sorter.py b/shopfloor/actions/location_content_transfer_sorter.py new file mode 100644 index 0000000000..154dedb281 --- /dev/null +++ b/shopfloor/actions/location_content_transfer_sorter.py @@ -0,0 +1,60 @@ +from odoo.addons.component.core import Component + + +class LocationContentTransferSorter(Component): + + _name = "shopfloor.location.content.transfer.sorter" + _inherit = "shopfloor.process.action" + _usage = "location_content_transfer.sorter" + + def __init__(self, work_context): + super().__init__(work_context) + self._pickings = self.env["stock.picking"].browse() + self._content = None + + def feed_pickings(self, pickings): + self._pickings |= pickings + + def move_lines(self): + return self._pickings.move_line_ids.filtered( + # lines without package level only (raw products) + lambda line: not line.package_level_id + and line.state not in ("cancel", "done") + ) + + def package_levels(self): + return self._pickings.package_level_ids.filtered( + lambda level: level.state not in ("cancel", "done") + ) + + @staticmethod + def _sort_key(content): + # content can be either a move line, either a package + # level + return ( + # postponed content after other contents + int(content.shopfloor_postponed), + # sort by shopfloor picking sequence + content.location_dest_id.shopfloor_picking_sequence, + # sort by similar destination + content.location_dest_id.complete_name, + # lines before packages (if we have raw products and packages, raw + # will be on top? wild guess) + 0 if content._name == "stock.move.line" else 1, + # to have a deterministic sort + content.id, + ) + + def sort(self): + content = [line for line in self.move_lines()] + [ + level for level in self.package_levels() + ] + self._content = sorted(content, key=self._sort_key) + + def __iter__(self): + if self._content is None: + self.sort() + return iter(self._content) + + def __next__(self): + return next(iter(self)) diff --git a/shopfloor/actions/message.py b/shopfloor/actions/message.py index 143bb83e7b..300a1af679 100644 --- a/shopfloor/actions/message.py +++ b/shopfloor/actions/message.py @@ -299,6 +299,20 @@ def transfer_complete(self, picking): "body": _("Transfer {} complete").format(picking.name), } + def location_content_transfer_item_complete(self, location_dest): + return { + "message_type": "success", + "body": _("Content transfer to {} completed").format(location_dest.name), + } + + def location_content_transfer_complete(self, location): + return { + "message_type": "success", + "body": _("Location Content Transfer from {} complete").format( + location.name + ), + } + def transfer_confirm_done(self): return { "message_type": "warning", @@ -315,3 +329,21 @@ def transfer_no_qty_done(self): "No quantity has been processed, unable to complete the transfer." ), } + + def recovered_previous_session(self): + return { + "message_type": "info", + "body": _("Recovered previous session."), + } + + def no_lines_to_process(self): + return { + "message_type": "info", + "body": _("No lines to process."), + } + + def location_empty(self, location): + return { + "message_type": "info", + "body": _("Location {} empty").format(location.name), + } diff --git a/shopfloor/demo/shopfloor_menu_demo.xml b/shopfloor/demo/shopfloor_menu_demo.xml index 0c2f671777..4929c516c0 100644 --- a/shopfloor/demo/shopfloor_menu_demo.xml +++ b/shopfloor/demo/shopfloor_menu_demo.xml @@ -36,4 +36,14 @@ eval="[(4, ref('shopfloor.picking_type_delivery_demo'))]" /> + + Location Content Transfer + 60 + + location_content_transfer + + diff --git a/shopfloor/demo/stock_picking_type_demo.xml b/shopfloor/demo/stock_picking_type_demo.xml index c64cf5060d..2dbedd83ad 100644 --- a/shopfloor/demo/stock_picking_type_demo.xml +++ b/shopfloor/demo/stock_picking_type_demo.xml @@ -60,4 +60,19 @@ + + Location Content Transfer + LCT + + + + + + + + + internal + + + diff --git a/shopfloor/models/shopfloor_menu.py b/shopfloor/models/shopfloor_menu.py index 362c96c58a..51f97a56ff 100644 --- a/shopfloor/models/shopfloor_menu.py +++ b/shopfloor/models/shopfloor_menu.py @@ -6,7 +6,10 @@ class ShopfloorMenu(models.Model): _description = "Menu displayed in the scanner application" _order = "sequence" - _scenario_allowing_create_moves = ("single_pack_transfer",) + _scenario_allowing_create_moves = ( + "single_pack_transfer", + "location_content_transfer", + ) name = fields.Char(translate=True) sequence = fields.Integer() @@ -37,6 +40,7 @@ def _selection_scenario(self): ("cluster_picking", "Cluster Picking"), ("checkout", "Checkout/Packing"), ("delivery", "Delivery"), + ("location_content_transfer", "Location Content Transfer"), ] @api.depends("scenario", "picking_type_ids") diff --git a/shopfloor/models/stock_package_level.py b/shopfloor/models/stock_package_level.py index 1db120cc2f..1aa691c6e9 100644 --- a/shopfloor/models/stock_package_level.py +++ b/shopfloor/models/stock_package_level.py @@ -1,9 +1,16 @@ -from odoo import models +from odoo import fields, models class StockPackageLevel(models.Model): _inherit = "stock.package_level" + shopfloor_postponed = fields.Boolean( + default=False, + copy=False, + help="Technical field. " + "Indicates if a the package level has been postponed in a barcode scenario.", + ) + def replace_package(self, new_package): """Replace a package on an assigned package level and related records diff --git a/shopfloor/models/stock_picking.py b/shopfloor/models/stock_picking.py index d266f38cce..cdcd93cd08 100644 --- a/shopfloor/models/stock_picking.py +++ b/shopfloor/models/stock_picking.py @@ -30,3 +30,9 @@ def _calc_weight(self): for move_line in self.mapped("move_line_ids"): weight += move_line.product_qty * move_line.product_id.weight return weight + + def _create_backorder(self): + if self.env.context.get("_sf_no_backorder"): + return self.browse() + else: + return super()._create_backorder() diff --git a/shopfloor/services/__init__.py b/shopfloor/services/__init__.py index 85000d1508..09470e83ee 100644 --- a/shopfloor/services/__init__.py +++ b/shopfloor/services/__init__.py @@ -14,4 +14,5 @@ from . import checkout from . import cluster_picking from . import delivery +from . import location_content_transfer from . import single_pack_transfer diff --git a/shopfloor/services/checkout.py b/shopfloor/services/checkout.py index d73ee56489..3f08861d8a 100644 --- a/shopfloor/services/checkout.py +++ b/shopfloor/services/checkout.py @@ -308,7 +308,9 @@ def scan_line(self, picking_id, barcode): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) search = self.actions_for("search") @@ -457,7 +459,9 @@ def select_line(self, picking_id, package_id=None, move_line_id=None): picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) selection_lines = self._lines_to_pack(picking) if not selection_lines: @@ -475,7 +479,9 @@ def _change_line_qty( ): picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) move_lines = self.env["stock.move.line"].browse(move_line_ids).exists() @@ -669,7 +675,9 @@ def scan_package_action(self, picking_id, selected_line_ids, barcode): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) search = self.actions_for("search") selected_lines = self.env["stock.move.line"].browse(selected_line_ids).exists() @@ -718,7 +726,9 @@ def new_package(self, picking_id, selected_line_ids): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) selected_lines = self.env["stock.move.line"].browse(selected_line_ids).exists() return self._create_and_assign_new_packaging(picking, selected_lines) @@ -734,7 +744,9 @@ def no_package(self, picking_id, selected_line_ids): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) selected_lines = self.env["stock.move.line"].browse(selected_line_ids).exists() selected_lines.write( {"shopfloor_checkout_done": True, "result_package_id": False} @@ -759,7 +771,9 @@ def list_dest_package(self, picking_id, selected_line_ids): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) lines = self.env["stock.move.line"].browse(selected_line_ids).exists() return self._response_for_select_dest_package(picking, lines) @@ -797,7 +811,9 @@ def scan_dest_package(self, picking_id, selected_line_ids, barcode): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) lines = self.env["stock.move.line"].browse(selected_line_ids).exists() search = self.actions_for("search") package = search.package_from_scan(barcode) @@ -817,7 +833,9 @@ def set_dest_package(self, picking_id, selected_line_ids, package_id): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) lines = self.env["stock.move.line"].browse(selected_line_ids).exists() package = self.env["stock.quant.package"].browse(package_id).exists() return self._set_dest_package_from_selection(picking, lines, package) @@ -830,7 +848,9 @@ def summary(self, picking_id): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) return self._response_for_summary(picking) def _get_allowed_packaging(self): @@ -848,7 +868,9 @@ def list_packaging(self, picking_id, package_id): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) package = self.env["stock.quant.package"].browse(package_id).exists() packaging_list = self._get_allowed_packaging() @@ -863,7 +885,9 @@ def set_packaging(self, picking_id, package_id, packaging_id): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) package = self.env["stock.quant.package"].browse(package_id).exists() packaging = self.env["product.packaging"].browse(packaging_id).exists() @@ -898,7 +922,9 @@ def cancel_line(self, picking_id, package_id=None, line_id=None): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) package = self.env["stock.quant.package"].browse(package_id).exists() line = self.env["stock.move.line"].browse(line_id).exists() @@ -941,7 +967,9 @@ def done(self, picking_id, confirmation=False): """ picking = self.env["stock.picking"].browse(picking_id) if not picking.exists(): - return self._response_stock_picking_does_not_exist() + return self._response_for_select_document( + message=self.msg_store.stock_picking_not_found() + ) lines = picking.move_line_ids if not confirmation: if not all(line.qty_done == line.product_uom_qty for line in lines): @@ -1167,7 +1195,7 @@ def _schema_stock_picking(self, lines_with_packaging=False): { "move_lines": self.schemas._schema_list_of( self.schemas.move_line(with_packaging=lines_with_packaging) - ), + ) } ) return {"picking": self.schemas._schema_dict_of(schema, required=True)} diff --git a/shopfloor/services/location_content_transfer.py b/shopfloor/services/location_content_transfer.py new file mode 100644 index 0000000000..c2869d7692 --- /dev/null +++ b/shopfloor/services/location_content_transfer.py @@ -0,0 +1,944 @@ +from odoo import _ + +from odoo.addons.base_rest.components.service import to_int +from odoo.addons.component.core import Component + +from .service import to_float + +# NOTE for the implementation: share several similarities with the "cluster +# picking" scenario + + +# TODO add picking and package content in package level? + + +class LocationContentTransfer(Component): + """ + Methods for the Location Content Transfer Process + + Move the full content of a location to one or another location. + + Generally used to move a pallet with multiple boxes to either: + + * 1 destination location, unloading the full pallet + * To multiple destination locations, unloading one product/lot per + locations + * To multiple destination locations, unloading one product/lot per + locations and then unloading all remaining product/lot to a single final + destination + + The move lines must exist beforehand, the workflow only moves them. + + Expected: + + * All move lines and package level have a destination set, and are done. + + 2 complementary actions are possible on the screens allowing to move a line: + + * Declare a stock out for a product or package (nothing found in the + location) + * Skip to the next line (will be asked again at the end) + + Flow Diagram: https://www.draw.io/#G1qRenBcezk50ggIazDuu2qOfkTsoIAxXP + """ + + _inherit = "base.shopfloor.process" + _name = "shopfloor.location.content.transfer" + _usage = "location_content_transfer" + _description = __doc__ + + def _response_for_start(self, message=None): + """Transition to the 'start' state""" + return self._response(next_state="start", message=message) + + def _response_for_scan_destination_all( + self, pickings, message=None, confirmation_required=False + ): + """Transition to the 'scan_destination_all' state + + The client screen shows a summary of all the lines and packages + to move to a single destination. + + If `confirmation_required` is set, + the client will ask to scan again the destination + """ + data = self._data_content_all_for_location(pickings=pickings) + data["confirmation_required"] = confirmation_required + if confirmation_required and not message: + message = self.msg_store.need_confirmation() + return self._response( + next_state="scan_destination_all", data=data, message=message, + ) + + def _response_for_start_single(self, pickings, message=None): + """Transition to the 'start_single' state + + The client screen shows details of the package level or move line to move. + """ + location = pickings.mapped("location_id") + next_content = self._next_content(pickings) + if not next_content: + # TODO test (no more lines) + return self._response_for_start( + message=self.msg_store.location_content_transfer_complete(location) + ) + return self._response( + next_state="start_single", + data=self._data_content_line_for_location(location, next_content), + message=message, + ) + + def _response_for_scan_destination( + self, location, next_content, message=None, confirmation_required=False + ): + """Transition to the 'scan_destination' state + + The client screen shows details of the package level or move line to move. + """ + data = self._data_content_line_for_location(location, next_content) + data["confirmation_required"] = confirmation_required + if confirmation_required and not message: + message = self.msg_store.need_confirmation() + return self._response( + next_state="scan_destination", data=data, message=message, + ) + + def _data_content_all_for_location(self, pickings): + sorter = self.actions_for("location_content_transfer.sorter") + sorter.feed_pickings(pickings) + lines = sorter.move_lines() + package_levels = sorter.package_levels() + location = pickings.mapped("move_line_ids.location_id") + assert len(location) == 1, "There should be only one src location at this stage" + return { + "location": self.data.location(location), + "move_lines": self.data.move_lines(lines), + "package_levels": self.data.package_levels(package_levels), + } + + def _data_content_line_for_location(self, location, next_content): + assert next_content._name in ("stock.move.line", "stock.package_level") + line_data = ( + self.data.move_line(next_content) + if next_content._name == "stock.move.line" + else None + ) + level_data = ( + self.data.package_level(next_content) + if next_content._name == "stock.package_level" + else None + ) + return {"move_line": line_data, "package_level": level_data} + + def _next_content(self, pickings): + sorter = self.actions_for("location_content_transfer.sorter") + sorter.feed_pickings(pickings) + try: + next_content = next(sorter) + except StopIteration: + # TODO set picking to done + return None + return next_content + + def _router_single_or_all_destination(self, pickings, message=None): + if len(pickings.mapped("move_line_ids.location_dest_id")) == 1: + return self._response_for_scan_destination_all(pickings, message=message) + else: + return self._response_for_start_single(pickings, message=message) + + def _domain_recover_pickings(self): + return [ + ("user_id", "=", self.env.uid), + ("state", "in", ("assigned", "partially_available")), + ("picking_type_id", "in", self.picking_types.ids), + ] + + def _search_recover_pickings(self): + candidate_pickings = self.env["stock.picking"].search( + self._domain_recover_pickings() + ) + started_pickings = candidate_pickings.filtered( + lambda picking: any(line.qty_done for line in picking.move_line_ids) + ) + return started_pickings + + def start_or_recover(self): + """Start a new session or recover an existing one + + If the current user had transfers in progress in this scenario + and reopen the menu, we want to directly reopen the screens to choose + destinations. Otherwise, we go to the "start" state. + """ + started_pickings = self._search_recover_pickings() + if started_pickings: + return self._router_single_or_all_destination( + started_pickings, message=self.msg_store.recovered_previous_session() + ) + return self._response_for_start() + + def _find_location_move_lines_domain(self, location): + return [ + ("location_id", "=", location.id), + ("qty_done", "=", 0), + ("state", "in", ("assigned", "partially_available")), + ] + + def _find_location_move_lines(self, location): + """Find lines that potentially are to move in the location""" + return self.env["stock.move.line"].search( + self._find_location_move_lines_domain(location) + ) + + def _create_moves_from_location(self, location): + # get all quants from the scanned location + quants = self.env["stock.quant"].search( + [("location_id", "=", location.id), ("quantity", ">", 0)] + ) + # create moves for each quant + picking_type = self.work.menu.picking_type_ids + move_vals_list = [] + for quant in quants: + move_vals_list.append( + { + "name": quant.product_id.name, + "company_id": picking_type.company_id.id, + "product_id": quant.product_id.id, + "product_uom": quant.product_uom_id.id, + "product_uom_qty": quant.quantity, + "location_id": location.id, + "location_dest_id": picking_type.default_location_dest_id.id, + "origin": self.work.menu.name, + "picking_type_id": picking_type.id, + } + ) + return self.env["stock.move"].create(move_vals_list) + + def scan_location(self, barcode): + """Scan start location + + Called at the beginning at the workflow to select the location from which + we want to move the content. + + All the move lines and package levels must have the same picking type. + + If the scanned location has no move lines, new move lines to move the + whole content of the location are created if: + + * the menu has the option "Allow to create move(s)" + * the menu is linked to only one picking type. + + When move lines and package levels have different destinations, the + first line without package level or package level is sent to the client. + + Transitions: + * start: location not found, ... + * scan_destination_all: if the destination of all the lines and package + levels have the same destination + * start_single: if any line or package level has a different destination + """ + location = self.actions_for("search").location_from_scan(barcode) + if not location: + return self._response_for_start(message=self.msg_store.barcode_not_found()) + move_lines = self._find_location_move_lines(location) + pickings = move_lines.mapped("picking_id") + picking_types = pickings.mapped("picking_type_id") + + if len(picking_types) > 1: + return self._response_for_start( + message={ + "message_type": "error", + "body": _("This location content can't be moved at once."), + } + ) + if picking_types - self.picking_types: + return self._response_for_start( + message={ + "message_type": "error", + "body": _("This location content can't be moved using this menu."), + } + ) + # If the following criteria are met: + # - no move lines have been found + # - the menu is configured to allow the creation of moves + # - the menu is bind to one picking type + # - scanned location is a child of the picking type source location + # then prepare new stock moves to move goods from the scanned location. + menu = self.work.menu + if ( + not move_lines + and menu.allow_move_create + and len(menu.picking_type_ids) == 1 + and location.is_sublocation_of( + menu.picking_type_ids.default_location_src_id + ) + ): + new_moves = self._create_moves_from_location(location) + new_moves._action_confirm(merge=False) + new_moves._action_assign() + if not all([x.state == "assigned" for x in new_moves]): + new_moves._action_cancel() + return self._response_for_start( + message={ + "message_type": "error", + "body": _("New move lines cannot be assigned: canceled."), + } + ) + pickings = new_moves.mapped("picking_id") + move_lines = new_moves.move_line_ids + + if not pickings: + return self._response_for_start( + message=self.msg_store.location_empty(location) + ) + + for line in move_lines: + line.qty_done = line.product_uom_qty + + pickings.user_id = self.env.uid + + return self._router_single_or_all_destination(pickings) + + def _find_transfer_move_lines_domain(self, location): + return [ + ("location_id", "=", location.id), + ("state", "in", ("assigned", "partially_available")), + ("qty_done", ">", 0), + # TODO check generated SQL + ("picking_id.user_id", "=", self.env.uid), + ] + + def _find_transfer_move_lines(self, location): + """Find move lines currently being moved by the user""" + lines = self.env["stock.move.line"].search( + self._find_transfer_move_lines_domain(location) + ) + return lines + + def _set_destination_lines(self, pickings, move_lines, dest_location): + move_lines.location_dest_id = dest_location + move_lines.package_level_id.location_dest_id = dest_location + pickings.action_done() + + def _split_other_move_lines(self, move, move_lines): + """Substract `move_lines` from `move.move_line_ids` and put the result + in a new move. + """ + other_move_lines = move.move_line_ids - move_lines + if other_move_lines: + qty_to_split = sum(other_move_lines.mapped("product_uom_qty")) + backorder_move_id = move._split(qty_to_split) + backorder_move = self.env["stock.move"].browse(backorder_move_id) + backorder_move.move_line_ids = other_move_lines + backorder_move._action_assign() + return backorder_move + return False + + def set_destination_all(self, location_id, barcode, confirmation=False): + """Scan destination location for all the moves of the location + + barcode is a stock.location for the destination + + Transitions: + * scan_destination_all: invalid destination or could not set moves to done + * start: moves are done + """ + location = self.env["stock.location"].browse(location_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + move_lines = self._find_transfer_move_lines(location) + pickings = move_lines.mapped("picking_id") + scanned_location = self.actions_for("search").location_from_scan(barcode) + if not scanned_location: + return self._response_for_scan_destination_all( + pickings, message=self.msg_store.barcode_not_found() + ) + + if not scanned_location.is_sublocation_of( + self.picking_types.mapped("default_location_dest_id") + ): + return self._response_for_scan_destination_all( + pickings, message=self.msg_store.dest_location_not_allowed() + ) + if not confirmation and not scanned_location.is_sublocation_of( + move_lines.mapped("location_dest_id") + ): + # the scanned location is valid (child of picking type's destination) + # but not the expected one: ask for confirmation + return self._response_for_scan_destination_all( + pickings, confirmation_required=True + ) + + self._set_destination_lines(pickings, move_lines, scanned_location) + + return self._response_for_start( + message={ + "message_type": "success", + "body": _("Content transferred from {}.").format(location.name), + } + ) + + def go_to_single(self, location_id): + """Ask the first move line or package level + + If the user was brought to the screen allowing to move everything to + the same location, but they want to move them to different locations, + this method will return the first move line or package level. + + Transitions: + * start: no remaining lines in the location + * start_single: if any line or package level has a different destination + """ + location = self.env["stock.location"].browse(location_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + move_lines = self._find_transfer_move_lines(location) + if not move_lines: + return self._response_for_start( + message=self.msg_store.no_lines_to_process() + ) + return self._response_for_start_single(move_lines.mapped("picking_id")) + + def scan_package(self, location_id, package_level_id, barcode): + """Scan a package level to move + + It validates that the user scanned the correct package, lot or product. + + Transitions: + * start: no remaining lines in the location + * start_single: barcode not found, ... + * scan_destination: the barcode matches + """ + location = self.env["stock.location"].browse(location_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + package_level = self.env["stock.package_level"].browse(package_level_id) + if not package_level.exists(): + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single( + move_lines.mapped("picking_id"), + message=self.msg_store.record_not_found(), + ) + + search = self.actions_for("search") + package = search.package_from_scan(barcode) + if package and package_level.package_id == package: + return self._response_for_scan_destination(location, package_level) + + move_lines = self._find_transfer_move_lines(location) + package_move_lines = package_level.move_line_ids + other_move_lines = move_lines - package_move_lines + + product = search.product_from_scan(barcode) + # Normally the user scan the barcode of the package. But if they scan the + # product and we can be sure it's the correct package, it's tolerated. + if product and product in package_move_lines.mapped("product_id"): + if product in other_move_lines.mapped("product_id") or product.tracking in ( + "lot", + "serial", + ): + # When the product exists in other move lines as raw products + # or part of another package, we can't be sure they scanned + # the correct package, so ask to scan the package. + return self._response_for_start_single( + move_lines.mapped("picking_id"), + message={"message_type": "error", "body": _("Scan the package")}, + ) + else: + return self._response_for_scan_destination(location, package_level) + + lot = search.lot_from_scan(barcode) + if lot and lot in package_move_lines.mapped("lot_id"): + if lot in other_move_lines.mapped("lot_id"): + return self._response_for_start_single( + move_lines.mapped("picking_id"), + message={"message_type": "error", "body": _("Scan the package")}, + ) + else: + return self._response_for_scan_destination(location, package_level) + + return self._response_for_start_single( + move_lines.mapped("picking_id"), message=self.msg_store.barcode_not_found() + ) + + def scan_line(self, location_id, move_line_id, barcode): + """Scan a move line to move + + It validates that the user scanned the correct package, lot or product. + + Transitions: + * start: no remaining lines in the location + * start_single: barcode not found, ... + * scan_destination: the barcode matches + """ + location = self.env["stock.location"].browse(location_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + move_line = self.env["stock.move.line"].browse(move_line_id) + if not move_line.exists(): + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single( + move_lines.mapped("picking_id"), + message=self.msg_store.record_not_found(), + ) + + search = self.actions_for("search") + product = search.product_from_scan(barcode) + if product and product == move_line.product_id: + if product.tracking in ("lot", "serial"): + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single( + move_lines.mapped("picking_id"), + message=self.msg_store.scan_lot_on_product_tracked_by_lot(), + ) + else: + return self._response_for_scan_destination(location, move_line) + + lot = search.lot_from_scan(barcode) + if lot and lot == move_line.lot_id: + return self._response_for_scan_destination(location, move_line) + + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single( + move_lines.mapped("picking_id"), message=self.msg_store.barcode_not_found() + ) + + def set_destination_package( + self, location_id, package_level_id, barcode, confirmation=False + ): + """Scan destination location for package level + + If the move has other move lines / package levels it has to be split + so we can post only this part. + + After the destination is set, the move is set to done. + + Beware, when _action_done() is called on the move, the normal behavior + of Odoo would be to create a backorder transfer. We don't want this or + we would have a backorder per move. The context key + ``_sf_no_backorder`` disables the creation of backorders, it must be set + on all moves, but the last one of a transfer (so in case something was not + available, a backorder is created). + + Transitions: + * scan_destination: invalid destination or could not + * start_single: continue with the next package level / line + * start: if there is no more package level / line to process + """ + location = self.env["stock.location"].browse(location_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + package_level = self.env["stock.package_level"].browse(package_level_id) + if not package_level.exists(): + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single(move_lines.mapped("picking_id")) + search = self.actions_for("search") + scanned_location = search.location_from_scan(barcode) + if not scanned_location: + return self._response_for_scan_destination( + location, package_level, message=self.msg_store.no_location_found() + ) + if not scanned_location.is_sublocation_of( + package_level.picking_id.picking_type_id.default_location_dest_id + ): + return self._response_for_scan_destination( + location, + package_level, + message=self.msg_store.dest_location_not_allowed(), + ) + if not scanned_location.is_sublocation_of(package_level.location_dest_id): + if not confirmation: + return self._response_for_scan_destination( + location, package_level, confirmation_required=True + ) + package_move_lines = package_level.move_line_ids + package_moves = package_move_lines.mapped("move_id") + for package_move in package_moves: + # Check if there is no other lines linked to the move others than + # the lines related to the package itself. In such case we have to + # split the move to process only the lines related to the package. + self._split_other_move_lines(package_move, package_move_lines) + package_level.location_dest_id = scanned_location + package_moves.with_context(_sf_no_backorder=True)._action_done() + move_lines = self._find_transfer_move_lines(location) + message = self.msg_store.location_content_transfer_item_complete( + scanned_location + ) + return self._response_for_start_single( + move_lines.mapped("picking_id"), message=message + ) + + def set_destination_line( + self, location_id, move_line_id, quantity, barcode, confirmation=False + ): + """Scan destination location for move line + + If the quantity < qty of the line, split the move and reserve it. + If the move has other move lines / package levels it has to be split + so we can post only this part. + + After the destination and quantity are set, the move is set to done. + + Beware, when _action_done() is called on the move, the normal behavior + of Odoo would be to create a backorder transfer. We don't want this or + we would have a backorder per move. The context key + ``_sf_no_backorder`` disables the creation of backorders, it must be set + on all moves, but the last one of a transfer (so in case something was not + available, a backorder is created). + + Transitions: + * scan_destination: invalid destination or could not + * start_single: continue with the next package level / line + * start: if there is no more package level / line to process + """ + location = self.env["stock.location"].browse(location_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + move_line = self.env["stock.move.line"].browse(move_line_id) + if not move_line.exists(): + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single(move_lines.mapped("picking_id")) + search = self.actions_for("search") + scanned_location = search.location_from_scan(barcode) + if not scanned_location: + return self._response_for_scan_destination( + location, move_line, message=self.msg_store.no_location_found() + ) + if not scanned_location.is_sublocation_of( + move_line.picking_id.picking_type_id.default_location_dest_id + ): + return self._response_for_scan_destination( + location, move_line, message=self.msg_store.dest_location_not_allowed() + ) + if not scanned_location.is_sublocation_of(move_line.location_dest_id): + if not confirmation: + return self._response_for_scan_destination( + location, move_line, confirmation_required=True + ) + if quantity < move_line.product_uom_qty: + # Update the current move line quantity and + # put the scanned qty (the move line) in its own move + # (by splitting the current one) + move_line.product_uom_qty = move_line.qty_done = quantity + current_move = move_line.move_id + new_move_id = current_move._split(quantity) + new_move = self.env["stock.move"].browse(new_move_id) + new_move.move_line_ids = move_line + # Ensure that the remaining qty to process is reserved as before + current_move._recompute_state() + (new_move | current_move)._action_assign() + for remaining_move_line in current_move.move_line_ids: + remaining_move_line.qty_done = remaining_move_line.product_uom_qty + self._split_other_move_lines(move_line.move_id, move_line) + move_line.location_dest_id = scanned_location + move_line.move_id.with_context(_sf_no_backorder=True)._action_done() + move_lines = self._find_transfer_move_lines(location) + message = self.msg_store.location_content_transfer_item_complete( + scanned_location + ) + return self._response_for_start_single( + move_lines.mapped("picking_id"), message=message + ) + + def postpone_package(self, location_id, package_level_id): + """Mark a package level as postponed and return the next level/line + + Transitions: + * start_single: continue with the next package level / line + """ + location = self.env["stock.location"].browse(location_id) + package_level = self.env["stock.package_level"].browse(package_level_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + if package_level.exists(): + package_level.shopfloor_postponed = True + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single(move_lines.mapped("picking_id")) + + def postpone_line(self, location_id, move_line_id): + """Mark a move line as postponed and return the next level/line + + Transitions: + * start_single: continue with the next package level / line + """ + location = self.env["stock.location"].browse(location_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + move_line = self.env["stock.move.line"].browse(move_line_id) + if move_line.exists(): + move_line.shopfloor_postponed = True + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single(move_lines.mapped("picking_id")) + + def stock_out_package(self, location_id, package_level_id): + """Declare a stock out on a package level + + It first ensures the stock.move only has this package level. If not, it + splits the move to have no side-effect on the other package levels/move + lines. + + It unreserves the move, create an inventory at 0 in the move's source + location, create a second draft inventory (if none exists) to check later. + Finally, it cancels the move. + + Transitions: + * start: no more content to move + * start_single: continue with the next package level / line + """ + location = self.env["stock.location"].browse(location_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + package_level = self.env["stock.package_level"].browse(package_level_id) + if not package_level.exists(): + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single(move_lines.mapped("picking_id")) + inventory = self.actions_for("inventory") + package_move_lines = package_level.move_line_ids + package_moves = package_move_lines.mapped("move_id") + for package_move in package_moves: + # Check if there is no other lines linked to the move others than + # the lines related to the package itself. In such case we have to + # split the move to process only the lines related to the package. + self._split_other_move_lines(package_move, package_move_lines) + lot = package_move.move_line_ids.lot_id + package_move._do_unreserve() + package_move._recompute_state() + # Create an inventory at 0 in the move's source location + inventory.create_stock_issue( + package_move, location, package_level.package_id, lot + ) + # Create a draft inventory to control stock + inventory.create_control_stock( + location, package_move.product_id, package_level.package_id, lot + ) + package_move._action_cancel() + # remove the package level (this is what does the `picking.do_unreserve()` + # method, but here we want to unreserve+unlink this package alone) + assert package_level.state == "draft", "Package level has to be in draft" + if package_level.state != "draft": + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single( + move_lines.mapped("picking_id"), + message={ + "message_type": "error", + "body": _("Package level has to be in draft"), + }, + ) + package_level.unlink() + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single(move_lines.mapped("picking_id")) + + def stock_out_line(self, location_id, move_line_id): + """Declare a stock out on a move line + + It first ensures the stock.move only has this move line. If not, it + splits the move to have no side-effect on the other package levels/move + lines. + + It unreserves the move, create an inventory at 0 in the move's source + location, create a second draft inventory (if none exists) to check later. + Finally, it cancels the move. + + Transitions: + * start: no more content to move + * start_single: continue with the next package level / line + """ + location = self.env["stock.location"].browse(location_id) + if not location.exists(): + return self._response_for_start(message=self.msg_store.record_not_found()) + move_line = self.env["stock.move.line"].browse(move_line_id) + if not move_line.exists(): + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single(move_lines.mapped("picking_id")) + inventory = self.actions_for("inventory") + self._split_other_move_lines(move_line.move_id, move_line) + move_line_src_location = move_line.location_id + move = move_line.move_id + package = move_line.package_id + lot = move_line.lot_id + move._do_unreserve() + move._recompute_state() + # Create an inventory at 0 in the move's source location + inventory.create_stock_issue(move, move_line_src_location, package, lot) + # Create a draft inventory to control stock + inventory.create_control_stock( + move_line_src_location, move.product_id, package, lot + ) + move._action_cancel() + move_lines = self._find_transfer_move_lines(location) + return self._response_for_start_single(move_lines.mapped("picking_id")) + + +class ShopfloorLocationContentTransferValidator(Component): + """Validators for the Location Content Transfer endpoints""" + + _inherit = "base.shopfloor.validator" + _name = "shopfloor.location.content.transfer.validator" + _usage = "location_content_transfer.validator" + + def start_or_recover(self): + return {} + + def scan_location(self): + return {"barcode": {"required": True, "type": "string"}} + + def set_destination_all(self): + return { + "location_id": {"coerce": to_int, "required": True, "type": "integer"}, + "barcode": {"required": True, "type": "string"}, + "confirmation": {"type": "boolean", "nullable": True, "required": False}, + } + + def go_to_single(self): + return {"location_id": {"coerce": to_int, "required": True, "type": "integer"}} + + def scan_package(self): + return { + "location_id": {"coerce": to_int, "required": True, "type": "integer"}, + "package_level_id": {"coerce": to_int, "required": True, "type": "integer"}, + "barcode": {"required": True, "type": "string"}, + } + + def scan_line(self): + return { + "location_id": {"coerce": to_int, "required": True, "type": "integer"}, + "move_line_id": {"coerce": to_int, "required": True, "type": "integer"}, + "barcode": {"required": True, "type": "string"}, + } + + def set_destination_package(self): + return { + "location_id": {"coerce": to_int, "required": True, "type": "integer"}, + "package_level_id": {"coerce": to_int, "required": True, "type": "integer"}, + "barcode": {"required": True, "type": "string"}, + "confirmation": {"type": "boolean", "nullable": True, "required": False}, + } + + def set_destination_line(self): + return { + "location_id": {"coerce": to_int, "required": True, "type": "integer"}, + "move_line_id": {"coerce": to_int, "required": True, "type": "integer"}, + "quantity": {"coerce": to_float, "required": True, "type": "float"}, + "barcode": {"required": True, "type": "string"}, + "confirmation": {"type": "boolean", "nullable": True, "required": False}, + } + + def postpone_package(self): + return { + "location_id": {"coerce": to_int, "required": True, "type": "integer"}, + "package_level_id": {"coerce": to_int, "required": True, "type": "integer"}, + } + + def postpone_line(self): + return { + "location_id": {"coerce": to_int, "required": True, "type": "integer"}, + "move_line_id": {"coerce": to_int, "required": True, "type": "integer"}, + } + + def stock_out_package(self): + return { + "location_id": {"coerce": to_int, "required": True, "type": "integer"}, + "package_level_id": {"coerce": to_int, "required": True, "type": "integer"}, + } + + def stock_out_line(self): + return { + "location_id": {"coerce": to_int, "required": True, "type": "integer"}, + "move_line_id": {"coerce": to_int, "required": True, "type": "integer"}, + } + + +class ShopfloorLocationContentTransferValidatorResponse(Component): + """Validators for the Location Content Transfer endpoints responses""" + + _inherit = "base.shopfloor.validator.response" + _name = "shopfloor.location.content.transfer.validator.response" + _usage = "location_content_transfer.validator.response" + + def _states(self): + """List of possible next states + + With the schema of the data send to the client to transition + to the next state. + """ + return { + "start": {}, + "scan_destination_all": self._schema_all, + "start_single": self._schema_single, + "scan_destination": self._schema_single, + } + + @property + def _schema_all(self): + package_level_schema = self.schemas.package_level() + move_line_schema = self.schemas.move_line() + return { + "location": self.schemas._schema_dict_of(self.schemas.location()), + # we'll display all the packages and move lines *without package + # levels* + "package_levels": self.schemas._schema_list_of(package_level_schema), + "move_lines": self.schemas._schema_list_of(move_line_schema), + "confirmation_required": { + "type": "boolean", + "nullable": True, + "required": False, + }, + } + + @property + def _schema_single(self): + schema_package_level = self.schemas.package_level() + schema_move_line = self.schemas.move_line() + return { + # we'll have one or the other... + "package_level": self.schemas._schema_dict_of(schema_package_level), + "move_line": self.schemas._schema_dict_of(schema_move_line), + "confirmation_required": { + "type": "boolean", + "nullable": True, + "required": False, + }, + } + + def start_or_recover(self): + return self._response_schema( + next_states={"start", "scan_destination_all", "start_single"} + ) + + def scan_location(self): + return self._response_schema( + next_states={"start", "scan_destination_all", "start_single"} + ) + + def set_destination_all(self): + return self._response_schema(next_states={"start", "scan_destination_all"}) + + def go_to_single(self): + return self._response_schema(next_states={"start", "start_single"}) + + def scan_package(self): + return self._response_schema( + next_states={"start", "start_single", "scan_destination"} + ) + + def scan_line(self): + return self._response_schema( + next_states={"start", "start_single", "scan_destination"} + ) + + def set_destination_package(self): + return self._response_schema(next_states={"start_single", "scan_destination"}) + + def set_destination_line(self): + return self._response_schema(next_states={"start_single", "scan_destination"}) + + def postpone_package(self): + return self._response_schema(next_states={"start_single"}) + + def postpone_line(self): + return self._response_schema(next_states={"start_single"}) + + def stock_out_package(self): + return self._response_schema(next_states={"start", "start_single"}) + + def stock_out_line(self): + return self._response_schema(next_states={"start", "start_single"}) diff --git a/shopfloor/services/schema.py b/shopfloor/services/schema.py index f894f82250..c7d78c39e3 100644 --- a/shopfloor/services/schema.py +++ b/shopfloor/services/schema.py @@ -76,35 +76,21 @@ def move_line(self, with_packaging=False): "id": {"type": "integer", "required": True}, "qty_done": {"type": "float", "required": True}, "quantity": {"type": "float", "required": True}, - "product": {"type": "dict", "required": True, "schema": self.product()}, + "product": self._schema_dict_of(self.product()), "lot": { "type": "dict", "required": False, "nullable": True, "schema": self.lot(), }, - "package_src": { - "type": "dict", - "required": True, - "nullable": True, - "schema": self.package(with_packaging=with_packaging), - }, - "package_dest": { - "type": "dict", - "required": False, - "nullable": True, - "schema": self.package(with_packaging=with_packaging), - }, - "location_src": { - "type": "dict", - "required": True, - "schema": self.location(), - }, - "location_dest": { - "type": "dict", - "required": True, - "schema": self.location(), - }, + "package_src": self._schema_dict_of( + self.package(with_packaging=with_packaging) + ), + "package_dest": self._schema_dict_of( + self.package(with_packaging=with_packaging), required=False + ), + "location_src": self._schema_dict_of(self.location()), + "location_dest": self._schema_dict_of(self.location()), } def product(self): @@ -170,9 +156,10 @@ def picking_batch(self, with_pickings=False): def package_level(self): return { "id": {"required": True, "type": "integer"}, - "name": {"type": "string", "nullable": False, "required": True}, - "location_src": {"type": "dict", "schema": self.location()}, - "location_dest": {"type": "dict", "schema": self.location()}, - "product": {"type": "dict", "schema": self.product()}, - "picking": {"type": "dict", "schema": self.picking()}, + "is_done": {"type": "boolean", "nullable": False, "required": True}, + "package_src": self._schema_dict_of(self.package()), + "location_src": self._schema_dict_of(self.location()), + "location_dest": self._schema_dict_of(self.location()), + "product": self._schema_dict_of(self.product()), + "quantity": {"type": "float", "required": True}, } diff --git a/shopfloor/services/single_pack_transfer.py b/shopfloor/services/single_pack_transfer.py index 085f851a56..2306432b18 100644 --- a/shopfloor/services/single_pack_transfer.py +++ b/shopfloor/services/single_pack_transfer.py @@ -13,6 +13,7 @@ class SinglePackTransfer(Component): def _data_after_package_scanned(self, package_level): move_line = package_level.move_line_ids[0] package = package_level.package_id + # TODO use data.package_level (but the "name" moves in "package.name") return { "id": package_level.id, "name": package.name, @@ -288,4 +289,12 @@ def validate(self): @property def _schema_for_package_level_details(self): - return self.schemas.package_level() + # TODO use schemas.package_level (but the "name" moves in "package.name") + return { + "id": {"required": True, "type": "integer"}, + "name": {"type": "string", "nullable": False, "required": True}, + "location_src": {"type": "dict", "schema": self.schemas.location()}, + "location_dest": {"type": "dict", "schema": self.schemas.location()}, + "product": {"type": "dict", "schema": self.schemas.product()}, + "picking": {"type": "dict", "schema": self.schemas.picking()}, + } diff --git a/shopfloor/tests/__init__.py b/shopfloor/tests/__init__.py index 5b470c54df..61dfbc481b 100644 --- a/shopfloor/tests/__init__.py +++ b/shopfloor/tests/__init__.py @@ -33,3 +33,8 @@ from . import test_delivery_set_qty_done_line from . import test_delivery_list_stock_picking from . import test_delivery_select +from . import test_location_content_transfer_base +from . import test_location_content_transfer_start +from . import test_location_content_transfer_set_destination_all +from . import test_location_content_transfer_single +from . import test_location_content_transfer_set_destination_package_or_line diff --git a/shopfloor/tests/common.py b/shopfloor/tests/common.py index 1f83cde1eb..b55dc7f269 100644 --- a/shopfloor/tests/common.py +++ b/shopfloor/tests/common.py @@ -202,7 +202,7 @@ def setUpClassBaseData(cls): .create( { "name": "Box", - "product_id": cls.product_b.id, + "product_id": cls.product_c.id, "barcode": "ProductCBox", } ) @@ -288,13 +288,15 @@ def _update_qty_in_location( ) @classmethod - def _fill_stock_for_moves(cls, moves, in_package=False, in_lot=False): + def _fill_stock_for_moves( + cls, moves, in_package=False, in_lot=False, location=False + ): product_locations = {} package = None if in_package: package = cls.env["stock.quant.package"].create({}) for move in moves: - key = (move.product_id, move.location_id) + key = (move.product_id, location or move.location_id) product_locations.setdefault(key, 0) product_locations[key] += move.product_qty for (product, location), qty in product_locations.items(): diff --git a/shopfloor/tests/test_actions_data.py b/shopfloor/tests/test_actions_data.py index e266525036..4be811ac45 100644 --- a/shopfloor/tests/test_actions_data.py +++ b/shopfloor/tests/test_actions_data.py @@ -88,6 +88,15 @@ def _expected_packaging(self, record, **kw): data.update(kw) return data + def _expected_package(self, record, **kw): + data = { + "id": record.id, + "name": record.name, + "weight": record.pack_weight, + } + data.update(kw) + return data + class ActionsDataCase(ActionsDataCaseBase): def test_data_packaging(self): @@ -133,6 +142,25 @@ def test_data_package(self): } self.assertDictEqual(data, expected) + def test_data_package_level(self): + package_level = self.picking.package_level_ids[0] + data = self.data.package_level(package_level) + self.assert_schema(self.schema.package_level(), data) + expected = { + "id": package_level.id, + "is_done": False, + "package_src": self._expected_package(package_level.package_id), + "location_dest": self._expected_location(package_level.location_dest_id), + "location_src": self._expected_location( + package_level.picking_id.location_id + ), + "product": self._expected_product( + package_level.package_id.single_product_id + ), + "quantity": package_level.package_id.single_product_qty, + } + self.assertDictEqual(data, expected) + def test_data_picking(self): self.picking.write({"origin": "created by test", "note": "read me"}) data = self.data.picking(self.picking) diff --git a/shopfloor/tests/test_location_content_transfer_base.py b/shopfloor/tests/test_location_content_transfer_base.py new file mode 100644 index 0000000000..0bf62b4ef7 --- /dev/null +++ b/shopfloor/tests/test_location_content_transfer_base.py @@ -0,0 +1,114 @@ +from .common import CommonCase + + +class LocationContentTransferCommonCase(CommonCase): + @classmethod + def setUpClassVars(cls, *args, **kwargs): + super().setUpClassVars(*args, **kwargs) + cls.menu = cls.env.ref("shopfloor.shopfloor_menu_location_content_transfer") + cls.profile = cls.env.ref("shopfloor.shopfloor_profile_shelf_1_demo") + cls.wh = cls.profile.warehouse_id + cls.picking_type = cls.menu.picking_type_ids + + @classmethod + def setUpClassBaseData(cls, *args, **kwargs): + super().setUpClassBaseData(*args, **kwargs) + cls.content_loc = ( + cls.env["stock.location"] + .sudo() + .create( + { + "name": "Content Location", + "barcode": "Content", + "location_id": cls.picking_type.default_location_src_id.id, + } + ) + ) + + def setUp(self): + super().setUp() + with self.work_on_services(menu=self.menu, profile=self.profile) as work: + self.service = work.component(usage="location_content_transfer") + + @classmethod + def _simulate_pickings_selected(cls, pickings): + """Create a state as if pickings has been selected + + ... during a Location content transfer. + + It means a user scanned the location with the pickings. They are: + + * assigned to the user + * the qty_done of all their move lines is set to they reserved qty + + """ + pickings.user_id = cls.env.uid + for line in pickings.mapped("move_line_ids"): + line.qty_done = line.product_uom_qty + + def assert_response_start(self, response, message=None): + self.assert_response(response, next_state="start", message=message) + + def _assert_response_scan_destination_all( + self, state, response, pickings, message=None, confirmation_required=False + ): + # this code is repeated from the implementation, not great, but we + # mostly want to ensure the selection of pickings is right, and the + # data methods have their own tests + lines = pickings.move_line_ids.filtered(lambda line: not line.package_level_id) + package_levels = pickings.package_level_ids + location = lines.mapped("location_id") + self.assert_response( + response, + next_state=state, + data={ + "move_lines": self.data.move_lines(lines), + "package_levels": self.data.package_levels(package_levels), + "location": self.data.location(location), + "confirmation_required": confirmation_required, + }, + message=message, + ) + + def assert_response_scan_destination_all( + self, response, pickings, message=None, confirmation_required=False + ): + self._assert_response_scan_destination_all( + "scan_destination_all", + response, + pickings, + message=message, + confirmation_required=confirmation_required, + ) + + def assert_response_start_single(self, response, pickings, message=None): + sorter = self.service.actions_for("location_content_transfer.sorter") + sorter.feed_pickings(pickings) + location = pickings.mapped("location_id") + self.assert_response( + response, + next_state="start_single", + data=self.service._data_content_line_for_location(location, next(sorter)), + message=message, + ) + + def _assert_response_scan_destination( + self, state, response, next_content, message=None, confirmation_required=False + ): + location = next_content.location_id + data = self.service._data_content_line_for_location(location, next_content) + data["confirmation_required"] = confirmation_required + self.assert_response( + response, next_state=state, data=data, message=message, + ) + + def assert_response_scan_destination( + self, response, next_content, message=None, confirmation_required=False + ): + self._assert_response_scan_destination( + "scan_destination", + response, + next_content, + message=message, + confirmation_required=confirmation_required, + ) diff --git a/shopfloor/tests/test_location_content_transfer_set_destination_all.py b/shopfloor/tests/test_location_content_transfer_set_destination_all.py new file mode 100644 index 0000000000..35903b2dbf --- /dev/null +++ b/shopfloor/tests/test_location_content_transfer_set_destination_all.py @@ -0,0 +1,181 @@ +from .test_location_content_transfer_base import LocationContentTransferCommonCase + + +class LocationContentTransferSetDestinationAllCase(LocationContentTransferCommonCase): + """Tests for endpoint used from scan_destination_all + + * /set_destination_all + * /go_to_single + + """ + + # TODO see what can be common + @classmethod + def setUpClassBaseData(cls): + super().setUpClassBaseData() + products = cls.product_a + cls.product_b + cls.product_c + cls.product_d + for product in products: + cls.env["stock.putaway.rule"].sudo().create( + { + "product_id": product.id, + "location_in_id": cls.stock_location.id, + "location_out_id": cls.shelf1.id, + } + ) + + cls.picking1 = picking1 = cls._create_picking( + lines=[(cls.product_a, 10), (cls.product_b, 10)] + ) + cls.picking2 = picking2 = cls._create_picking( + lines=[(cls.product_c, 10), (cls.product_d, 10)] + ) + cls.pickings = picking1 | picking2 + cls._fill_stock_for_moves( + picking1.move_lines, in_package=True, location=cls.content_loc + ) + cls._fill_stock_for_moves(picking2.move_lines, location=cls.content_loc) + cls.pickings.action_assign() + cls._simulate_pickings_selected(cls.pickings) + + def assert_all_done(self, destination): + self.assertRecordValues(self.pickings, [{"state": "done"}, {"state": "done"}]) + self.assertRecordValues( + self.pickings.move_line_ids, + [ + {"qty_done": 10.0, "state": "done", "location_dest_id": destination.id}, + {"qty_done": 10.0, "state": "done", "location_dest_id": destination.id}, + {"qty_done": 10.0, "state": "done", "location_dest_id": destination.id}, + {"qty_done": 10.0, "state": "done", "location_dest_id": destination.id}, + ], + ) + self.assertRecordValues( + self.picking1.package_level_ids, + [{"is_done": True, "state": "done", "location_dest_id": destination.id}], + ) + + def test_set_destination_all_dest_location_ok(self): + """Scanned destination location valid, moves set to done accepted""" + sub_shelf1 = ( + self.env["stock.location"] + .sudo() + .create( + { + "name": "Sub Shelf 1", + "barcode": "subshelf1", + "location_id": self.shelf1.id, + } + ) + ) + response = self.service.dispatch( + "set_destination_all", + params={"location_id": self.content_loc.id, "barcode": sub_shelf1.barcode}, + ) + self.assert_response_start( + response, + message={ + "message_type": "success", + "body": "Content transferred from {}.".format(self.content_loc.name), + }, + ) + self.assert_all_done(sub_shelf1) + + def test_set_destination_all_dest_location_not_found(self): + """Barcode scanned for destination location is not found""" + response = self.service.dispatch( + "set_destination_all", + params={"location_id": self.content_loc.id, "barcode": "NOT_FOUND"}, + ) + self.assert_response_scan_destination_all( + response, self.pickings, message=self.service.msg_store.barcode_not_found() + ) + + def test_set_destination_all_dest_location_need_confirm(self): + """Scanned dest. location != child but in picking type location + + So it needs confirmation. + """ + response = self.service.dispatch( + "set_destination_all", + params={ + "location_id": self.content_loc.id, + # expected location was shelf1, but shelf2 is valid as still in the + # picking type's default dest location, ask confirmation (second scan) + # from the user + "barcode": self.shelf2.barcode, + }, + ) + self.assert_response_scan_destination_all( + response, + self.pickings, + message=self.service.msg_store.need_confirmation(), + confirmation_required=True, + ) + + def test_set_destination_all_dest_location_confirmation(self): + """Scanned dest. location != child but in picking type location: confirm + + use the confirmation flag to confirm + """ + response = self.service.dispatch( + "set_destination_all", + params={ + "location_id": self.content_loc.id, + # expected location was shelf1, but shelf2 is valid as still in the + # picking type's default dest location, ask confirmation (second scan) + # from the user + "barcode": self.shelf2.barcode, + "confirmation": True, + }, + ) + self.assert_response_start( + response, + message={ + "message_type": "success", + "body": "Content transferred from {}.".format(self.content_loc.name), + }, + ) + self.assert_all_done(self.shelf2) + + def test_set_destination_all_dest_location_invalid(self): + """The scanned destination location is not in the menu's picking types""" + response = self.service.dispatch( + "set_destination_all", + params={ + "location_id": self.content_loc.id, + "barcode": self.dispatch_location.barcode, + }, + ) + self.assert_response_scan_destination_all( + response, + self.pickings, + message=self.service.msg_store.dest_location_not_allowed(), + ) + + def test_go_to_single(self): + """User used to 'split by lines' button to process line per line""" + response = self.service.dispatch( + "go_to_single", params={"location_id": self.content_loc.id} + ) + self.assert_response_start_single(response, self.pickings) + + +class LocationContentTransferSetDestinationAllSpecialCase( + LocationContentTransferCommonCase +): + """Tests for endpoint used from scan_destination_all (special cases without setup) + + * /set_destination_all + * /go_to_single + + """ + + def test_go_to_single_no_lines_to_process(self): + """User used to 'split by lines' button to process line per line, + but no lines to process. + """ + response = self.service.dispatch( + "go_to_single", params={"location_id": self.content_loc.id} + ) + self.assert_response_start( + response, message=self.service.msg_store.no_lines_to_process() + ) diff --git a/shopfloor/tests/test_location_content_transfer_set_destination_package_or_line.py b/shopfloor/tests/test_location_content_transfer_set_destination_package_or_line.py new file mode 100644 index 0000000000..b46855f270 --- /dev/null +++ b/shopfloor/tests/test_location_content_transfer_set_destination_package_or_line.py @@ -0,0 +1,529 @@ +from .test_location_content_transfer_base import LocationContentTransferCommonCase + + +class LocationContentTransferSetDestinationXCase(LocationContentTransferCommonCase): + """Tests for endpoint used from scan_destination + + * /set_destination_package + * /set_destination_line + + """ + + # TODO see what can be common + @classmethod + def setUpClassBaseData(cls): + super().setUpClassBaseData() + products = cls.product_a + cls.product_b + cls.product_c + cls.product_d + for product in products: + cls.env["stock.putaway.rule"].sudo().create( + { + "product_id": product.id, + "location_in_id": cls.stock_location.id, + "location_out_id": cls.shelf1.id, + } + ) + + cls.picking1 = picking1 = cls._create_picking( + lines=[(cls.product_a, 10), (cls.product_b, 10)] + ) + cls.picking2 = picking2 = cls._create_picking( + lines=[(cls.product_c, 10), (cls.product_d, 10)] + ) + cls.pickings = picking1 | picking2 + cls._fill_stock_for_moves( + picking1.move_lines, in_package=True, location=cls.content_loc + ) + cls._fill_stock_for_moves(picking2.move_lines, location=cls.content_loc) + cls.pickings.action_assign() + cls._simulate_pickings_selected(cls.pickings) + cls.dest_location = ( + cls.env["stock.location"] + .sudo() + .create( + { + "name": "Sub Shelf 1", + "barcode": "subshelf1", + "location_id": cls.shelf1.id, + } + ) + ) + + def test_set_destination_package_wrong_parameters(self): + """Wrong 'location' and 'package_level_id' parameters, redirect the + user to the 'start' screen. + """ + package_level = self.picking1.package_level_ids[0] + response = self.service.dispatch( + "set_destination_package", + params={ + "location_id": 1234567890, # Doesn't exist + "package_level_id": package_level.id, + "barcode": "TEST", + }, + ) + self.assert_response_start( + response, message=self.service.msg_store.record_not_found() + ) + response = self.service.dispatch( + "set_destination_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": 1234567890, # Doesn't exist + "barcode": "TEST", + }, + ) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + def test_set_destination_package_dest_location_nok(self): + """Scanned destination location not valid, redirect to 'scan_destination'.""" + package_level = self.picking1.package_level_ids[0] + # Unknown destination location + response = self.service.dispatch( + "set_destination_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + "barcode": "UNKNOWN_LOCATION", + }, + ) + self.assert_response_scan_destination( + response, package_level, message=self.service.msg_store.no_location_found(), + ) + # Destination location not allowed + customer_location = self.env.ref("stock.stock_location_customers") + customer_location.sudo().barcode = "CUSTOMER" + response = self.service.dispatch( + "set_destination_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + "barcode": customer_location.barcode, + }, + ) + self.assert_response_scan_destination( + response, + package_level, + message=self.service.msg_store.dest_location_not_allowed(), + ) + + def test_set_destination_package_dest_location_to_confirm(self): + """Scanned destination location valid, but need a confirmation.""" + package_level = self.picking1.package_level_ids[0] + response = self.service.dispatch( + "set_destination_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + "barcode": self.env.ref("stock.stock_location_14").barcode, + }, + ) + self.assert_response_scan_destination( + response, + package_level, + message=self.service.msg_store.need_confirmation(), + confirmation_required=True, + ) + + def test_set_destination_package_dest_location_ok(self): + """Scanned destination location valid, moves set to done.""" + package_level = self.picking1.package_level_ids[0] + response = self.service.dispatch( + "set_destination_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + "barcode": self.dest_location.barcode, + }, + ) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, + move_lines.mapped("picking_id"), + message=self.service.msg_store.location_content_transfer_item_complete( + self.dest_location + ), + ) + for move in package_level.move_line_ids.mapped("move_id"): + self.assertEqual(move.state, "done") + + def test_set_destination_line_wrong_parameters(self): + """Wrong 'location' and 'move_line_id' parameters, redirect the + user to the 'start' screen. + """ + move_line = self.picking2.move_line_ids[0] + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": 1234567890, # Doesn't exist + "move_line_id": move_line.id, + "quantity": move_line.product_uom_qty, + "barcode": "TEST", + }, + ) + self.assert_response_start( + response, message=self.service.msg_store.record_not_found() + ) + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": 1234567890, # Doesn't exist + "quantity": move_line.product_uom_qty, + "barcode": "TEST", + }, + ) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + def test_set_destination_line_dest_location_nok(self): + """Scanned destination location not valid, redirect to 'scan_destination'.""" + move_line = self.picking2.move_line_ids[0] + # Unknown destination location + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": move_line.id, + "quantity": move_line.product_uom_qty, + "barcode": "UNKNOWN_LOCATION", + }, + ) + self.assert_response_scan_destination( + response, move_line, message=self.service.msg_store.no_location_found(), + ) + # Destination location not allowed + customer_location = self.env.ref("stock.stock_location_customers") + customer_location.sudo().barcode = "CUSTOMER" + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": move_line.id, + "quantity": move_line.product_uom_qty, + "barcode": customer_location.barcode, + }, + ) + self.assert_response_scan_destination( + response, + move_line, + message=self.service.msg_store.dest_location_not_allowed(), + ) + + def test_set_destination_line_dest_location_to_confirm(self): + """Scanned destination location valid, but need a confirmation.""" + move_line = self.picking2.move_line_ids[0] + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": move_line.id, + "quantity": move_line.product_uom_qty, + "barcode": self.env.ref("stock.stock_location_14").barcode, + }, + ) + self.assert_response_scan_destination( + response, + move_line, + message=self.service.msg_store.need_confirmation(), + confirmation_required=True, + ) + + def test_set_destination_line_dest_location_ok(self): + """Scanned destination location valid, moves set to done.""" + move_line = self.picking2.move_line_ids[0] + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": move_line.id, + "quantity": move_line.product_uom_qty, + "barcode": self.dest_location.barcode, + }, + ) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, + move_lines.mapped("picking_id"), + message=self.service.msg_store.location_content_transfer_item_complete( + self.dest_location + ), + ) + self.assertEqual(move_line.move_id.state, "done") + self.assertEqual(move_line.picking_id.state, "assigned") + + def test_set_destination_line_partial_qty(self): + """Scanned destination location with partial qty, but related moves + has to be splitted. + """ + move_line_c = self.picking2.move_line_ids.filtered( + lambda m: m.product_id == self.product_c + ) + self.assertEqual(move_line_c.product_uom_qty, 10) + self.assertEqual(move_line_c.qty_done, 10) + # Scan partial qty (6/10) + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": move_line_c.id, + "quantity": move_line_c.product_uom_qty - 4, # Scan 6 qty + "barcode": self.dest_location.barcode, + }, + ) + # Check move line data + self.assertEqual(move_line_c.move_id.product_uom_qty, 6) + self.assertEqual(move_line_c.product_uom_qty, 0) + self.assertEqual(move_line_c.qty_done, 6) + self.assertEqual(move_line_c.state, "done") + # Check the new move created to handle the remaining qty + move_product_c_splitted = self.picking2.move_lines.filtered( + lambda m: m.product_id == self.product_c and m.state == "assigned" + ) + self.assertEqual(move_product_c_splitted.state, "assigned") + self.assertEqual(move_product_c_splitted.product_id, self.product_c) + self.assertEqual(move_product_c_splitted.product_uom_qty, 4) + self.assertEqual(move_product_c_splitted.move_line_ids.product_uom_qty, 4) + self.assertEqual(move_product_c_splitted.move_line_ids.qty_done, 4) + # Check the response + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, + move_lines.mapped("picking_id"), + message=self.service.msg_store.location_content_transfer_item_complete( + self.dest_location + ), + ) + self.assertEqual(move_line_c.move_id.state, "done") + # Scan remaining qty (4/10) + remaining_move_line_c = move_product_c_splitted.move_line_ids + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": remaining_move_line_c.id, + "quantity": remaining_move_line_c.product_uom_qty, + "barcode": self.dest_location.barcode, + }, + ) + # Check move line data + self.assertEqual(remaining_move_line_c.move_id.product_uom_qty, 4) + self.assertEqual(remaining_move_line_c.product_uom_qty, 0) + self.assertEqual(remaining_move_line_c.qty_done, 4) + self.assertEqual(remaining_move_line_c.state, "done") + # All move lines related to product_c are now done + moves_product_c = self.picking2.move_lines.filtered( + lambda m: m.product_id == self.product_c + ) + moves_product_c_done = all(move.state == "done" for move in moves_product_c) + self.assertTrue(moves_product_c_done) + moves_product_c_qty_done = sum([move.quantity_done for move in moves_product_c]) + self.assertEqual(moves_product_c_qty_done, 10) + # The picking is still not done as product_d hasn't been processed + self.assertEqual(self.picking2.state, "assigned") + # Let scan product_d quantity and check picking state + move_line_d = self.picking2.move_line_ids.filtered( + lambda m: m.product_id == self.product_d + ) + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": move_line_d.id, + "quantity": move_line_d.product_uom_qty, + "barcode": self.dest_location.barcode, + }, + ) + self.assertEqual(move_line_d.move_id.product_uom_qty, 10) + self.assertEqual(move_line_d.product_uom_qty, 0) + self.assertEqual(move_line_d.qty_done, 10) + self.assertEqual(move_line_d.state, "done") + self.assertEqual(self.picking2.state, "done") + + +class LocationContentTransferSetDestinationXSpecialCase( + LocationContentTransferCommonCase +): + """Tests for endpoint used from scan_destination (special cases) + + * /set_destination_package + * /set_destination_line + + """ + + @classmethod + def setUpClassBaseData(cls): + super().setUpClassBaseData() + products = cls.product_a + for product in products: + cls.env["stock.putaway.rule"].sudo().create( + { + "product_id": product.id, + "location_in_id": cls.stock_location.id, + "location_out_id": cls.shelf1.id, + } + ) + + cls.picking = cls._create_picking( + lines=[(cls.product_a, 10), (cls.product_b, 10)] + ) + cls.move_product_a = cls.picking.move_lines.filtered( + lambda m: m.product_id == cls.product_a + ) + cls.move_product_b = cls.picking.move_lines.filtered( + lambda m: m.product_id == cls.product_b + ) + # Change the initial demand of product_a to get two move lines for + # reserved qties: + # - 10 from the package + # - 5 from the qty without package + cls._fill_stock_for_moves( + cls.move_product_a, in_package=True, location=cls.content_loc + ) + cls.move_product_a.product_uom_qty = 15 + cls._update_qty_in_location( + cls.picking.location_id, cls.product_a, 5, + ) + # Put product_b quantities in two different source locations to get + # two stock move lines (6 and 4 to satisfy 10 qties) + cls._update_qty_in_location(cls.picking.location_id, cls.product_b, 6) + cls._update_qty_in_location(cls.content_loc, cls.product_b, 4) + # Reserve quantities + cls.picking.action_assign() + cls._simulate_pickings_selected(cls.picking) + cls.dest_location = ( + cls.env["stock.location"] + .sudo() + .create( + { + "name": "Sub Shelf 1", + "barcode": "subshelf1", + "location_id": cls.shelf1.id, + } + ) + ) + + def test_set_destination_package_split_move(self): + """Scanned destination location valid for a package, but related moves + has to be splitted because it is linked to additional move lines. + """ + self.assertEqual(len(self.picking.move_lines), 2) + self.assertEqual(len(self.move_product_a.move_line_ids), 2) + package_level = self.picking.package_level_ids[0] + response = self.service.dispatch( + "set_destination_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + "barcode": self.dest_location.barcode, + }, + ) + # Check the picking data + self.assertEqual(package_level.location_dest_id, self.dest_location) + for move_line in package_level.move_line_ids: + self.assertEqual(move_line.location_dest_id, self.dest_location) + moves_product_a = self.picking.move_lines.filtered( + lambda m: m.product_id == self.product_a + ) + self.assertEqual(len(self.picking.move_lines), 3) + self.assertEqual(len(moves_product_a), 2) + for move in moves_product_a: + self.assertEqual(len(move.move_line_ids), 1) + move_lines_wo_pkg = self.picking.move_line_ids_without_package + move_lines_wo_pkg_states = set(move_lines_wo_pkg.mapped("state")) + self.assertEqual(len(move_lines_wo_pkg_states), 1) + self.assertEqual(move_lines_wo_pkg_states.pop(), "assigned") + self.assertEqual(self.picking.package_level_ids.state, "done") + # Check the response + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, + move_lines.mapped("picking_id"), + message=self.service.msg_store.location_content_transfer_item_complete( + self.dest_location + ), + ) + + def test_set_destination_line_split_move(self): + """Scanned destination location valid for a move line, but related moves + has to be splitted because it is linked to additional move lines. + """ + self.assertEqual(len(self.picking.move_lines), 2) + self.assertEqual(len(self.move_product_b.move_line_ids), 2) + move_line = self.move_product_b.move_line_ids.filtered( + lambda ml: ml.product_uom_qty == 6 + ) + response = self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": move_line.id, + "quantity": move_line.product_uom_qty, + "barcode": self.dest_location.barcode, + }, + ) + # Check the picking data + self.assertEqual(self.picking.state, "assigned") + self.assertEqual(move_line.move_id.product_uom_qty, 6) + self.assertEqual(move_line.product_uom_qty, 0) + self.assertEqual(move_line.qty_done, 6) + self.assertEqual(move_line.location_dest_id, self.dest_location) + moves_product_b = self.picking.move_lines.filtered( + lambda m: m.product_id == self.product_b + ) + self.assertEqual(len(self.picking.move_lines), 3) + self.assertEqual(len(moves_product_b), 2) + for move in moves_product_b: + self.assertEqual(len(move.move_line_ids), 1) + move_lines_wo_pkg = self.picking.move_line_ids_without_package + move_lines_wo_pkg_states = set(move_lines_wo_pkg.mapped("state")) + self.assertEqual(len(move_lines_wo_pkg_states), 2) + self.assertIn("assigned", move_lines_wo_pkg_states) + self.assertIn("done", move_lines_wo_pkg_states) + self.assertEqual(move_line.state, "done") + remaining_move = self.picking.move_lines.filtered( + lambda m: move_line.move_id != m and m.product_id == self.product_b + ) + self.assertEqual(remaining_move.state, "assigned") + self.assertEqual(remaining_move.product_uom_qty, 4) + self.assertEqual(remaining_move.move_line_ids.product_uom_qty, 4) + self.assertEqual(remaining_move.move_line_ids.qty_done, 4) + # Check the response + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, + move_lines.mapped("picking_id"), + message=self.service.msg_store.location_content_transfer_item_complete( + self.dest_location + ), + ) + # Process the other move lines (lines w/o package + package levels) + # to check the picking state + remaining_move_lines = self.picking.move_line_ids_without_package.filtered( + lambda ml: ml.state == "assigned" + ) + for ml in remaining_move_lines: + self.service.dispatch( + "set_destination_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": ml.id, + "quantity": ml.product_uom_qty, + "barcode": self.dest_location.barcode, + }, + ) + self.assertEqual(self.picking.state, "assigned") + package_level = self.picking.package_level_ids[0] + self.service.dispatch( + "set_destination_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + "barcode": self.dest_location.barcode, + }, + ) + self.assertEqual(self.picking.state, "done") diff --git a/shopfloor/tests/test_location_content_transfer_single.py b/shopfloor/tests/test_location_content_transfer_single.py new file mode 100644 index 0000000000..599b6aabd7 --- /dev/null +++ b/shopfloor/tests/test_location_content_transfer_single.py @@ -0,0 +1,617 @@ +from .test_location_content_transfer_base import LocationContentTransferCommonCase + + +class LocationContentTransferSingleCase(LocationContentTransferCommonCase): + """Tests for endpoint used from state start_single + + * /scan_package + * /scan_line + * /postpone_package + * /postpone_line + + """ + + # TODO common with set_destination_all? + @classmethod + def setUpClassBaseData(cls): + super().setUpClassBaseData() + products = cls.product_a + cls.product_b + cls.product_c + cls.product_d + for product in products: + cls.env["stock.putaway.rule"].sudo().create( + { + "product_id": product.id, + "location_in_id": cls.stock_location.id, + "location_out_id": cls.shelf1.id, + } + ) + + cls.product_d.tracking = "lot" + cls.picking1 = picking1 = cls._create_picking( + lines=[(cls.product_a, 10), (cls.product_b, 10)] + ) + cls.picking2 = picking2 = cls._create_picking( + lines=[(cls.product_c, 10), (cls.product_d, 10)] + ) + cls.pickings = picking1 | picking2 + cls._fill_stock_for_moves( + picking1.move_lines, in_package=True, location=cls.content_loc + ) + cls.product_d_lot = cls.env["stock.production.lot"].create( + {"product_id": cls.product_d.id, "company_id": cls.env.company.id} + ) + cls._fill_stock_for_moves(picking2.move_lines[0], location=cls.content_loc) + cls._fill_stock_for_moves( + picking2.move_lines[1], location=cls.content_loc, in_lot=cls.product_d_lot + ) + cls.pickings.action_assign() + cls._simulate_pickings_selected(cls.pickings) + + def _test_scan_package_ok(self, barcode): + package_level = self.picking1.move_line_ids.package_level_id + response = self.service.dispatch( + "scan_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + "barcode": barcode, + }, + ) + self.assert_response_scan_destination(response, package_level) + + def test_scan_package_location_not_found(self): + response = self.service.dispatch( + "scan_package", + params={ + "location_id": 1234567890, # Doesn't exist + "package_level_id": 42, + "barcode": "TEST", + }, + ) + self.assert_response_start( + response, message=self.service.msg_store.record_not_found() + ) + + def test_scan_package_package_ok(self): + package_level = self.picking1.move_line_ids.package_level_id + self._test_scan_package_ok(package_level.package_id.name) + + def test_scan_package_barcode_not_found(self): + package_level = self.picking1.move_line_ids.package_level_id + response = self.service.dispatch( + "scan_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + "barcode": "NOT_FOUND", + }, + ) + self.assert_response_start_single( + response, self.pickings, message=self.service.msg_store.barcode_not_found() + ) + + def test_scan_package_product_ok(self): + # product_a is in the package and anywhere else so it's + # accepted to check we scanned the correct package + self._test_scan_package_ok(self.product_a.barcode) + + def test_scan_package_product_packaging_ok(self): + # product_a is in the package and anywhere else so it's + # accepted to check we scanned the correct package + self._test_scan_package_ok(self.product_a.packaging_ids[0].barcode) + + def test_scan_package_lot_ok(self): + package_level = self.picking1.move_line_ids.package_level_id + line_product_a = package_level.move_line_ids[0] + self.product_a.tracking = "lot" + line_product_a.lot_id = self.env["stock.production.lot"].create( + {"product_id": self.product_a.id, "company_id": self.env.company.id} + ) + # lot of product_a is in the package and anywhere else so it's + # accepted to check we scanned the correct package + self._test_scan_package_ok(line_product_a.lot_id.name) + + def _test_scan_package_nok(self, pickings, barcode, message): + package_level = self.picking1.move_line_ids.package_level_id + response = self.service.dispatch( + "scan_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + "barcode": barcode, + }, + ) + self.assert_response_start_single(response, pickings, message=message) + + def test_scan_package_product_nok_different_package(self): + # add another picking with a package with product a, + # if we scan product A, we can't know for which package it is + picking = self._create_picking(lines=[(self.product_a, 10)]) + self._fill_stock_for_moves( + picking.move_lines, in_package=True, location=self.content_loc + ) + picking.action_assign() + self._simulate_pickings_selected(picking) + self._test_scan_package_nok( + self.pickings | picking, + self.product_a.barcode, + {"message_type": "error", "body": "Scan the package"}, + ) + + def test_scan_package_product_nok_different_line(self): + # add another picking with a raw line with product a, + # if we scan product A, we can't know which line/package we want + picking = self._create_picking(lines=[(self.product_a, 10)]) + self._fill_stock_for_moves(picking.move_lines, location=self.content_loc) + picking.action_assign() + self._simulate_pickings_selected(picking) + self._test_scan_package_nok( + self.pickings | picking, + self.product_a.barcode, + {"message_type": "error", "body": "Scan the package"}, + ) + + def test_scan_package_product_nok_product_tracked(self): + # we scan product_a's barcode but it's tracked by lot + self.product_a.tracking = "lot" + self._test_scan_package_nok( + self.pickings, + self.product_a.barcode, + {"message_type": "error", "body": "Scan the package"}, + ) + + def test_scan_package_lot_nok_different_package(self): + # add another picking with a package with the lot used in our package, + # if we scan the lot, we can't know for which package it is + package_level = self.picking1.move_line_ids.package_level_id + line_product_a = package_level.move_line_ids[0] + self.product_a.tracking = "lot" + line_product_a.lot_id = lot = self.env["stock.production.lot"].create( + {"product_id": self.product_a.id, "company_id": self.env.company.id} + ) + picking = self._create_picking(lines=[(self.product_a, 10)]) + self._fill_stock_for_moves( + picking.move_lines, in_package=True, in_lot=lot, location=self.content_loc + ) + picking.action_assign() + self._simulate_pickings_selected(picking) + self._test_scan_package_nok( + self.pickings | picking, + self.product_a.barcode, + {"message_type": "error", "body": "Scan the package"}, + ) + + def test_scan_package_lot_nok_different_line(self): + # add another picking with a raw line with a lot used in our package, + # if we scan the lot, we can't know which line/package we want + package_level = self.picking1.move_line_ids.package_level_id + line_product_a = package_level.move_line_ids[0] + self.product_a.tracking = "lot" + line_product_a.lot_id = lot = self.env["stock.production.lot"].create( + {"product_id": self.product_a.id, "company_id": self.env.company.id} + ) + picking = self._create_picking(lines=[(self.product_a, 10)]) + self._fill_stock_for_moves( + picking.move_lines, in_lot=lot, location=self.content_loc + ) + picking.action_assign() + self._simulate_pickings_selected(picking) + self._test_scan_package_nok( + self.pickings | picking, + self.product_a.barcode, + {"message_type": "error", "body": "Scan the package"}, + ) + + def test_scan_package_package_level_not_exists(self): + package_level = self.picking1.move_line_ids.package_level_id + package_level_id = package_level.id + package_level.unlink() + response = self.service.dispatch( + "scan_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level_id, + "barcode": self.product_a.barcode, + }, + ) + self.assert_response_start_single( + response, self.pickings, message=self.service.msg_store.record_not_found() + ) + + def _test_scan_line_ok(self, move_line, barcode): + response = self.service.dispatch( + "scan_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": move_line.id, + "barcode": barcode, + }, + ) + self.assert_response_scan_destination(response, move_line) + + def test_scan_line_product_ok(self): + move_line = self.picking2.move_line_ids[0] + # check we selected the good line + self.assertEqual(move_line.product_id, self.product_c) + self._test_scan_line_ok(move_line, self.product_c.barcode) + + def test_scan_line_product_packaging_ok(self): + move_line = self.picking2.move_line_ids[0] + # check we selected the good line + self.assertEqual(move_line.product_id, self.product_c) + self._test_scan_line_ok(move_line, self.product_c.packaging_ids[0].barcode) + + def test_scan_line_lot_ok(self): + move_line = self.picking2.move_line_ids[1] + # check we selected the good line (the one with a lot) + self.assertEqual(move_line.product_id, self.product_d) + self._test_scan_line_ok(move_line, self.product_d_lot.name) + + def _test_scan_line_nok(self, pickings, move_line_id, barcode, message): + response = self.service.dispatch( + "scan_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": move_line_id, + "barcode": barcode, + }, + ) + self.assert_response_start_single(response, pickings, message=message) + + def test_scan_line_product_nok_product_tracked(self): + # we scan product_d's barcode but it's tracked by lot + move_line = self.picking2.move_line_ids[1] + # check we selected the good line (the one with a lot) + self.assertEqual(move_line.product_id, self.product_d) + self._test_scan_line_nok( + self.pickings, + move_line.id, + self.product_d.barcode, + self.service.msg_store.scan_lot_on_product_tracked_by_lot(), + ) + + def test_scan_line_barcode_not_found(self): + move_line = self.picking2.move_line_ids[0] + self._test_scan_line_nok( + self.pickings, + move_line.id, + "NOT_FOUND", + self.service.msg_store.barcode_not_found(), + ) + + def test_scan_line_move_line_not_exists(self): + move_line = self.picking2.move_line_ids[0] + move_line_id = move_line.id + move_line.unlink() + self._test_scan_line_nok( + self.pickings, + move_line_id, + "NOT_FOUND", + self.service.msg_store.record_not_found(), + ) + + def test_postpone_package_wrong_parameters(self): + """Wrong 'location_id' and 'package_level_id' parameters, redirect the + user to the 'start' screen. + """ + package_level = self.picking1.move_line_ids.package_level_id + response = self.service.dispatch( + "postpone_package", + params={ + "location_id": 1234567890, # Doesn't exist + "package_level_id": package_level.id, + }, + ) + self.assert_response_start( + response, message=self.service.msg_store.record_not_found() + ) + response = self.service.dispatch( + "postpone_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": 1234567890, # Doesn't exist + }, + ) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + def test_postpone_package_ok(self): + package_level = self.picking1.move_line_ids.package_level_id + self.assertFalse(package_level.shopfloor_postponed) + response = self.service.dispatch( + "postpone_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + }, + ) + self.assertTrue(package_level.shopfloor_postponed) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + def test_postpone_sorter(self): + move_line = self.picking2.move_line_ids[0] + move_lines = self.service._find_transfer_move_lines(self.content_loc) + pickings = move_lines.mapped("picking_id") + sorter = self.service.actions_for("location_content_transfer.sorter") + sorter.feed_pickings(pickings) + content_sorted1 = list(sorter) + self.service.dispatch( + "postpone_line", + params={"location_id": self.content_loc.id, "move_line_id": move_line.id}, + ) + sorter.sort() + content_sorted2 = list(sorter) + self.assertTrue(content_sorted1 != content_sorted2) + + def test_postpone_line_wrong_parameters(self): + """Wrong 'location_id' and 'move_line_id' parameters, redirect the + user to the 'start' screen. + """ + move_line = self.picking2.move_line_ids[0] + response = self.service.dispatch( + "postpone_line", + params={ + "location_id": 1234567890, # Doesn't exist + "move_line_id": move_line.id, + }, + ) + self.assert_response_start( + response, message=self.service.msg_store.record_not_found() + ) + response = self.service.dispatch( + "postpone_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": 1234567890, # Doesn't exist + }, + ) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + def test_postpone_line_ok(self): + move_line = self.picking2.move_line_ids[0] + self.assertFalse(move_line.shopfloor_postponed) + response = self.service.dispatch( + "postpone_line", + params={"location_id": self.content_loc.id, "move_line_id": move_line.id}, + ) + self.assertTrue(move_line.shopfloor_postponed) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + def test_stock_out_package_wrong_parameters(self): + """Wrong 'location_id' and 'package_level_id' parameters, redirect the + user to the 'start' screen. + """ + package_level = self.picking1.move_line_ids.package_level_id + response = self.service.dispatch( + "stock_out_package", + params={ + "location_id": 1234567890, # Doesn't exist + "package_level_id": package_level.id, + }, + ) + self.assert_response_start( + response, message=self.service.msg_store.record_not_found() + ) + response = self.service.dispatch( + "stock_out_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": 1234567890, # Doesn't exist + }, + ) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + def test_stock_out_package_ok(self): + """Declare a stock out on a package_level.""" + package_level = self.picking1.move_line_ids.package_level_id + response = self.service.dispatch( + "stock_out_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + }, + ) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + def test_stock_out_line_wrong_parameters(self): + """Wrong 'location_id' and 'move_line_id' parameters, redirect the + user to the 'start' screen. + """ + move_line = self.picking2.move_line_ids[0] + response = self.service.dispatch( + "stock_out_line", + params={ + "location_id": 1234567890, # Doesn't exist + "move_line_id": move_line.id, + }, + ) + self.assert_response_start( + response, message=self.service.msg_store.record_not_found() + ) + response = self.service.dispatch( + "stock_out_line", + params={ + "location_id": self.content_loc.id, + "move_line_id": 1234567890, # Doesn't exist + }, + ) + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + +class LocationContentTransferSingleSpecialCase(LocationContentTransferCommonCase): + """Tests for endpoint used from state start_single (special cases) + + * /stock_out_package + * /stock_out_line + + """ + + @classmethod + def setUpClassBaseData(cls): + super().setUpClassBaseData() + products = cls.product_a | cls.product_b + for product in products: + cls.env["stock.putaway.rule"].sudo().create( + { + "product_id": product.id, + "location_in_id": cls.stock_location.id, + "location_out_id": cls.shelf1.id, + } + ) + + cls.picking = cls._create_picking( + lines=[(cls.product_a, 10), (cls.product_b, 10)] + ) + cls.move_product_a = cls.picking.move_lines.filtered( + lambda m: m.product_id == cls.product_a + ) + cls.move_product_b = cls.picking.move_lines.filtered( + lambda m: m.product_id == cls.product_b + ) + # Change the initial demand of product_a to get two move lines for + # reserved qties: + # - 10 from the package + # - 5 from the qty without package + cls._fill_stock_for_moves( + cls.move_product_a, in_package=True, location=cls.content_loc + ) + cls.move_product_a.product_uom_qty = 15 + cls._update_qty_in_location( + cls.content_loc, cls.product_a, 5, + ) + # Put product_b quantities in two different source locations to get + # two stock move lines (6 and 4 to satisfy 10 qties) + cls._update_qty_in_location(cls.picking.location_id, cls.product_b, 6) + cls._update_qty_in_location(cls.content_loc, cls.product_b, 4) + # Reserve quantities + cls.picking.action_assign() + cls._simulate_pickings_selected(cls.picking) + + def test_stock_out_package_split_move(self): + """Declare a stock out on a package_level related to moves containing + other unrelated move lines. + """ + package_level = self.picking.move_line_ids.package_level_id + self.assertEqual(self.product_a.qty_available, 15) + response = self.service.dispatch( + "stock_out_package", + params={ + "location_id": self.content_loc.id, + "package_level_id": package_level.id, + }, + ) + # Check the picking data + self.assertFalse(package_level.exists()) + moves_product_a = self.picking.move_lines.filtered( + lambda m: m.product_id == self.product_a + ) + self.assertEqual(len(moves_product_a), 2) + move_product_a = moves_product_a.filtered( + lambda m: m.state not in ("cancel", "done") + ) + self.assertEqual(len(move_product_a), 1) + self.assertEqual(move_product_a.state, "assigned") + self.assertEqual(len(move_product_a.move_line_ids), 1) + # Check the inventories + stock_issue_inventory = self.env["stock.inventory"].search( + [ + ("line_ids.location_id", "=", self.content_loc.id), + ("line_ids.product_id", "=", self.product_a.id), + ("state", "=", "done"), + ] + ) + self.assertTrue(stock_issue_inventory) + stock_issue_inventory_line = stock_issue_inventory.line_ids.filtered( + lambda l: l.product_id == self.product_a + ) + # 5/15 remaining + self.assertEqual(stock_issue_inventory_line.product_qty, 0) + self.assertEqual(self.product_a.qty_available, 5) + control_inventory = self.env["stock.inventory"].search( + [ + ("location_ids", "in", self.content_loc.id), + ("product_ids", "in", self.product_a.id), + ("state", "in", ("draft", "confirm")), + ] + ) + self.assertTrue(control_inventory) + # Check the response + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) + + def test_stock_out_line_split_move(self): + """Declare a stock out on a move line related to moves containing + other move lines. + """ + self.assertEqual(len(self.picking.move_lines), 2) + self.assertEqual(len(self.move_product_b.move_line_ids), 2) + move_line = self.move_product_b.move_line_ids.filtered( + lambda ml: ml.product_uom_qty == 4 # 4/10 to stock out + ) + self.assertEqual(self.product_b.qty_available, 10) + response = self.service.dispatch( + "stock_out_line", + params={"location_id": self.content_loc.id, "move_line_id": move_line.id}, + ) + # Check the picking data + self.assertFalse(move_line.exists()) + moves_product_b = self.picking.move_lines.filtered( + lambda m: m.product_id == self.product_b + ) + self.assertEqual(len(moves_product_b), 2) + move_product_b = moves_product_b.filtered( + lambda m: m.state not in ("cancel", "done") + ) + self.assertEqual(len(move_product_b), 1) + self.assertEqual(move_product_b.state, "assigned") + self.assertEqual(len(move_product_b.move_line_ids), 1) + # Check the inventories + stock_issue_inventory = self.env["stock.inventory"].search( + [ + ("line_ids.location_id", "=", self.content_loc.id), + ("line_ids.product_id", "=", self.product_b.id), + ("state", "=", "done"), + ] + ) + self.assertTrue(stock_issue_inventory) + stock_issue_inventory_line = stock_issue_inventory.line_ids.filtered( + lambda l: l.product_id == self.product_b + ) + # 0/4 remaining in the move line's source location + self.assertEqual(stock_issue_inventory_line.product_qty, 0) + # 6/10 remaining elsewhere in the stock + self.assertEqual(self.product_b.qty_available, 6) + control_inventory = self.env["stock.inventory"].search( + [ + ("location_ids", "in", self.content_loc.id), + ("product_ids", "in", self.product_b.id), + ("state", "in", ("draft", "confirm")), + ] + ) + self.assertTrue(control_inventory) + # Check the response + move_lines = self.service._find_transfer_move_lines(self.content_loc) + self.assert_response_start_single( + response, move_lines.mapped("picking_id"), + ) diff --git a/shopfloor/tests/test_location_content_transfer_start.py b/shopfloor/tests/test_location_content_transfer_start.py new file mode 100644 index 0000000000..66786e1023 --- /dev/null +++ b/shopfloor/tests/test_location_content_transfer_start.py @@ -0,0 +1,225 @@ +from .test_location_content_transfer_base import LocationContentTransferCommonCase + + +class LocationContentTransferStartCase(LocationContentTransferCommonCase): + """Tests for start state and recover + + Endpoints: + + * /start_or_recover + * /scan_location + """ + + @classmethod + def setUpClassBaseData(cls): + super().setUpClassBaseData() + products = cls.product_a + cls.product_b + cls.product_c + cls.product_d + for product in products: + cls.env["stock.putaway.rule"].sudo().create( + { + "product_id": product.id, + "location_in_id": cls.stock_location.id, + "location_out_id": cls.shelf1.id, + } + ) + + cls.picking1 = picking1 = cls._create_picking( + lines=[(cls.product_a, 10), (cls.product_b, 10)] + ) + cls.picking2 = picking2 = cls._create_picking( + lines=[(cls.product_c, 10), (cls.product_d, 10)] + ) + cls.pickings = picking1 | picking2 + cls._fill_stock_for_moves( + picking1.move_lines, in_package=True, location=cls.content_loc + ) + cls._fill_stock_for_moves(picking2.move_lines, location=cls.content_loc) + cls.pickings.action_assign() + + def test_start_fresh(self): + """Start a fresh session when there is no transfer to recover""" + response = self.service.dispatch("start_or_recover", params={}) + self.assert_response(response, next_state="start") + + def test_start_recover_destination_all(self): + """Recover transfers, all move lines have the same destination""" + self._simulate_pickings_selected(self.pickings) + # all lines go to the same destination (shelf1) + self.assertEqual(len(self.pickings.mapped("move_line_ids.location_dest_id")), 1) + + response = self.service.dispatch("start_or_recover", params={}) + self.assert_response_scan_destination_all( + response, + self.pickings, + message=self.service.msg_store.recovered_previous_session(), + ) + + def test_start_recover_destination_single(self): + """Recover transfers, at least one move line has a different destination""" + self._simulate_pickings_selected(self.pickings) + self.picking1.package_level_ids.location_dest_id = self.shelf2 + # we have different destinations + self.assertEqual(len(self.pickings.mapped("move_line_ids.location_dest_id")), 2) + response = self.service.dispatch("start_or_recover", params={}) + self.assert_response_start_single( + response, + self.pickings, + message=self.service.msg_store.recovered_previous_session(), + ) + + def test_scan_location_not_found(self): + """Scan a location with content to transfer, barcode not found""" + response = self.service.dispatch( + "scan_location", params={"barcode": "NOT_FOUND"} + ) + self.assert_response_start( + response, message=self.service.msg_store.barcode_not_found() + ) + + def test_scan_location_find_content_destination_all(self): + """Scan a location with content to transfer, all dest. identical""" + # all lines go to the same destination (shelf1) + self.assertEqual(len(self.pickings.mapped("move_line_ids.location_dest_id")), 1) + response = self.service.dispatch( + "scan_location", params={"barcode": self.content_loc.barcode} + ) + self.assert_response_scan_destination_all(response, self.pickings) + self.assertRecordValues( + self.pickings, [{"user_id": self.env.uid}, {"user_id": self.env.uid}] + ) + self.assertRecordValues( + self.pickings.move_line_ids, + [ + {"qty_done": 10.0}, + {"qty_done": 10.0}, + {"qty_done": 10.0}, + {"qty_done": 10.0}, + ], + ) + self.assertRecordValues(self.picking1.package_level_ids, [{"is_done": True}]) + + def test_scan_location_find_content_destination_single(self): + """Scan a location with content to transfer, different destinations""" + self.picking1.package_level_ids.location_dest_id = self.shelf2 + # we have different destinations + self.assertEqual(len(self.pickings.mapped("move_line_ids.location_dest_id")), 2) + response = self.service.dispatch( + "scan_location", params={"barcode": self.content_loc.barcode} + ) + self.assert_response_start_single(response, self.pickings) + self.assertRecordValues( + self.pickings, [{"user_id": self.env.uid}, {"user_id": self.env.uid}] + ) + self.assertRecordValues( + self.pickings.move_line_ids, + [ + {"qty_done": 10.0}, + {"qty_done": 10.0}, + {"qty_done": 10.0}, + {"qty_done": 10.0}, + ], + ) + self.assertRecordValues(self.picking1.package_level_ids, [{"is_done": True}]) + + def test_scan_location_different_picking_type(self): + """Content has different picking types, can't move""" + picking_other_type = self._create_picking( + picking_type=self.wh.pick_type_id, lines=[(self.product_a, 10)] + ) + self._fill_stock_for_moves( + picking_other_type.move_lines, location=self.content_loc + ) + picking_other_type.action_assign() + + response = self.service.dispatch( + "scan_location", params={"barcode": self.content_loc.barcode} + ) + self.assert_response_start( + response, + message={ + "message_type": "error", + "body": "This location content can't be moved at once.", + }, + ) + + +class LocationContentTransferStartSpecialCase(LocationContentTransferCommonCase): + """Tests for start state and recover (special cases without setup) + + Endpoints: + + * /start_or_recover + * /scan_location + """ + + def test_scan_location_wrong_picking_type(self): + """Content has different picking type than menu""" + picking = self._create_picking( + picking_type=self.wh.pick_type_id, + lines=[(self.product_a, 10), (self.product_b, 10)], + ) + self._fill_stock_for_moves( + picking.move_lines, in_package=True, location=self.content_loc + ) + picking.action_assign() + response = self.service.dispatch( + "scan_location", params={"barcode": self.content_loc.barcode} + ) + self.assert_response_start( + response, + message={ + "message_type": "error", + "body": "This location content can't be moved using this menu.", + }, + ) + + def test_scan_location_create_moves(self): + """The scanned location has no move lines but has some quants to move.""" + picking_type = self.menu.picking_type_ids + # product_a alone + self.env["stock.quant"]._update_available_quantity( + self.product_a, self.content_loc, 10, + ) + # product_b in a package + package = self.env["stock.quant.package"].create({}) + self.env["stock.quant"]._update_available_quantity( + self.product_b, self.content_loc, 10, package_id=package + ) + # product_c & product_d in a package + package2 = self.env["stock.quant.package"].create({}) + self.env["stock.quant"]._update_available_quantity( + self.product_c, self.content_loc, 5, package_id=package2 + ) + self.env["stock.quant"]._update_available_quantity( + self.product_d, self.content_loc, 5, package_id=package2 + ) + response = self.service.dispatch( + "scan_location", params={"barcode": self.content_loc.barcode} + ) + picking = self.env["stock.picking"].search( + [("picking_type_id", "=", picking_type.id)] + ) + self.assertEqual(len(picking), 1) + self.assert_response_scan_destination_all(response, picking) + move_line_id = response["data"]["scan_destination_all"]["move_lines"][0]["id"] + package_levels = response["data"]["scan_destination_all"]["package_levels"] + self.assertIn(move_line_id, picking.move_line_ids.ids) + self.assertEqual(package_levels[0]["id"], picking.package_level_ids[0].id) + self.assertEqual(package_levels[0]["package_src"]["id"], package.id) + self.assertEqual(package_levels[1]["id"], picking.package_level_ids[1].id) + self.assertEqual(package_levels[1]["package_src"]["id"], package2.id) + # product_a in a move line without package + self.assertEqual( + picking.move_line_ids_without_package.mapped("product_id"), self.product_a + ) + # all other products are in package levels + self.assertEqual( + picking.package_level_ids.mapped("package_id.quant_ids.product_id"), + self.product_b | self.product_c | self.product_d, + ) + # all products are in move lines + self.assertEqual( + picking.move_line_ids.mapped("product_id"), + self.product_a | self.product_b | self.product_c | self.product_d, + ) + self.assertEqual(picking.state, "assigned")