Permalink
Browse files

[FIX] stock: block any duplicate serial numbers

Serial numbers should be unique by design. There are currently some checks
to empeach any duplication into transfers, inventory adjustements or production.
However, a global constraint checking accross all quants is missing. Indeed,
nothing blocks you to have the same serial number used in two sub locations of
your stock. You can also deliver a tracked product then produce the same number.

This commit adds a constraint at the stock move lines validation. After the creation of
the quant (as it's tracked by serial number, it should always be a creation and
not an update). We check that the serial number is not present elsewhere except in
location 'supplier', 'production' and 'inventory. Those 3 last one are virtuals ones
and the quantity is most of the time negatives.

This commit also add a correction process for inventory adjutment. If a serial number
is stored in loc1 and the inventory aims to change quantity to 1 for sn1 in loc2.
We raise a correction wizard to warn the user there is a problem and to add in the
inventory some line to balance the new stock moves.

Ex: we produce sn1, we sell it to a customer. The quant situation is the following
Production : -1
Customer   : +1
We create an inventory adjustement to increase the quantity of sn1 in Stock. Without
the commit, the situation whould have been
Production : -1
Inventory  : -1
Customer   : +1
Stock      : +1
What it does is inverting the first movement (stock -> customer) thanks to an inventory line
The situation after the inventory is the following
Production : -1
Stock      : +1

Task : 1924758
  • Loading branch information...
Whenrow committed Feb 11, 2019
1 parent b9902d9 commit 1ec67c5e38fdc24599b8bb82d2c846402f4d7485
@@ -5,6 +5,7 @@
from datetime import datetime, timedelta

from odoo.fields import Datetime as Dt
from odoo.exceptions import UserError
from odoo.addons.mrp.tests.common import TestMrpCommon


@@ -776,3 +777,41 @@ def test_product_produce_uom_2(self):
self.assertEqual(product_produce.workorder_line_ids[0].product_uom_id, unit, 'Should be the product uom so "unit"')
self.assertEqual(product_produce.workorder_line_ids[1].qty_to_consume, 1, 'Should be 1 unit since the tracking is serial and quantity 2.')
self.assertEqual(product_produce.workorder_line_ids[1].product_uom_id, unit, 'should be the product uom so "unit"')

def test_duplicate_serial_1(self):
"""The production of a serial tracked product is forbidden if the serial
number has already be produced"""
mo, bom, p_final, compo1, compo2 = self.generate_mo('serial', qty_final=1, qty_base_1=1, qty_base_2=1)
sn1 = self.env['stock.production.lot'].create({
'name': 'sn1',
'product_id': p_final.id
})
# First production
mo.action_assign()
# produce product
produce_form = Form(self.env['mrp.product.produce'].with_context({
'active_id': mo.id,
'active_ids': [mo.id],
}))
produce_form.lot_id = sn1
product_produce = produce_form.save()
product_produce.do_produce()
mo.button_mark_done()

# move product to another location
quant = self.env['stock.quant']._gather(p_final, self.env.ref('stock.stock_location_stock'))
quant.location_id = self.env.ref('stock.stock_location_customers')

# Second production
mo2 = mo.copy()
mo2.action_confirm()
mo2.action_assign()
produce_form = Form(self.env['mrp.product.produce'].with_context({
'active_id': mo2.id,
'active_ids': [mo2.id],
}))
produce_form.lot_id = sn1
product_produce = produce_form.save()
product_produce.do_produce()
with self.assertRaises(UserError):
mo2.button_mark_done()
@@ -2,7 +2,7 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.

from . import common
from odoo.exceptions import except_orm
from odoo.exceptions import except_orm, UserError
from odoo.tests import Form


@@ -97,11 +97,40 @@ def test_manufacturing_scrap(self):
# Scrap Product Wood with lot.
self.env['stock.scrap'].with_context(active_model='mrp.production', active_id=production_3.id).create({'product_id': self.product_2.id, 'scrap_qty': 1.0, 'product_uom_id': self.product_2.uom_id.id, 'location_id': location_id, 'lot_id': lot_product_2.id, 'production_id': production_3.id})

#Check scrap move is created for this production order.
#TODO: should check with scrap objects link in between
def test_manufacturing_scrap_2(self):
""" Testing to do a scrap of serial tracked consumed materialself.
Then consuming again the same serial number. We should be blocked """

