diff --git a/stock_move_manual_lot/models/stock_move_line.py b/stock_move_manual_lot/models/stock_move_line.py index 4dc4b7f9fb86..bf32c0331ca7 100644 --- a/stock_move_manual_lot/models/stock_move_line.py +++ b/stock_move_manual_lot/models/stock_move_line.py @@ -1,7 +1,6 @@ # Copyright 2021 Hunki Enterprises BV # Copyright 2021 Opener B.V. # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). -import operator from odoo import _, api, fields, models from odoo.exceptions import UserError from odoo.tools.float_utils import float_compare @@ -19,13 +18,33 @@ class StockMoveLine(models.Model): copy=False, ) + def _store_product_qty(self): + """Ensure consistent values for product_qty in the database + + This module triggers (re)assignment of affected pickings within a + x2many write on a picking's move lines. The quantity to assign depends + on stock.move's reserved_quantity, which depends on the recompute of + stock.move.line's `product_qty` whenever `product_uom_quantity is + written. Because the x2many commands are processed under the scope of + env.norecompute(), we have to take care of this ourselves. + """ + field = self._fields["product_qty"] + mls = self.env.field_todo(field).exists() + for ml in mls: + ml._write({"product_qty": ml.product_qty}) + self.env.remove_todo(field, mls) + @api.multi def unlink(self): """Keep move lines if they are assigned a manual lot in a batch""" if self.env.context.get("manual_lot_move_lines"): to_keep = self.filtered( lambda ml: ml.id in self.env.context["manual_lot_move_lines"]) - to_keep.write({"product_uom_qty": 0, "lot_id": False}) + to_keep.write({ + "lot_id": False, + "manual_lot_id": False, + "product_uom_qty": 0, + }) self -= to_keep if not self: return True @@ -51,12 +70,16 @@ def create(self, vals_list): else: to_super.append(vals) result += super().create(to_super) + if not self.env.recompute: + self._store_product_qty() for vals, this in zip(vals_list, result): this._reserve_manual_lot(vals) return result def write(self, vals): result = super().write(vals) + if "product_uom_qty" in vals and not self.env.recompute: + self._store_product_qty() self._reserve_manual_lot(vals) return result @@ -110,11 +133,9 @@ def _reserve_manual_lot(self, vals): ml_to_ignore=self, ) self.env.cr.execute( - 'select id from stock_picking where write_date = %s', (now,), + 'select distinct picking_id from stock_move where write_date = %s', (now,), ) - updated_picking_ids = list(map( - operator.itemgetter(0), self.env.cr.fetchall(), - )) + updated_picking_ids = [row for row, in self.env.cr.fetchall() if row] this.lot_id = this.manual_lot_id if this.product_qty < product_qty: diff --git a/stock_move_manual_lot/tests/test_stock_move_manual_lot.py b/stock_move_manual_lot/tests/test_stock_move_manual_lot.py index 9e3b2f6a090b..c288bc9ac17a 100644 --- a/stock_move_manual_lot/tests/test_stock_move_manual_lot.py +++ b/stock_move_manual_lot/tests/test_stock_move_manual_lot.py @@ -1,5 +1,6 @@ # Copyright 2021 Opener B.V. # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +from datetime import timedelta from uuid import uuid1 from odoo.exceptions import UserError from odoo.tests.common import TransactionCase @@ -14,32 +15,41 @@ def _create_lot(self): } ) - def _create_quant(self, lot): - return self.env["stock.quant"].create( - { - "product_id": self.product.id, - "quantity": 1, - "location_id": self.picking.location_id.id, - "lot_id": lot.id, - } - ) + def _create_quant(self, lot=None, qty=1): + """Create any number of quants. - def setUp(self): - super().setUp() - self.product = self.env["product.product"].create( - { - "name": "Tracked product", - "tracking": "serial", - "type": "product", - } - ) - self.lot1 = self._create_lot() - self.lot2 = self._create_lot() - self.picking = self.env["stock.picking"].create( + :param lot: create quants with these lot(s). + When None, create lots on the fly. + When empty recordset, create quants without lot. + :param qty: number of quants to create. If more than one lot is + passed, the number of lots overrides this. + """ + if lot and len(lot) > qty: + qty = len(lot) + quants = self.env["stock.quant"] + for i in range(0, qty): + if lot and len(lot) > 0: + use_lot = lot[i] + elif lot is None: + use_lot = self._create_lot() + else: + use_lot = lot + quants += self.env["stock.quant"].create( + { + "product_id": self.product.id, + "quantity": 1, + "location_id": self.location.id, + "lot_id": use_lot.id, + } + ) + return quants + + def _create_picking(self, qty=1, confirm=True, assign=True): + picking = self.env["stock.picking"].create( { - "picking_type_id": self.env.ref("stock.picking_type_out").id, - "location_id": self.env.ref("stock.stock_location_stock").id, - "location_dest_id": self.env.ref("stock.stock_location_customers").id, + "picking_type_id": self.picking_type.id, + "location_id": self.location.id, + "location_dest_id": self.dest_location.id, "move_lines": [ ( 0, @@ -48,81 +58,123 @@ def setUp(self): "name": "testmove", "product_id": self.product.id, "product_uom": self.product.uom_id.id, - "product_uom_qty": 1, + "product_uom_qty": qty, }, ) ], } ) - self.picking.picking_type_id.use_manual_lot_selection = True - self.quant1 = self._create_quant(self.lot1) - self.picking.action_assign() - self.quant2 = self._create_quant(self.lot2) + if confirm: + picking.action_confirm() + if assign: + picking.action_assign() + return picking + + def _backdate_moves(self): + """Moves are selected for reassignment based on their write dates""" + for move in self.env["stock.move"].search([]): + move.write_date -= timedelta(minutes=1) + + def setUp(self): + super().setUp() + self.product = self.env["product.product"].create( + { + "name": "Tracked product", + "tracking": "serial", + "type": "product", + } + ) + self.picking_type = self.env.ref("stock.picking_type_out") + self.picking_type.use_manual_lot_selection = True + self.location = self.env.ref("stock.stock_location_stock") + self.dest_location = self.env.ref("stock.stock_location_customers") def test_01_force_manual_selection(self): """Picking can not be validated without manual lot selection""" - self.picking.move_line_ids.qty_done = self.picking.move_line_ids.product_qty + lot = self._create_quant().lot_id + picking = self._create_picking() + picking.move_line_ids.qty_done = picking.move_line_ids.product_qty with self.assertRaisesRegex( UserError, "Serial" ), self.env.clear_upon_failure(), self.env.cr.savepoint(): - self.picking.button_validate() - self.picking.move_line_ids.manual_lot_id = self.lot1 - self.picking.button_validate() - self.assertEqual(self.picking.state, "done") + picking.button_validate() + picking.move_line_ids.manual_lot_id = lot + picking.button_validate() + self.assertEqual(picking.state, "done") self.assertEqual( self.env["stock.quant"] .search( [ - ("location_id", "=", self.picking.location_dest_id.id), + ("location_id", "=", self.dest_location.id), ("product_id", "=", self.product.id), ] ) .mapped("lot_id"), - self.lot1, + lot, ) def test_02_reassign_reservation(self): """Pickings are rereserved after their lots were fetched on another""" - self.assertEqual(self.quant1.reserved_quantity, 1) - picking2 = self.picking.copy() - self.assertFalse(self.quant2.reserved_quantity, 1) - picking2.action_assign() - self.assertEqual(self.quant2.reserved_quantity, 1) - self.assertEqual(picking2.move_line_ids.lot_id, self.lot2) - self.assertTrue(self.picking.move_line_ids) - picking2.move_line_ids.manual_lot_id = self.lot1 - self.assertTrue(self.picking.move_line_ids) - self.assertEqual(self.picking.move_line_ids.lot_id, self.lot2) + quant1 = self._create_quant() + picking = self._create_picking() + self.assertEqual(quant1.reserved_quantity, 1) + quant2 = self._create_quant() + picking2 = self._create_picking() + self.assertEqual(quant2.reserved_quantity, 1) + self.assertEqual(picking.state, 'assigned') + self.assertEqual(picking2.state, 'assigned') + self.assertEqual(picking.move_line_ids.lot_id, quant1.lot_id) + self.assertEqual(picking2.move_line_ids.lot_id, quant2.lot_id) + self._backdate_moves() + + # set the lot from the other picking + picking2.write({ + 'move_line_ids_without_package': [[1, picking2.move_line_ids.id, { + 'manual_lot_id': quant1.lot_id.id + }]]}) + + # both pickings are still assigned + self.assertEqual(picking.state, 'assigned') + self.assertEqual(picking2.state, 'assigned') + + # the lots are swapped + self.assertEqual(picking.move_line_ids.lot_id, quant2.lot_id) + self.assertEqual(picking2.move_line_ids.lot_id, quant1.lot_id) + + self.assertEqual(quant2.reserved_quantity, 1) + self.assertTrue(picking.move_line_ids) + picking2.move_line_ids.manual_lot_id = quant1.lot_id + self.assertTrue(picking.move_line_ids) + self.assertEqual(picking.move_line_ids.lot_id, quant2.lot_id) with self.assertRaises( UserError ), self.env.clear_upon_failure(), self.env.cr.savepoint(): - self.picking.button_validate() + picking.button_validate() self.assertEqual( - self.picking.move_lines.product_qty, self.picking.move_line_ids.product_qty + picking.move_lines.product_qty, picking.move_line_ids.product_qty ) - self.picking.move_line_ids.update( + picking.move_line_ids.update( dict( - qty_done=self.picking.move_line_ids.product_qty, - manual_lot_id=self.lot2, + qty_done=picking.move_line_ids.product_qty, + manual_lot_id=quant2.lot_id, ) ) quant_domain = [ - ("location_id", "=", self.picking.location_dest_id.id), + ("location_id", "=", self.dest_location.id), ("product_id", "=", self.product.id), ] self.assertFalse(self.env["stock.quant"].search(quant_domain)) - self.picking.button_validate() + picking.button_validate() self.assertEqual( - self.env["stock.quant"].search(quant_domain).mapped("lot_id"), self.lot2 + self.env["stock.quant"].search(quant_domain).lot_id, quant2.lot_id ) def test_03_multiple_lines_flip(self): """Multiple lines (with each other's lots) can be reassigned at once. """ - self.picking.action_cancel() - picking = self.picking.copy() - picking.move_lines.product_uom_qty = 2 - picking.action_assign() + quant1, quant2 = self._create_quant(qty=2) + picking = self._create_picking(2) + self.assertEqual(quant1.reserved_quantity, 1) move_line1, move_line2 = picking.move_line_ids lot1 = move_line1.lot_id lot2 = move_line2.lot_id @@ -149,8 +201,8 @@ def test_03_multiple_lines_flip(self): (1, picking.move_line_ids[1].id, {"manual_lot_id": lot1.id}), ], }) - self.assertEqual(self.quant1.reserved_quantity, 1) - self.assertEqual(self.quant2.reserved_quantity, 1) + self.assertEqual(quant1.reserved_quantity, 1) + self.assertEqual(quant2.reserved_quantity, 1) self.assertTrue(picking.move_line_ids[0].lot_id) self.assertTrue(picking.move_line_ids[1].lot_id) self.assertEqual( @@ -162,48 +214,41 @@ def test_03_multiple_lines_flip(self): picking.button_validate() self.assertEqual(picking.state, "done") - def _setup_multiple_lines(self): - """ - Replace self.picking with a picking with multiple lines, where one - line has a manual lot set - """ - self.picking.action_cancel() - self.picking = self.picking.copy() - self.picking.move_lines.product_uom_qty = 2 - self.picking.action_confirm() - self.picking.action_assign() - line_with_lot = self.picking.move_line_ids[0] - line_with_lot.manual_lot_id = self.lot1 - line_without_lot = self.picking.move_line_ids[1] - return line_with_lot, line_without_lot - def test_03_multiple_lines_move_lot(self): """ Test that moving a manual lot from one line to the other works """ - line_with_lot, line_without_lot = self._setup_multiple_lines() - self.picking.write({ + self._create_quant(qty=2) + picking = self._create_picking(2) + line_with_lot, line_without_lot = picking.move_line_ids + lot1 = line_with_lot.lot_id + line_with_lot.manual_lot_id = lot1 + picking.write({ 'move_line_ids': [ (1, line_without_lot.id, { - 'manual_lot_id': line_with_lot.manual_lot_id.id, + 'manual_lot_id': lot1.id, }), (1, line_with_lot.id, { 'manual_lot_id': False, }), ], }) - self.assertEqual(line_without_lot.lot_id, self.lot1) - other_line = self.picking.move_line_ids - line_without_lot + self.assertEqual(line_without_lot.lot_id, lot1) + other_line = picking.move_line_ids - line_without_lot self.assertTrue(other_line.lot_id) self.assertFalse(other_line.manual_lot_id) self.assertEqual(other_line.product_qty, 1) - def test_03_multiple_lines_assign(self): + def test_04_multiple_lines_assign(self): """ Test that just reassigning a manual lot from one line to the other works """ - line_with_lot, line_without_lot = self._setup_multiple_lines() - self.picking.write({ + self._create_quant(qty=2) + picking = self._create_picking(2) + line_with_lot, line_without_lot = picking.move_line_ids + lot1 = line_with_lot.lot_id + line_with_lot.manual_lot_id = lot1 + picking.write({ 'move_line_ids': [ (1, line_without_lot.id, { 'manual_lot_id': line_with_lot.manual_lot_id.id, @@ -211,21 +256,25 @@ def test_03_multiple_lines_assign(self): # this is what the webclient sends when we don't change # the line (4, line_with_lot.id, False), - ], + ], }) - self.assertEqual(line_without_lot.lot_id, self.lot1) - other_line = self.picking.move_line_ids - line_without_lot + self.assertEqual(line_without_lot.lot_id, lot1) + other_line = picking.move_line_ids - line_without_lot self.assertTrue(other_line.lot_id) self.assertFalse(other_line.manual_lot_id) self.assertEqual(other_line.product_qty, 1) - def test_03_multiple_lines_delete(self): + def test_05_multiple_lines_delete(self): """ Test that reassigning a manual lot from one line to the other works when deleting the first one """ - line_with_lot, line_without_lot = self._setup_multiple_lines() - self.picking.write({ + self._create_quant(qty=2) + picking = self._create_picking(2) + line_with_lot, line_without_lot = picking.move_line_ids + lot1 = line_with_lot.lot_id + line_with_lot.manual_lot_id = lot1 + picking.write({ 'move_line_ids': [ (1, line_without_lot.id, { 'manual_lot_id': line_with_lot.manual_lot_id.id, @@ -233,28 +282,25 @@ def test_03_multiple_lines_delete(self): (2, line_with_lot.id, False), ], }) - self.assertEqual(line_without_lot.lot_id, self.lot1) + self.assertEqual(line_without_lot.lot_id, lot1) self.assertFalse(line_with_lot.exists()) - def test_04_serial_not_in_stock(self): + def test_06_serial_not_in_stock(self): """It is not allowed to assign a serial that is not in stock""" - lot3 = self.env["stock.production.lot"].create({ - "name": "lot3", - "product_id": self.product.id, - }) + self._create_quant() + picking = self._create_picking() with self.assertRaisesRegex(UserError, "than you have in stock"): - self.picking.move_line_ids.manual_lot_id = lot3 + picking.move_line_ids.manual_lot_id = self._create_lot() - def test_05_backorder(self): + def test_07_backorder(self): """An assigned move line on a picking can be left untransferred.""" - self.picking.action_cancel() - picking = self.picking.copy() + picking = self._create_picking(confirm=False) line = picking.move_lines # Create up to 4 lines line.copy() line.copy() line.copy() - self._create_quant(self._create_lot()) + self._create_quant(qty=3) picking.action_assign() # Of the 4 lines, 3 are assigned. ml1, ml2, ml3 = picking.move_line_ids @@ -270,74 +316,135 @@ def test_05_backorder(self): self.assertEqual(backorder.move_lines.product_uom_qty, 3) self.assertEqual(backorder.move_lines.reserved_availability, 2) - def test_06_non_tracked_product(self): + def test_08_non_tracked_product(self): """Non tracked product, but having lots in stock Manual lot is synced with lot_id automatically. If the manual lot is unset, reservation of stock without serial is enforced. """ + lot1, lot2 = self._create_quant(qty=2).mapped("lot_id") + picking = self._create_picking() self.product.tracking = 'none' - self.picking.do_unreserve() - self.picking.action_assign() - self.assertTrue(self.picking.move_line_ids.manual_lot_id) + picking.do_unreserve() + picking.action_assign() + self.assertTrue(picking.move_line_ids.manual_lot_id) self.assertEqual( - self.picking.move_line_ids.manual_lot_id, - self.picking.move_line_ids.lot_id) + picking.move_line_ids.manual_lot_id, + picking.move_line_ids.lot_id) # Lot is synced with manual lot - self.picking.move_line_ids.manual_lot_id = self.lot2 - self.assertEqual(self.picking.move_line_ids.lot_id, self.lot2) + picking.move_line_ids.manual_lot_id = lot2 + self.assertEqual(picking.move_line_ids.lot_id, lot2) # Manual lot is synced with lot - self.picking.move_line_ids.lot_id = self.lot1 - self.assertEqual(self.picking.move_line_ids.manual_lot_id, self.lot1) + picking.move_line_ids.lot_id = lot1 + self.assertEqual(picking.move_line_ids.manual_lot_id, lot1) # Unsetting the serial raises if there is stock without serial. with self.assertRaisesRegex( UserError, "than you have in stock" ), self.env.clear_upon_failure(), self.env.cr.savepoint(): - self.picking.move_line_ids.manual_lot_id = False + picking.move_line_ids.manual_lot_id = False # Create stock without serial. Serial can now be unset self._create_quant(self.env["stock.production.lot"]) - self.picking.move_line_ids.manual_lot_id = False - self.assertFalse(self.picking.move_line_ids.lot_id) - self.assertTrue(self.picking.move_line_ids.product_qty) + picking.move_line_ids.manual_lot_id = False + self.assertFalse(picking.move_line_ids.lot_id) + self.assertTrue(picking.move_line_ids.product_qty) - self.picking.move_line_ids.qty_done = ( - self.picking.move_line_ids.product_qty + picking.move_line_ids.qty_done = ( + picking.move_line_ids.product_qty ) - self.picking.button_validate() - self.assertEqual(self.picking.state, "done") + picking.button_validate() + self.assertEqual(picking.state, "done") - def test_07_unset_manual_lot(self): + def test_09_unset_manual_lot(self): """Unsetting a manual lot does not unreserve the move line.""" - ml = self.picking.move_line_ids + lot1 = self._create_quant(qty=2).mapped("lot_id")[0] + picking = self._create_picking() + ml = picking.move_line_ids ml.qty_done = ml.product_qty - self.picking.move_line_ids.manual_lot_id = self.lot1 - self.picking.move_line_ids.manual_lot_id = False + picking.move_line_ids.manual_lot_id = lot1 + picking.move_line_ids.manual_lot_id = False self.assertEqual(ml.product_qty, 1) with self.assertRaisesRegex( UserError, "Serial" ), self.env.clear_upon_failure(), self.env.cr.savepoint(): - self.picking.button_validate() - self.picking.move_line_ids.manual_lot_id = self.lot1 - self.picking.button_validate() - self.assertEqual(self.picking.state, "done") + picking.button_validate() + picking.move_line_ids.manual_lot_id = lot1 + picking.button_validate() + self.assertEqual(picking.state, "done") - def test_08_no_manual_selection(self): + def test_10_no_manual_selection(self): """Manual lot is kept in sync with lot. Also, picking can be validated if manual lot is not set. """ - self.picking.picking_type_id.use_manual_lot_selection = False - ml = self.picking.move_line_ids + lot2 = self._create_quant(qty=2).mapped("lot_id")[1] + picking = self._create_picking() + self.picking_type.use_manual_lot_selection = False + ml = picking.move_line_ids ml.qty_done = ml.product_qty - self.picking.move_line_ids.lot_id = self.lot2 - self.assertEqual(self.picking.move_line_ids.manual_lot_id, self.lot2) - self.picking.move_line_ids.manual_lot_id = False - self.assertEqual(self.picking.move_line_ids.lot_id, self.lot2) - self.assertFalse(self.picking.move_line_ids.manual_lot_id) - self.assertEqual(self.picking.move_line_ids.lot_id, self.lot2) - self.picking.move_line_ids.qty_done = ( - self.picking.move_line_ids.product_qty) - self.picking.button_validate() - self.assertEqual(self.picking.state, "done") - self.assertEqual(self.picking.move_line_ids.manual_lot_id, self.lot2) + picking.move_line_ids.lot_id = lot2 + self.assertEqual(picking.move_line_ids.manual_lot_id, lot2) + picking.move_line_ids.manual_lot_id = False + self.assertEqual(picking.move_line_ids.lot_id, lot2) + self.assertFalse(picking.move_line_ids.manual_lot_id) + self.assertEqual(picking.move_line_ids.lot_id, lot2) + picking.move_line_ids.qty_done = ( + picking.move_line_ids.product_qty) + picking.button_validate() + self.assertEqual(picking.state, "done") + self.assertEqual(picking.move_line_ids.manual_lot_id, lot2) + + def test_11_no_overassignment(self): + """Overassignment does not occur after assigning manual lots. + + This problem can occur when pickings are reassigned before the values + for stock.move.line's product_qty are stored in the database. + """ + self._create_quant(qty=2) + picking = self._create_picking(3) + self._create_quant(qty=5) + picking2 = self._create_picking(3) + self.assertEqual(picking.move_lines.state, "partially_available") + self.assertEqual(len(picking.move_line_ids), 2) + self.assertEqual(picking2.move_lines.state, "assigned") + self._backdate_moves() + picking2.write({ + "move_line_ids_without_package": [ + ( + 1, picking2.move_line_ids[0].id, + {"manual_lot_id": picking.move_line_ids[0].lot_id.id} + ), + ( + 1, picking2.move_line_ids[1].id, + {"manual_lot_id": picking.move_line_ids[1].lot_id.id} + ), + ], + }) + self.assertEqual(len(picking.move_line_ids), 3) + self.assertEqual(picking.move_lines.state, "assigned") + self.assertEqual(len(picking2.move_line_ids), 3) + self.assertEqual(picking2.move_lines.state, "assigned") + + def test_12_reassign_multiple(self): + """This specific reassignment breaks if product_qty is not up to date. + + Having two assigned pickings with a quantity of two, assigning a lot + from the first picking to the first picking while assigning the + original lot from the assigned move line to the second move line of + the same picking should not raise an error. + """ + self._create_quant(qty=4) + picking = self._create_picking(2) + picking2 = self._create_picking(2) + self._backdate_moves() + vals = [ + (1, picking2.move_line_ids[0].id, + {"manual_lot_id": picking.move_line_ids[0].lot_id.id}), + (1, picking2.move_line_ids[1].id, + {"manual_lot_id": picking2.move_line_ids[0].lot_id.id}), + ] + picking2.write({"move_line_ids_without_package": vals}) + self.assertEqual(len(picking.move_line_ids), 2) + self.assertEqual(picking.move_lines.state, "assigned") + self.assertEqual(len(picking2.move_line_ids), 2) + self.assertEqual(picking2.move_lines.state, "assigned")