diff --git a/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py b/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py index 648b91ab9d4..6ce37f7c4ef 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py +++ b/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: - row = frappe._dict({"serial_nos": serial_nos[0 : cint(item_details.qty)]}) + if not serial_nos: + return - _id = create_serial_and_batch_bundle( - self.doc, - row, - frappe._dict( - { - "item_code": item_details.item_code, - "warehouse": item_details.t_warehouse, - } - ), - ) + row = frappe._dict({"serial_nos": serial_nos[0 : cint(item_details.qty)]}) + _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): - batches = get_empty_batches_based_work_order(self.doc.work_order, self.doc.pro_doc.production_item) + def set_batchwise_finished_goods(self, item_details, existing_row=None): + 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 - self.doc.append("items", item_details) + 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 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: diff --git a/erpnext/stock/doctype/stock_entry/test_stock_entry.py b/erpnext/stock/doctype/stock_entry/test_stock_entry.py index 85870874644..e42641cdfa8 100644 --- a/erpnext/stock/doctype/stock_entry/test_stock_entry.py +++ b/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)