[IMP] mrp: block usage of duplicate serial number

With this commit, the manufacturer will now be noticed earlier in
the process if he produced/consumed a serial number already produced/consumed in a
previous production. Before this commit, an error was triggered as
well but only at the very last step of the production.
This leaded to two issues :
 - The user had to unlock-edit-lock to update the wrong serial number
 - On large production, it was difficult to figure out which was/were
   the product(s) to fix.

The search on previous production is done each time the produce wizard is closed
or once the production is done on a workorder

Task 2002133

addons/mrp/models/mrp_abstract_workorder.py

@@ -299,6 +299,18 @@ def _strict_consumption_check(self):
                 if float_compare(qty_done, qty_to_consume, precision_rounding=rounding) != 0:
                     raise UserError(_('You should consume the quantity of %s defined in the BoM. If you want to consume more or less components, change the consumption setting on the BoM.') % lines[0].product_id.name)
 
+    def _check_sn_uniqueness(self):
+        """ Alert the user if the serial number as already been produced """
+        if self.product_tracking == 'serial' and self.finished_lot_id:
+            sml = self.env['stock.move.line'].search_count([
+                ('lot_id', '=', self.finished_lot_id.id),
+                ('location_id.usage', '=', 'production'),
+                ('qty_done', '=', 1),
+                ('state', '=', 'done')
+            ])
+            if sml:
+                raise UserError(_('This serial number for product %s has already been produced') % self.product_id.name)
+
 
 class MrpAbstractWorkorderLine(models.AbstractModel):
     _name = "mrp.abstract.workorder.line"
@@ -361,9 +373,6 @@ def _update_move_lines(self):
         if self.product_id.tracking != 'none' and not self.lot_id:
             raise UserError(_('Please enter a lot or serial number for %s !' % self.product_id.display_name))
 
-        if self.lot_id and self.product_id.tracking == 'serial' and self.lot_id in self.move_id.move_line_ids.filtered(lambda ml: ml.qty_done).mapped('lot_id'):
-            raise UserError(_('You cannot consume the same serial number twice. Please correct the serial numbers encoded.'))
-
         # Update reservation and quantity done
         for ml in move_lines:
             rounding = ml.product_uom_id.rounding
@@ -446,6 +455,51 @@ def _create_extra_move_lines(self):
 
         return vals_list
 
+    def _check_line_sn_uniqueness(self):
+        """ Alert the user if the line serial number as already been consumed/produced """
+        self.ensure_one()
+        if self.product_tracking == 'serial' and self.lot_id:
+            domain = [
+                ('lot_id', '=', self.lot_id.id),
+                ('qty_done', '=', 1),
+                ('state', '=', 'done')
+            ]
+            # Adapt domain and error message in case of component or byproduct
+            if self[self._get_raw_workorder_inverse_name()]:
+                message = _('The serial number %s used for component %s has already been consumed') % (self.lot_id.name, self.product_id.name)
+                co_prod_move_lines = self._get_production().move_raw_ids.move_line_ids
+                co_prod_wo_lines = self[self._get_raw_workorder_inverse_name()].raw_workorder_line_ids
+                domain_unbuild = domain + [
+                    ('production_id', '=', False),
+                    ('location_id.usage', '=', 'production')
+                ]
+                domain.append(('location_dest_id.usage', '=', 'production'))
+            else:
+                message = _('The serial number %s used for byproduct %s has already been produced') % (self.lot_id.name, self.product_id.name)
+                co_prod_move_lines = self._get_production().move_finished_ids.move_line_ids.filtered(lambda ml: ml.product_id != self._get_production().product_id)
+                co_prod_wo_lines = self[self._get_finished_workoder_inverse_name()].finished_workorder_line_ids
+                domain_unbuild = domain + [
+                    ('production_id', '=', False),
+                    ('location_dest_id.usage', '=', 'production')
+                ]
+                domain.append(('location_id.usage', '=', 'production'))
+
+            # Check presence of same sn in previous productions
+            duplicates = self.env['stock.move.line'].search_count(domain)
+            if duplicates:
+                # Maybe some move lines have been compensated by unbuild
+                duplicates_unbuild = self.env['stock.move.line'].search_count(domain_unbuild)
+                if not (duplicates_unbuild and duplicates - duplicates_unbuild == 0):
+                    raise UserError(message)
+            # Check presence of same sn in current production
+            duplicates = co_prod_move_lines.filtered(lambda ml: ml.qty_done and ml.lot_id == self.lot_id)
+            if duplicates:
+                raise UserError(message)
+            # Check presence of same sn in current wizard/workorder
+            duplicates = co_prod_wo_lines.filtered(lambda wol: wol.lot_id == self.lot_id) - self
+            if duplicates:
+                raise UserError(message)
+
     def _unreserve_order(self):
         """ Unreserve line with lower reserved quantity first """
         self.ensure_one()

