Merge pull request #54785 from aerele/fix/support-#67579

fix(stock): add validation for work order serial nos and batch nos

erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py

@@ -7,10 +7,13 @@ from frappe.query_builder.functions import Sum
 from frappe.utils import ceil, cint, flt, get_link_to_form
 
 from erpnext.manufacturing.doctype.bom.bom import add_additional_cost
+from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
 from erpnext.stock.serial_batch_bundle import (
 	SerialBatchCreation,
 	get_batch_nos,
+	get_batches_from_bundle,
 	get_empty_batches_based_work_order,
+	get_serial_nos_from_bundle,
 )
 
 from .base import BaseStockEntry
@@ -171,25 +174,30 @@ class BaseManufactureStockEntry(BaseStockEntry):
 		else:
 			self.doc.append("items", item_details)
 
-	def set_serial_nos_for_finished_good(self, item_details):
+	def set_serial_nos_for_finished_good(self, item_details, existing_row=None):
 		serial_nos = self.get_available_serial_nos_for_fg(item_details.item_code)
-		if serial_nos:
+		if not serial_nos:
-			row = frappe._dict({"serial_nos": serial_nos[0 : cint(item_details.qty)]})
+			return
 
-			_id = create_serial_and_batch_bundle(
+		row = frappe._dict({"serial_nos": serial_nos[0 : cint(item_details.qty)]})
-				self.doc,
-				row,
-				frappe._dict(
-					{
-						"item_code": item_details.item_code,
-						"warehouse": item_details.t_warehouse,
-					}
-				),
-			)
 
+		_id = create_serial_and_batch_bundle(
+			self.doc,
+			row,
+			frappe._dict(
+				{
+					"item_code": item_details.item_code,
+					"warehouse": item_details.t_warehouse,
+				}
+			),
+		)
+
+		if existing_row:
+			existing_row.serial_and_batch_bundle = _id
+			existing_row.use_serial_batch_fields = 0
+		else:
 			item_details.serial_and_batch_bundle = _id
 			item_details.use_serial_batch_fields = 0
-
 			self.doc.append("items", item_details)
 
 	def get_available_serial_nos_for_fg(self, item_code) -> list[str]:
@@ -205,22 +213,23 @@ class BaseManufactureStockEntry(BaseStockEntry):
 			order_by="creation asc",
 		)
 
-	def set_batchwise_finished_goods(self, item_details):
+	def set_batchwise_finished_goods(self, item_details, existing_row=None):
-		batches = get_empty_batches_based_work_order(self.doc.work_order, self.doc.pro_doc.production_item)
+		batches = get_empty_batches_based_work_order(self.doc.work_order, self.wo_doc.production_item)
 
 		if not batches:
-			self.doc.append("items", item_details)
+			if not existing_row:
+				self.doc.append("items", item_details)
 		else:
-			self.add_batchwise_finished_good(batches, item_details)
+			self.add_batchwise_finished_good(batches, item_details, existing_row=existing_row)
 
-	def add_batchwise_finished_good(self, batches, item_details):
+	def add_batchwise_finished_good(self, batches, item_details, existing_row=None):
 		qty = flt(self.doc.fg_completed_qty)
 		row = frappe._dict({"batches_to_be_consume": defaultdict(float)})
 		self.update_batches_to_be_consume(batches, row, qty)
 		if row.batches_to_be_consume:
-			self._link_fg_bundle_and_append(item_details, row)
+			self._link_fg_bundle_and_append(item_details, row, existing_row=existing_row)
 
-	def _link_fg_bundle_and_append(self, item_details, row):
+	def _link_fg_bundle_and_append(self, item_details, row, existing_row=None):
 		_id = create_serial_and_batch_bundle(
 			self.doc,
 			row,
@@ -228,8 +237,13 @@ class BaseManufactureStockEntry(BaseStockEntry):
 				{"item_code": self.wo_doc.production_item, "warehouse": item_details.get("t_warehouse")}
 			),
 		)