# scrap_move = production_3.move_raw_ids.filtered(lambda x: x.product_id == self.product_2 and x.scrapped)
# self.assertTrue(scrap_move, "There are no any scrap move created for production order.")
mo, bom, p_final, compo1, compo2 = self.generate_mo(tracking_base_1='serial', qty_final=1, qty_base_1=1, qty_base_2=1)
sn1 = self.env['stock.production.lot'].create({
'name': 'sn1',
'product_id': compo1.id
})
stock_inv_compo1 = self.env['stock.inventory'].create({
'name': 'Stock Inventory for Stick',
'filter': 'product',
'product_id': compo1.id,
'line_ids': [
(0, 0, {'product_id': compo1.id, 'product_uom_id': compo1.uom_id.id, 'product_qty': 1, 'prod_lot_id': sn1.id, 'location_id': self.ref('stock.stock_location_14')}),
]})

stock_inv_compo1.action_start()
stock_inv_compo1.action_validate()

# Scrap Product.
scrap_id = self.env['stock.scrap'].with_context(active_model='mrp.production', active_id=mo.id).create({'product_id': compo1.id, 'scrap_qty': 1.0, 'lot_id': sn1.id, 'product_uom_id': compo1.uom_id.id, 'location_id': self.ref('stock.stock_location_14'), 'production_id': mo.id})
scrap_id.do_scrap()

# Try to consume sn1
produce_form = Form(self.env['mrp.product.produce'].with_context({
'active_id': mo.id,
'active_ids': [mo.id],
}))
product_produce = produce_form.save()
product_produce.produce_line_ids.filtered(lambda line: line.product_id == compo1).lot_id = sn1.id
product_produce.do_produce()
with self.assertRaises(UserError):
mo.button_mark_done()


class TestKitPicking(common.TestMrpCommon):
@@ -52,6 +52,7 @@
'wizard/product_replenish_views.xml',
'wizard/stock_track_confirmation_views.xml',
'wizard/stock_package_destination_views.xml',
'wizard/stock_duplicate_warning_views.xml',

'views/res_partner_views.xml',
'views/product_strategy_views.xml',
@@ -88,6 +88,12 @@ def _default_location_id(self):
readonly=True, states={'draft': [('readonly', False)]},
help="Specify Product Category to focus your inventory on a particular Category.")
exhausted = fields.Boolean('Include Exhausted Products', readonly=True, states={'draft': [('readonly', False)]})
show_locations = fields.Boolean(compute='_compute_show_locations')

@api.depends('line_ids')
def _compute_show_locations(self):
group = self.user_has_groups("stock.group_stock_multi_locations")
self.show_locations = any([line.correction_line != False for line in self.line_ids]) or group

@api.one
@api.depends('product_id', 'line_ids.product_qty')
@@ -164,6 +170,46 @@ def action_reset_product_qty(self):
return True

def action_validate(self):
# Check the serial numbers uniquiness
serials = self.line_ids.filtered(lambda l: l.product_id.tracking == 'serial' and l.prod_lot_id and l.theoretical_qty < l.product_qty)
wiz_lines = []
for line in serials:
similar_quant = self.env['stock.quant'].search([
('lot_id', '=', line.prod_lot_id.id),
('location_id', '!=', line.location_id.id),
'|',
('location_id.usage', 'not in', ('supplier', 'inventory', 'production')),
('location_id.scrap_location', '=', True),
('quantity', '>', 0)
])
# Extract the quant to correct
if len(similar_quant) > 1:
same_usage = similar_quant.filtered(lambda q: q.location_id.usage == line.location_id.usage)
if same_usage:
similar_quant = same_usage[0]
else:
similar_quant = similar_quant[0]
# Fill the wizard lines
if similar_quant and similar_quant.quantity - sum(self.line_ids.filtered(lambda line: line.correction_line == 'warning' and line.prod_lot_id == similar_quant.lot_id).mapped('theoretical_qty')) != 0:
wiz_lines.append({
'product_id': similar_quant.product_id.id,
'location_id': similar_quant.location_id.id,
'lot_id': similar_quant.lot_id.id,
'inventory_line_id': line.id,
})
if wiz_lines:
wiz = self.env['stock.duplicate.warning'].create({
'inventory_id': self.id,
'tracking_line_ids': [(0, 0, line) for line in wiz_lines]
})
return {
'name': _('Duplicated serial numbers'),
'type': 'ir.actions.act_window',
'view_mode': 'form',
'res_model': 'stock.duplicate.warning',
'target': 'new',
'res_id': wiz.id,
}
inventory_lines = self.line_ids.filtered(lambda l: l.product_id.tracking in ['lot', 'serial'] and not l.prod_lot_id and l.theoretical_qty != l.product_qty)
lines = self.line_ids.filtered(lambda l: float_compare(l.product_qty, 1, precision_rounding=l.product_uom_id.rounding) > 0 and l.product_id.tracking == 'serial' and l.prod_lot_id)
if inventory_lines and not lines:
@@ -356,6 +402,7 @@ class InventoryLine(models.Model):
inventory_location_id = fields.Many2one(
'stock.location', 'Inventory Location', related='inventory_id.location_id', related_sudo=False, readonly=False)
product_tracking = fields.Selection('Tracking', related='product_id.tracking', readonly=True)
correction_line = fields.Selection([('warning', 'Warning'), ('danger', 'Danger')], default=False)

