fix(stock): validate serial inventory dimensions (#58394)

* fix(stock): validate serial inventory dimensions

* test(stock): cover serial inventory dimensions

(cherry picked from commit c940bd1e66)

# Conflicts:
#	erpnext/stock/doctype/inventory_dimension/test_inventory_dimension.py
#	erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py

erpnext/stock/doctype/inventory_dimension/test_inventory_dimension.py

@@ -13,10 +13,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
 
 
@@ -504,6 +507,193 @@ class TestInventoryDimension(FrappeTestCase):
 
 		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)
+
 	def test_validate_negative_stock_with_multiple_dimension(self):
 		frappe.db.set_single_value("Stock Settings", "allow_negative_stock", 0)
 		item_code = "Test Negative Multi Inventory Dimension Item"
@@ -709,13 +899,16 @@ def create_inventory_dimension(**args):
 	args = frappe._dict(args)
 
 	if frappe.db.exists("Inventory Dimension", args.dimension_name):
-		return frappe.get_doc("Inventory Dimension", args.dimension_name)
+		doc = frappe.get_doc("Inventory Dimension", args.dimension_name)
+	else:
+		doc = frappe.new_doc("Inventory Dimension")
+		doc.update(args)
 
-	doc = frappe.new_doc("Inventory Dimension")
+		if not args.do_not_save:
-	doc.update(args)
+			doc.insert(ignore_permissions=True)
 
-	if not args.do_not_save:
+	frappe.local.inventory_dimensions = {}
-		doc.insert(ignore_permissions=True)
+	frappe.local.document_wise_inventory_dimensions = {}
 
 	return doc
 

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 _, bold
 from frappe.core.doctype.role.role import get_users
 from frappe.model.document import Document
-from frappe.query_builder.functions import Sum
+from frappe.query_builder.functions import Concat_ws, 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
 from erpnext.stock.stock_ledger import get_previous_sle
 
 
@@ -30,6 +32,10 @@ class InventoryDimensionNegativeStockError(frappe.ValidationError):
 	pass
 
 
+class SerialNoInventoryDimensionError(frappe.ValidationError):
+	pass
+
+
 exclude_from_linked_with = True
 
 
@@ -98,6 +104,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
@@ -172,6 +179,87 @@ class StockLedgerEntry(Document):
 
 		return inv_dimension_dict
 
+	def validate_serial_no_inventory_dimension(self):
+		if self.is_cancelled or self.actual_qty >= 0:
+			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()