-		item_details["serial_and_batch_bundle"] = _id
+		if existing_row:
-		self.doc.append("items", item_details)
+			existing_row.serial_and_batch_bundle = _id
+			existing_row.use_serial_batch_fields = 0
+		else:
+			item_details["serial_and_batch_bundle"] = _id
+			item_details["use_serial_batch_fields"] = 0
+			self.doc.append("items", item_details)
 
 	def update_batches_to_be_consume(self, batches, row, qty):
 		qty_to_be_consumed = qty
@@ -259,6 +273,81 @@ class ManufactureStockEntry(BaseManufactureStockEntry):
 		self.validate_warehouse()
 		self.validate_raw_materials_exists()
 		self.validate_component_and_quantities()
+		self.validate_finished_good_serial_batch_for_work_order()
+
+	def validate_finished_good_serial_batch_for_work_order(self):
+		if not (
+			self.doc.work_order
+			and self.wo_doc
+			and self.wo_doc.track_semi_finished_goods != 1
+			and cint(
+				frappe.db.get_single_value(
+					"Manufacturing Settings", "make_serial_no_batch_from_work_order", cache=True
+				)
+			)
+			and (self.wo_doc.has_serial_no or self.wo_doc.has_batch_no)
+		):
+			return
+
+		for row in self.doc.items:
+			if not row.is_finished_item:
+				continue
+
+			if self.check_invalid_serial_batch_nos_for_finished_good_item(row):
+				self.reset_serial_batch_on_fg_row(row)
+				frappe.msgprint(
+					_(
+						"Row {0}: Serial/Batch has been reset to values linked with Work Order {1}"
+						" because the previously selected serial/batch does not belong to this Work Order."
+					).format(row.idx, frappe.bold(self.doc.work_order))
+				)
+
+	def check_invalid_serial_batch_nos_for_finished_good_item(self, row) -> bool:
+		if self.wo_doc.has_serial_no:
+			serial_nos = get_serial_nos(row.serial_no) if row.serial_no else []
+			if not serial_nos and row.serial_and_batch_bundle:
+				serial_nos = get_serial_nos_from_bundle(row.serial_and_batch_bundle)
+			if serial_nos:
+				valid_serial_nos = frappe.get_all(
+					"Serial No",
+					filters={"name": ("in", serial_nos), "work_order": self.doc.work_order},
+					pluck="name",
+				)
+				return bool(set(serial_nos) - set(valid_serial_nos))
+			else:
+				return True
+
+		if self.wo_doc.has_batch_no:
+			batch_nos = [row.batch_no] if row.batch_no else []
+			if not batch_nos and row.serial_and_batch_bundle:
+				batch_nos = list(get_batches_from_bundle(row.serial_and_batch_bundle).keys())
+			if batch_nos:
+				valid_batch_nos = frappe.get_all(
+					"Batch",
+					filters={"name": ("in", batch_nos), "reference_name": self.doc.work_order},
+					pluck="name",
+				)
+				return bool(set(batch_nos) - set(valid_batch_nos))
+			else:
+				return True
+
+	def reset_serial_batch_on_fg_row(self, row):
+		item_details = frappe._dict(
+			{
+				"item_code": row.item_code,
+				"t_warehouse": row.t_warehouse,
+				"qty": row.qty,
+			}
+		)
+
+		row.serial_no = None
+		row.batch_no = None
+		row.serial_and_batch_bundle = None
+
+		if self.wo_doc.has_serial_no:
+			self.set_serial_nos_for_finished_good(item_details, existing_row=row)
+		elif self.wo_doc.has_batch_no:
+			self.set_batchwise_finished_goods(item_details, existing_row=row)
 
 	def set_job_card_data(self):
 		if self.doc.job_card and not self.doc.work_order:

erpnext/stock/doctype/stock_entry/test_stock_entry.py

@@ -2886,6 +2886,88 @@ class TestStockEntryCoverage(ERPNextTestSuite):
 		if key in materials:
 			self.assertEqual(materials[key].qty, 0)
 