@api.one
@api.depends('location_id', 'product_id', 'package_id', 'product_uom_id', 'company_id', 'prod_lot_id', 'partner_id')
@@ -399,6 +446,9 @@ def create(self, vals_list):

@api.multi
def write(self,vals):
# reset decoration style
if self.correction_line == 'danger' and not vals.get('correction_line'):
vals['correction_line'] = False
res = super(InventoryLine, self).write(vals)
self._check_no_duplicate_line()
return res
@@ -2,6 +2,7 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.

from collections import Counter
from itertools import groupby

from odoo import api, fields, models, _
from odoo.addons import decimal_precision as dp
@@ -316,7 +317,7 @@ def write(self, vals):
if not location_id.should_bypass_reservation():
ml._free_reservation(ml.product_id, location_id, untracked_qty, lot_id=False, package_id=package_id, owner_id=owner_id)
Quant._update_available_quantity(product_id, location_dest_id, quantity, lot_id=lot_id, package_id=result_package_id, owner_id=owner_id, in_date=in_date)

ml._check_duplicates()
# Unreserve and reserve following move in order to have the real reserved quantity on move_line.
next_moves |= ml.move_id.move_dest_ids.filtered(lambda move: move.state not in ('done', 'cancel'))

@@ -451,12 +452,61 @@ def _action_done(self, cancel_backorder=False):
Quant._update_available_quantity(ml.product_id, ml.location_id, taken_from_untracked_qty, lot_id=ml.lot_id, package_id=ml.package_id, owner_id=ml.owner_id)
Quant._update_available_quantity(ml.product_id, ml.location_dest_id, quantity, lot_id=ml.lot_id, package_id=ml.result_package_id, owner_id=ml.owner_id, in_date=in_date)
done_ml |= ml

ml._check_duplicates()

# Reset the reserved quantity as we just moved it to the destination location.
(self - ml_to_delete).with_context(bypass_reservation_update=True).write({
'product_uom_qty': 0.00,
'date': fields.Datetime.now(),
})

def _check_duplicates(self):
# Unicity of serial tracked product. Search in quants if the sn is
# already present in stock or delivered to a customer
if self.tracking == 'serial' and self.lot_id:
similar_quant = self.env['stock.quant'].search([
('lot_id', '=', self.lot_id.id),
'|',
('location_id.usage', 'not in', ('supplier', 'inventory', 'production')),
('location_id.scrap_location', '=', True)
])
error_message = _('The serial number %s for product %s is already in another location. Correct your inventory with an inventory adjustment before validating this product move.' % (self.lot_id.name, self.product_id.display_name))

# Check if SN has been scrapped
scrapped = any([quant.location_id.scrap_location for quant in similar_quant])
if scrapped and not self.location_dest_id.scrap_location:
raise UserError(error_message)
if similar_quant:
# the following step will group the quants by location usage
# if similar_quant =
# 1 in customer in PACK1
# 1 in customer in PACK2
# -1 in stock in PACK3
# -1 in output without pack
# 1 in stock without pack
# The group_by becomes
# 2 in customer
# -1 in internal (+1 -1 -1 = -1)

# First, we sort the quant by location usage
sq = sorted([(q.location_id.usage, q.quantity) for q in similar_quant], key=lambda x: x[0])
# -> [('internal', 1), ('customer', 1), ('internal', -1), ('internal', -1), ('customer', 1)]

# Second, we group by location usage and sum the quantity in the groups
sq = [(x[0], sum([y[1] for y in x[1]])) for x in groupby(sq, key=lambda x:x[0])]
# -> [('internal', [('internal',1), ('internal', -1), ('internal', 1)]), ('customer', [('customer', 1), ('customer', 1)])]
# -> [('internal', -1), ('customer', 2)]

# Third, we search group with quantity > 1
duplicates = any([x[1] > 1 for x in sq])
# -> duplicates = True (for customer usage)

# Fourth we sum the total quantity
quantity = sum([x[1] >= 1 for x in sq])
if duplicates or quantity > 1:
raise UserError(error_message)

