diff --git a/erpnext/stock/doctype/inventory_dimension/test_inventory_dimension.py b/erpnext/stock/doctype/inventory_dimension/test_inventory_dimension.py index 655d781126c..dd287fa1b52 100644 --- a/erpnext/stock/doctype/inventory_dimension/test_inventory_dimension.py +++ b/erpnext/stock/doctype/inventory_dimension/test_inventory_dimension.py @@ -12,10 +12,13 @@ from erpnext.stock.doctype.inventory_dimension.inventory_dimension import ( DoNotChangeError, delete_dimension, ) -from erpnext.stock.doctype.item.test_item import create_item +from erpnext.stock.doctype.item.test_item import create_item, make_item from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry -from erpnext.stock.doctype.stock_ledger_entry.stock_ledger_entry import InventoryDimensionNegativeStockError +from erpnext.stock.doctype.stock_ledger_entry.stock_ledger_entry import ( + InventoryDimensionNegativeStockError, + SerialNoInventoryDimensionError, +) from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse from erpnext.tests.utils import ERPNextTestSuite @@ -492,6 +495,193 @@ class TestInventoryDimension(ERPNextTestSuite): self.assertEqual(site_name, "Site 1") + def test_serial_no_cannot_be_issued_from_incorrect_inventory_dimension(self): + item = make_item( + "Test Serialized Inventory Dimension Item", + {"has_serial_no": 1, "is_stock_item": 1}, + ) + serial_no = "Test Serialized Inventory Dimension Serial No" + warehouse = create_warehouse("Serialized Inventory Dimension Warehouse") + + create_inventory_dimension( + apply_to_all_doctypes=1, + dimension_name="Serial Rack", + reference_document="Rack", + validate_negative_stock=0, + ) + + receipt = make_stock_entry( + item_code=item.name, + to_warehouse=warehouse, + qty=1, + serial_no=serial_no, + use_serial_batch_fields=1, + do_not_submit=True, + ) + receipt.items[0].to_serial_rack = "Rack 1" + receipt.save() + receipt.submit() + + transfer = make_stock_entry( + item_code=item.name, + from_warehouse=warehouse, + to_warehouse=warehouse, + qty=1, + serial_no=serial_no, + use_serial_batch_fields=1, + do_not_submit=True, + ) + transfer.items[0].serial_rack = "Rack 1" + transfer.items[0].to_serial_rack = "Rack 2" + transfer.save() + transfer.submit() + + issue = make_stock_entry( + item_code=item.name, + from_warehouse=warehouse, + qty=1, + serial_no=serial_no, + use_serial_batch_fields=1, + do_not_submit=True, + ) + issue.items[0].serial_rack = "Rack 1" + issue.save() + + self.assertRaises(SerialNoInventoryDimensionError, issue.submit) + self.assertFalse( + frappe.db.exists( + "Stock Ledger Entry", + {"voucher_no": issue.name, "is_cancelled": 0}, + ) + ) + + def test_serial_no_cannot_move_from_empty_inventory_dimension(self): + item = make_item( + "Test Serialized Empty Inventory Dimension Item", + {"has_serial_no": 1, "is_stock_item": 1}, + ) + serial_no = "Test Serialized Empty Inventory Dimension Serial No" + warehouse = create_warehouse("Serialized Empty Inventory Dimension Warehouse") + + create_inventory_dimension( + apply_to_all_doctypes=1, + dimension_name="Empty Serial Rack", + reference_document="Rack", + validate_negative_stock=0, + ) + + make_stock_entry( + item_code=item.name, + to_warehouse=warehouse, + qty=1, + serial_no=serial_no, + use_serial_batch_fields=1, + ) + + issue = make_stock_entry( + item_code=item.name, + from_warehouse=warehouse, + qty=1, + serial_no=serial_no, + use_serial_batch_fields=1, + do_not_submit=True, + ) + issue.items[0].empty_serial_rack = "Rack 1" + issue.save() + + self.assertRaises(SerialNoInventoryDimensionError, issue.submit) + + def test_serial_no_cannot_be_issued_without_inventory_dimension(self): + item = make_item( + "Test Serialized Required Inventory Dimension Item", + {"has_serial_no": 1, "is_stock_item": 1}, + ) + serial_no = "Test Serialized Required Inventory Dimension Serial No" + warehouse = create_warehouse("Serialized Required Inventory Dimension Warehouse") + + create_inventory_dimension( + apply_to_all_doctypes=1, + dimension_name="Required Serial Rack", + reference_document="Rack", + validate_negative_stock=0, + ) + + receipt = make_stock_entry( + item_code=item.name, + to_warehouse=warehouse, + qty=1, + serial_no=serial_no, + use_serial_batch_fields=1, + do_not_submit=True, + ) + receipt.items[0].to_required_serial_rack = "Rack 1" + receipt.save() + receipt.submit() + + issue = make_stock_entry( + item_code=item.name, + from_warehouse=warehouse, + qty=1, + serial_no=serial_no, + use_serial_batch_fields=1, + do_not_submit=True, + ) + issue.save() + + self.assertRaises(SerialNoInventoryDimensionError, issue.submit) + self.assertFalse( + frappe.db.exists( + "Stock Ledger Entry", + {"voucher_no": issue.name, "is_cancelled": 0}, + ) + ) + + def test_serial_no_inventory_dimension_with_legacy_inward_sle(self): + item = make_item( + "Test Serialized Legacy Inventory Dimension Item", + {"has_serial_no": 1, "is_stock_item": 1}, + ) + serial_no = "Test Serialized Legacy Inventory Dimension Serial No" + warehouse = create_warehouse("Serialized Legacy Inventory Dimension Warehouse") + + create_inventory_dimension( + apply_to_all_doctypes=1, + dimension_name="Legacy Serial Rack", + reference_document="Rack", + validate_negative_stock=0, + ) + + receipt = make_stock_entry( + item_code=item.name, + to_warehouse=warehouse, + qty=1, + serial_no=serial_no, + use_serial_batch_fields=1, + do_not_submit=True, + ) + receipt.items[0].to_legacy_serial_rack = "Rack 2" + receipt.save() + receipt.submit() + + frappe.db.set_value( + "Stock Ledger Entry", + {"voucher_no": receipt.name, "actual_qty": (">", 0), "is_cancelled": 0}, + {"serial_and_batch_bundle": None, "serial_no": f"Other Legacy Serial, {serial_no}"}, + ) + + issue = make_stock_entry( + item_code=item.name, + from_warehouse=warehouse, + qty=1, + serial_no=serial_no, + use_serial_batch_fields=1, + do_not_submit=True, + ) + issue.items[0].legacy_serial_rack = "Rack 1" + issue.save() + + self.assertRaises(SerialNoInventoryDimensionError, issue.submit) + @ERPNextTestSuite.change_settings("Stock Settings", {"allow_negative_stock": 0}) def test_validate_negative_stock_with_multiple_dimension(self): item_code = "Test Negative Multi Inventory Dimension Item" diff --git a/erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py b/erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py index 45cdd579b96..da8eab3042d 100644 --- a/erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py +++ b/erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py @@ -2,19 +2,21 @@ # License: GNU General Public License v3. See license.txt +import re from datetime import date import frappe from frappe import _ from frappe.core.doctype.role.role import get_users from frappe.model.document import Document -from frappe.query_builder.functions import Max, Sum +from frappe.query_builder.functions import Concat_ws, Max, Sum from frappe.utils import add_days, cint, flt, formatdate, get_datetime, getdate from erpnext.accounts.utils import get_fiscal_year from erpnext.controllers.item_variant import ItemTemplateCannotHaveStock from erpnext.stock.doctype.inventory_dimension.inventory_dimension import get_inventory_dimensions -from erpnext.stock.serial_batch_bundle import SerialBatchBundle +from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos as get_parsed_serial_nos +from erpnext.stock.serial_batch_bundle import SerialBatchBundle, get_serial_nos class StockFreezeError(frappe.ValidationError): @@ -29,6 +31,10 @@ class InventoryDimensionNegativeStockError(frappe.ValidationError): pass +class SerialNoInventoryDimensionError(frappe.ValidationError): + pass + + exclude_from_linked_with = True @@ -97,6 +103,7 @@ class StockLedgerEntry(Document): self.block_transactions_against_group_warehouse() self.validate_with_last_transaction_posting_time() self.validate_inventory_dimension_negative_stock() + self.validate_serial_no_inventory_dimension() def set_posting_datetime(self): from erpnext.stock.utils import get_combine_datetime @@ -171,6 +178,87 @@ class StockLedgerEntry(Document): return inv_dimension_dict + def validate_serial_no_inventory_dimension(self): + if self.is_cancelled or self.actual_qty >= 0 or not self.has_serial_no: + return + + dimensions = get_inventory_dimensions() + if not dimensions: + return + + serial_nos = get_serial_nos(self.serial_and_batch_bundle) + if not serial_nos and self.serial_no: + serial_nos = get_parsed_serial_nos(self.serial_no) + + if not serial_nos: + return + + for serial_no, values in self.get_last_inward_dimensions(serial_nos, dimensions).items(): + mismatches = [] + for dimension in dimensions: + fieldname = dimension.fieldname + expected_value = values.get(fieldname) + if expected_value != self.get(fieldname): + mismatches.append( + _('{0}: expected "{1}", got "{2}"').format( + dimension.dimension_name, + expected_value or _("Not Set"), + self.get(fieldname), + ) + ) + + if mismatches: + frappe.throw( + _("Serial No {0} is not available in the selected inventory dimensions: {1}").format( + frappe.bold(serial_no), frappe.bold(", ".join(mismatches)) + ), + title=_("Incorrect Inventory Dimension"), + exc=SerialNoInventoryDimensionError, + ) + + def get_last_inward_dimensions(self, serial_nos, dimensions): + sle = frappe.qb.DocType("Stock Ledger Entry") + serial_entry = frappe.qb.DocType("Serial and Batch Entry") + dimension_fields = [sle[dimension.fieldname].as_(dimension.fieldname) for dimension in dimensions] + escaped_serial_nos = [re.escape(serial_no) for serial_no in serial_nos] + legacy_serial_pattern = r"[\n,][[:space:]]*(" + "|".join(escaped_serial_nos) + r")[[:space:]]*[\n,]" + legacy_serial_condition = ( + sle.serial_and_batch_bundle.isnull() | (sle.serial_and_batch_bundle == "") + ) & Concat_ws("", "\n", sle.serial_no, "\n").regexp(legacy_serial_pattern) + + rows = ( + frappe.qb.from_(sle) + .left_join(serial_entry) + .on(serial_entry.parent == sle.serial_and_batch_bundle) + .select( + serial_entry.serial_no.as_("bundle_serial_no"), + sle.serial_no.as_("legacy_serial_nos"), + *dimension_fields, + ) + .where( + (serial_entry.serial_no.isin(serial_nos) | legacy_serial_condition) + & (sle.item_code == self.item_code) + & (sle.actual_qty > 0) + & (sle.is_cancelled == 0) + & (sle.posting_datetime <= self.posting_datetime) + ) + .orderby(sle.posting_datetime, order=frappe.qb.desc) + .orderby(sle.creation, order=frappe.qb.desc) + ).run(as_dict=True) + + serial_nos = set(serial_nos) + last_inward_dimensions = {} + for row in rows: + row_serial_nos = ( + [row.bundle_serial_no] + if row.bundle_serial_no + else get_parsed_serial_nos(row.legacy_serial_nos) + ) + for serial_no in serial_nos.intersection(row_serial_nos): + last_inward_dimensions.setdefault(serial_no, row) + + return last_inward_dimensions + def on_submit(self): self.check_stock_frozen_date()