+	@ERPNextTestSuite.change_settings("Manufacturing Settings", {"make_serial_no_batch_from_work_order": 1})
+	@ERPNextTestSuite.change_settings("Global Defaults", {"default_company": "_Test Company"})
+	def test_validate_fg_resets_invalid_serial_no_on_manufacture(self):
+		from erpnext.manufacturing.doctype.bom.test_bom import create_nested_bom
+		from erpnext.manufacturing.doctype.work_order.test_work_order import make_wo_order_test_record
+		from erpnext.manufacturing.doctype.work_order.work_order import (
+			make_stock_entry as _make_stock_entry,
+		)
+
+		fg_item = "_FG Serial No Item"
+		rm_item = "RM for serial item"
+		create_nested_bom({fg_item: {rm_item: {}}}, prefix="")
+
+		item = frappe.get_doc("Item", fg_item)
+		item.has_serial_no = 1
+		item.serial_no_series = "FSNI-.####"
+		item.save()
+
+		make_stock_entry(item_code=rm_item, target="_Test Warehouse - _TC", qty=20, basic_rate=100)
+
+		wo1 = make_wo_order_test_record(item=fg_item, qty=2, skip_transfer=True)
+		wo2 = make_wo_order_test_record(item=fg_item, qty=2, skip_transfer=True)
+		wo1_serial_nos = frappe.get_all("Serial No", filters={"work_order": wo1.name}, pluck="name")
+		wo2_serial_nos = frappe.get_all("Serial No", filters={"work_order": wo2.name}, pluck="name")
+
+		se = frappe.get_doc(_make_stock_entry(wo1.name, "Manufacture", 2))
+		for row in se.items:
+			if row.is_finished_item:
+				row.serial_no = wo2_serial_nos[0]
+				row.serial_and_batch_bundle = None
+
+		se.save()
+
+		for row in se.items:
+			if row.is_finished_item:
+				self.assertIsNone(row.serial_no)
+				self.assertTrue(row.serial_and_batch_bundle)
+				for sn in get_serial_nos_from_bundle(row.serial_and_batch_bundle):
+					self.assertIn(sn, wo1_serial_nos)
+
+	@ERPNextTestSuite.change_settings("Manufacturing Settings", {"make_serial_no_batch_from_work_order": 1})
+	@ERPNextTestSuite.change_settings("Global Defaults", {"default_company": "_Test Company"})
+	def test_validate_fg_resets_invalid_batch_no_on_manufacture(self):
+		from erpnext.manufacturing.doctype.bom.test_bom import create_nested_bom
+		from erpnext.manufacturing.doctype.work_order.test_work_order import make_wo_order_test_record
+		from erpnext.manufacturing.doctype.work_order.work_order import (
+			make_stock_entry as _make_stock_entry,
+		)
+		from erpnext.stock.serial_batch_bundle import get_batches_from_bundle
+
+		fg_item = "_FG Batch No Item"
+		rm_item = "RM for Batch Item"
+		create_nested_bom({fg_item: {rm_item: {}}}, prefix="")
+
+		item = frappe.get_doc("Item", fg_item)
+		item.has_batch_no = 1
+		item.create_new_batch = 1
+		item.batch_number_series = "FBNI-.####"
+		item.save()
+
+		make_stock_entry(item_code=rm_item, target="_Test Warehouse - _TC", qty=20, basic_rate=100)
+
+		wo1 = make_wo_order_test_record(item=fg_item, qty=2, skip_transfer=True)
+		wo2 = make_wo_order_test_record(item=fg_item, qty=2, skip_transfer=True)
+		wo1_batches = frappe.get_all("Batch", filters={"reference_name": wo1.name}, pluck="name")
+		wo2_batches = frappe.get_all("Batch", filters={"reference_name": wo2.name}, pluck="name")
+
+		se = frappe.get_doc(_make_stock_entry(wo1.name, "Manufacture", 2))
+		for row in se.items:
+			if row.is_finished_item:
+				row.batch_no = wo2_batches[0]
+				row.serial_and_batch_bundle = None
+
+		se.save()
+
+		for row in se.items:
+			if row.is_finished_item:
+				self.assertIsNone(row.batch_no)
+				self.assertTrue(row.serial_and_batch_bundle)
+				for bn in list(get_batches_from_bundle(row.serial_and_batch_bundle).keys()):
+					self.assertIn(bn, wo1_batches)
+
 
 def make_serialized_item(self, **args):
 	args = frappe._dict(args)