def _reservation_is_updatable(self, quantity, reserved_quant):
self.ensure_one()
if (self.product_id.tracking != 'serial' and
@@ -73,12 +73,6 @@ def check_product_id(self):
if any(elem.product_id.type != 'product' for elem in self):
raise ValidationError(_('Quants cannot be created for consumables or services.'))

@api.constrains('quantity')
def check_quantity(self):
for quant in self:
if float_compare(quant.quantity, 1, precision_rounding=quant.product_uom_id.rounding) > 0 and quant.lot_id and quant.product_id.tracking == 'serial':
raise ValidationError(_('A serial number should only be linked to a single product.'))

@api.constrains('location_id')
def check_location_id(self):
for quant in self:
@@ -43,6 +43,7 @@ def setUp(self):
self.productC = self.ProductObj.create({'name': 'Product C', 'type': 'product'})
self.productD = self.ProductObj.create({'name': 'Product D', 'type': 'product'})
self.productE = self.ProductObj.create({'name': 'Product E', 'type': 'product'})
self.productS = self.ProductObj.create({'name': 'Product S', 'type': 'product', 'tracking': 'serial'})

# Configure unit of measure.
self.uom_kg = self.env['uom.uom'].search([('category_id', '=', self.categ_kgm), ('uom_type', '=', 'reference')], limit=1)
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.exceptions import ValidationError
from odoo.exceptions import UserError
from odoo.tests.common import TransactionCase


@@ -102,7 +102,7 @@ def test_inventory_3(self):
inventory.line_ids.prod_lot_id = lot1
inventory.line_ids.product_qty = 2

with self.assertRaises(ValidationError):
with self.assertRaises(UserError):
inventory.action_validate()

def test_inventory_4(self):
@@ -297,3 +297,64 @@ def test_inventory_7(self):
self.assertEqual(len(inventory.line_ids), 1)
self.assertEqual(inventory.line_ids.theoretical_qty, 2)

def test_duplicate_serial_1(self):
# add sn1 for a serial tracked products in stock
sn1 = self.env['stock.production.lot'].create({
'product_id': self.product2.id,
'name': 'SN1'
})
inventory = self.env['stock.inventory'].create({
'name': 'serial product',
'filter': 'product',
'location_id': self.stock_location.id,
'product_id': self.product2.id,
'exhausted': True, # should be set by an onchange
})
inventory.action_start()
inventory.line_ids.prod_lot_id = sn1
inventory.line_ids.product_qty = 1
inventory.action_validate()
self.assertEqual(self.env['stock.quant']._get_available_quantity(self.product2, self.stock_location), 1.0)

# Make a move to customer for sn1
move_delivery = self.env['stock.move'].create({
'name': 'delivery1',
'location_id': self.stock_location.id,
'location_dest_id': self.customer_location.id,
'product_id': self.product2.id,
'product_uom': self.uom_unit.id,
'product_uom_qty': 1.0,
})

move_delivery._action_confirm()
move_delivery._action_assign()
self.assertEqual(move_delivery.state, 'assigned')
move_delivery.move_line_ids.qty_done = 1
move_delivery._action_done()
self.assertEqual(move_delivery.state, 'done')
self.assertEqual(self.env['stock.quant']._gather(self.product2, self.customer_location).quantity, 1.0)
self.assertEqual(self.env['stock.quant']._get_available_quantity(self.product2, self.stock_location), 0.0)

# Create a new inventory and try to put sn1 in stock
inventory = self.env['stock.inventory'].create({
'name': 'serial product',
'filter': 'product',
'location_id': self.stock_location.id,
'product_id': self.product2.id,
'exhausted': True, # should be set by an onchange
})
inventory.action_start()
inventory.line_ids.prod_lot_id = sn1
inventory.line_ids.product_qty = 1
res_dict_for_duplicate = inventory.action_validate()
wizard_duplicate = self.env[(res_dict_for_duplicate.get('res_model'))].browse(res_dict_for_duplicate.get('res_id'))
wizard_duplicate.action_confirm()
self.assertEqual(len(inventory.line_ids), 2, "There is a missing line")
correction_line = inventory.line_ids.filtered(lambda line: line.correction_line == 'warning')
self.assertEqual(correction_line.location_id, self.customer_location, "There is a missing line")
self.assertEqual(correction_line.theoretical_qty, 1, "The theoretical quantity is wrong")
self.assertEqual(correction_line.product_qty, 0, "The real quantity is wrong")
inventory.action_validate()
self.assertEqual(inventory.state, 'done')
self.assertEqual(self.env['stock.quant']._gather(self.product2, self.customer_location).quantity, 0)
self.assertEqual(self.env['stock.quant']._gather(self.product2, self.stock_location).quantity, 1.0)
Oops, something went wrong.

0 comments on commit 1ec67c5

Please sign in to comment.