addons/mrp/models/mrp_workorder.py

@@ -400,10 +400,13 @@ def record_production(self):
             return True
 
         self.ensure_one()
+        self._check_sn_uniqueness()
         self._check_company()
         if float_compare(self.qty_producing, 0, precision_rounding=self.product_uom_id.rounding) <= 0:
             raise UserError(_('Please set the quantity you are currently producing. It should be different from zero.'))
-
+        if 'check_ids' not in self:
+            for line in self.raw_workorder_line_ids | self.finished_workorder_line_ids:
+                line._check_line_sn_uniqueness()
         # If last work order, then post lots used
         if not self.next_work_order_id:
             self._update_finished_move()

addons/mrp/tests/test_order.py

@@ -974,6 +974,174 @@ def test_product_produce_11(self):
         produce_wizard = produce_form.save()
         produce_wizard.do_produce()
 
+    def test_product_produce_duplicate_1(self):
+        """ produce a finished product tracked by serial number 2 times with the
+        same SN. Check that an error is raised the second time"""
+        mo1, bom, p_final, p1, p2 = self.generate_mo(tracking_final='serial', qty_final=1, qty_base_1=1,)
+
+        produce_form = Form(self.env['mrp.product.produce'].with_context({
+            'active_id': mo1.id,
+            'active_ids': [mo1.id],
+        }))
+        product_produce = produce_form.save()
+        product_produce.action_generate_serial()
+        sn = product_produce.finished_lot_id
+        product_produce.do_produce()
+        mo1.button_mark_done()
+
+        mo_form = Form(self.env['mrp.production'])
+        mo_form.product_id = p_final
+        mo_form.bom_id = bom
+        mo_form.product_qty = 1
+        mo2 = mo_form.save()
+        mo2.action_confirm()
+
+        produce_form = Form(self.env['mrp.product.produce'].with_context({
+            'active_id': mo2.id,
+            'active_ids': [mo2.id],
+        }))
+        produce_form.finished_lot_id = sn
+        product_produce = produce_form.save()
+        with self.assertRaises(UserError):
+            product_produce.do_produce()
+
+    def test_product_produce_duplicate_2(self):
+        """ produce a finished product with component tracked by serial number 2
+        times with the same SN. Check that an error is raised the second time"""
+        mo1, bom, p_final, p1, p2 = self.generate_mo(tracking_base_2='serial', qty_final=1, qty_base_1=1,)
+        sn = self.env['stock.production.lot'].create({
+            'name': 'sn used twice',
+            'product_id': p2.id,
+            'company_id': self.env.company.id,
+        })
+        produce_form = Form(self.env['mrp.product.produce'].with_context({
+            'active_id': mo1.id,
+            'active_ids': [mo1.id],
+        }))
+        with produce_form.raw_workorder_line_ids.edit(0) as line:
+            line.lot_id = sn
+        product_produce = produce_form.save()
+        product_produce.do_produce()
+        mo1.button_mark_done()
+
+        mo_form = Form(self.env['mrp.production'])
+        mo_form.product_id = p_final
+        mo_form.bom_id = bom
+        mo_form.product_qty = 1
+        mo2 = mo_form.save()
+        mo2.action_confirm()
+
+        produce_form = Form(self.env['mrp.product.produce'].with_context({
+            'active_id': mo2.id,
+            'active_ids': [mo2.id],
+        }))
+        with produce_form.raw_workorder_line_ids.edit(0) as line:
+            line.lot_id = sn
+        product_produce = produce_form.save()
+        with self.assertRaises(UserError):
+            product_produce.do_produce()
+
+    def test_product_produce_duplicate_3(self):
+        """ produce a finished product with by-product tracked by serial number 2
+        times with the same SN. Check that an error is raised the second time"""
+        finished_product = self.env['product.product'].create({'name': 'finished product'})
+        byproduct = self.env['product.product'].create({'name': 'byproduct', 'tracking': 'serial'})
+        component = self.env['product.product'].create({'name': 'component'})
+        bom = self.env['mrp.bom'].create({
+            'product_id': finished_product.id,
+            'product_tmpl_id': finished_product.product_tmpl_id.id,
+            'product_uom_id': finished_product.uom_id.id,
+            'product_qty': 1.0,
+            'type': 'normal',
+            'bom_line_ids': [
+                (0, 0, {'product_id': component.id, 'product_qty': 1}),
+            ],
+            'byproduct_ids': [
+                (0, 0, {'product_id': byproduct.id, 'product_qty': 1, 'product_uom_id': byproduct.uom_id.id})
+            ]})
+        mo_form = Form(self.env['mrp.production'])
+        mo_form.product_id = finished_product
+        mo_form.bom_id = bom
+        mo_form.product_qty = 1
+        mo = mo_form.save()
+        mo.action_confirm()
+
+        sn = self.env['stock.production.lot'].create({
+            'name': 'sn used twice',
+            'product_id': byproduct.id,
+            'company_id': self.env.company.id,
+        })
+        produce_form = Form(self.env['mrp.product.produce'].with_context({
+            'active_id': mo.id,
+            'active_ids': [mo.id],
+        }))
+
+        with produce_form.finished_workorder_line_ids.edit(0) as line:
+            line.lot_id = sn
+        product_produce = produce_form.save()
+        product_produce.do_produce()
+        mo.button_mark_done()
+
+        mo_form = Form(self.env['mrp.production'])
+        mo_form.product_id = finished_product
+        mo_form.bom_id = bom
+        mo_form.product_qty = 1
+        mo2 = mo_form.save()
+        mo2.action_confirm()
+
+        produce_form = Form(self.env['mrp.product.produce'].with_context({
+            'active_id': mo2.id,
+            'active_ids': [mo2.id],
+        }))
+        with produce_form.finished_workorder_line_ids.edit(0) as line:
+            line.lot_id = sn
+        product_produce = produce_form.save()
+        with self.assertRaises(UserError):
+            product_produce.do_produce()
+
+    def test_product_produce_duplicate_4(self):
+        """ Consuming the same serial number two times should not give an error if
+        a repair order of the first production has been made before the second one"""
+        mo1, bom, p_final, p1, p2 = self.generate_mo(tracking_base_2='serial', qty_final=1, qty_base_1=1,)
+        sn = self.env['stock.production.lot'].create({
+            'name': 'sn used twice',
+            'product_id': p2.id,
+            'company_id': self.env.company.id,
+        })
+        produce_form = Form(self.env['mrp.product.produce'].with_context({
+            'active_id': mo1.id,
+            'active_ids': [mo1.id],
+        }))
+        with produce_form.raw_workorder_line_ids.edit(0) as line:
+            line.lot_id = sn
+        product_produce = produce_form.save()
+        product_produce.do_produce()
+        mo1.button_mark_done()
+
+        unbuild_form = Form(self.env['mrp.unbuild'])
+        unbuild_form.product_id = p_final
+        unbuild_form.bom_id = bom
+        unbuild_form.product_qty = 1
+        unbuild_form.mo_id = mo1
+        unbuild_order = unbuild_form.save()
+        unbuild_order.action_unbuild()
+
+        mo_form = Form(self.env['mrp.production'])
+        mo_form.product_id = p_final
+        mo_form.bom_id = bom
+        mo_form.product_qty = 1
+        mo2 = mo_form.save()
+        mo2.action_confirm()
+
+        produce_form = Form(self.env['mrp.product.produce'].with_context({
+            'active_id': mo2.id,
+            'active_ids': [mo2.id],
+        }))
+        with produce_form.raw_workorder_line_ids.edit(0) as line:
+            line.lot_id = sn
+        product_produce = produce_form.save()
+        product_produce.do_produce()
+
     def test_product_produce_uom(self):
         """ Produce a finished product tracked by serial number. Set another
         UoM on the bom. The produce wizard should keep the UoM of the product (unit)

addons/mrp/wizard/mrp_product_produce.py

@@ -137,6 +137,7 @@ def _record_production(self):
                     move_id = self.env['stock.move'].create(values)
                 line.move_id = move_id.id
 
+            line._check_line_sn_uniqueness()
         # because of an ORM limitation (fields on transient models are not
         # recomputed by updates in non-transient models), the related fields on
         # this model are not recomputed by the creations above
@@ -146,6 +147,8 @@ def _record_production(self):
         quantity = self.qty_producing
         if float_compare(quantity, 0, precision_rounding=self.product_uom_id.rounding) <= 0:
             raise UserError(_("The production order for '%s' has no quantity specified.") % self.product_id.display_name)
+
+        self._check_sn_uniqueness()
         self._update_finished_move()
         self._update_moves()
         if self.production_id.state == 'confirmed':