test: test cases to cover batch, serialized raw materials
diff --git a/erpnext/manufacturing/doctype/work_order/test_work_order.py b/erpnext/manufacturing/doctype/work_order/test_work_order.py
index 2aba482..bd19dec 100644
--- a/erpnext/manufacturing/doctype/work_order/test_work_order.py
+++ b/erpnext/manufacturing/doctype/work_order/test_work_order.py
@@ -1,6 +1,8 @@
# Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
+import copy
+
import frappe
from frappe.tests.utils import FrappeTestCase, change_settings, timeout
from frappe.utils import add_days, add_months, cint, flt, now, today
@@ -19,6 +21,7 @@
)
from erpnext.selling.doctype.sales_order.test_sales_order import make_sales_order
from erpnext.stock.doctype.item.test_item import create_item, make_item
+from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
from erpnext.stock.doctype.stock_entry import test_stock_entry
from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
from erpnext.stock.utils import get_bin
@@ -28,6 +31,7 @@
def setUp(self):
self.warehouse = "_Test Warehouse 2 - _TC"
self.item = "_Test Item"
+ prepare_data_for_backflush_based_on_materials_transferred()
def tearDown(self):
frappe.db.rollback()
@@ -518,6 +522,8 @@
work_order.cancel()
def test_work_order_with_non_transfer_item(self):
+ frappe.db.set_value("Manufacturing Settings", None, "backflush_raw_materials_based_on", "BOM")
+
items = {"Finished Good Transfer Item": 1, "_Test FG Item": 1, "_Test FG Item 1": 0}
for item, allow_transfer in items.items():
make_item(item, {"include_item_in_manufacturing": allow_transfer})
@@ -1062,7 +1068,7 @@
sm = frappe.get_doc(make_stock_entry(wo_order.name, "Material Transfer for Manufacture", 100))
for row in sm.get("items"):
if row.get("item_code") == "_Test Item":
- row.qty = 110
+ row.qty = 120
sm.submit()
cancel_stock_entry.append(sm.name)
@@ -1070,21 +1076,21 @@
s = frappe.get_doc(make_stock_entry(wo_order.name, "Manufacture", 90))
for row in s.get("items"):
if row.get("item_code") == "_Test Item":
- self.assertEqual(row.get("qty"), 100)
+ self.assertEqual(row.get("qty"), 108)
s.submit()
cancel_stock_entry.append(s.name)
s1 = frappe.get_doc(make_stock_entry(wo_order.name, "Manufacture", 5))
for row in s1.get("items"):
if row.get("item_code") == "_Test Item":
- self.assertEqual(row.get("qty"), 5)
+ self.assertEqual(row.get("qty"), 6)
s1.submit()
cancel_stock_entry.append(s1.name)
s2 = frappe.get_doc(make_stock_entry(wo_order.name, "Manufacture", 5))
for row in s2.get("items"):
if row.get("item_code") == "_Test Item":
- self.assertEqual(row.get("qty"), 5)
+ self.assertEqual(row.get("qty"), 6)
cancel_stock_entry.reverse()
for ste in cancel_stock_entry:
@@ -1194,6 +1200,269 @@
self.assertEqual(work_order.required_items[0].transferred_qty, 1)
self.assertEqual(work_order.required_items[1].transferred_qty, 2)
+ def test_backflushed_batch_raw_materials_based_on_transferred(self):
+ frappe.db.set_value(
+ "Manufacturing Settings",
+ None,
+ "backflush_raw_materials_based_on",
+ "Material Transferred for Manufacture",
+ )
+
+ batch_item = "Test Batch MCC Keyboard"
+ fg_item = "Test FG Item with Batch Raw Materials"
+
+ ste_doc = test_stock_entry.make_stock_entry(
+ item_code=batch_item, target="Stores - _TC", qty=2, basic_rate=100, do_not_save=True
+ )
+
+ ste_doc.append(
+ "items",
+ {
+ "item_code": batch_item,
+ "item_name": batch_item,
+ "description": batch_item,
+ "basic_rate": 100,
+ "t_warehouse": "Stores - _TC",
+ "qty": 2,
+ "uom": "Nos",
+ "stock_uom": "Nos",
+ "conversion_factor": 1,
+ },
+ )
+
+ # Inward raw materials in Stores warehouse
+ ste_doc.insert()
+ ste_doc.submit()
+
+ batch_list = [row.batch_no for row in ste_doc.items]
+
+ wo_doc = make_wo_order_test_record(production_item=fg_item, qty=4)
+ transferred_ste_doc = frappe.get_doc(
+ make_stock_entry(wo_doc.name, "Material Transfer for Manufacture", 4)
+ )
+
+ transferred_ste_doc.items[0].qty = 2
+ transferred_ste_doc.items[0].batch_no = batch_list[0]
+
+ new_row = copy.deepcopy(transferred_ste_doc.items[0])
+ new_row.name = ""
+ new_row.batch_no = batch_list[1]
+
+ # Transferred two batches from Stores to WIP Warehouse
+ transferred_ste_doc.append("items", new_row)
+ transferred_ste_doc.submit()
+
+ # First Manufacture stock entry
+ manufacture_ste_doc1 = frappe.get_doc(make_stock_entry(wo_doc.name, "Manufacture", 1))
+
+ # Batch no should be same as transferred Batch no
+ self.assertEqual(manufacture_ste_doc1.items[0].batch_no, batch_list[0])
+ self.assertEqual(manufacture_ste_doc1.items[0].qty, 1)
+
+ manufacture_ste_doc1.submit()
+
+ # Second Manufacture stock entry
+ manufacture_ste_doc2 = frappe.get_doc(make_stock_entry(wo_doc.name, "Manufacture", 2))
+
+ # Batch no should be same as transferred Batch no
+ self.assertEqual(manufacture_ste_doc2.items[0].batch_no, batch_list[0])
+ self.assertEqual(manufacture_ste_doc2.items[0].qty, 1)
+ self.assertEqual(manufacture_ste_doc2.items[1].batch_no, batch_list[1])
+ self.assertEqual(manufacture_ste_doc2.items[1].qty, 1)
+
+ def test_backflushed_serial_no_raw_materials_based_on_transferred(self):
+ frappe.db.set_value(
+ "Manufacturing Settings",
+ None,
+ "backflush_raw_materials_based_on",
+ "Material Transferred for Manufacture",
+ )
+
+ sn_item = "Test Serial No BTT Headphone"
+ fg_item = "Test FG Item with Serial No Raw Materials"
+
+ ste_doc = test_stock_entry.make_stock_entry(
+ item_code=sn_item, target="Stores - _TC", qty=4, basic_rate=100, do_not_save=True
+ )
+
+ # Inward raw materials in Stores warehouse
+ ste_doc.submit()
+
+ serial_nos_list = sorted(get_serial_nos(ste_doc.items[0].serial_no))
+
+ wo_doc = make_wo_order_test_record(production_item=fg_item, qty=4)
+ transferred_ste_doc = frappe.get_doc(
+ make_stock_entry(wo_doc.name, "Material Transfer for Manufacture", 4)
+ )
+
+ transferred_ste_doc.items[0].serial_no = "\n".join(serial_nos_list)
+ transferred_ste_doc.submit()
+
+ # First Manufacture stock entry
+ manufacture_ste_doc1 = frappe.get_doc(make_stock_entry(wo_doc.name, "Manufacture", 1))
+
+ # Serial nos should be same as transferred Serial nos
+ self.assertEqual(get_serial_nos(manufacture_ste_doc1.items[0].serial_no), serial_nos_list[0:1])
+ self.assertEqual(manufacture_ste_doc1.items[0].qty, 1)
+
+ manufacture_ste_doc1.submit()
+
+ # Second Manufacture stock entry
+ manufacture_ste_doc2 = frappe.get_doc(make_stock_entry(wo_doc.name, "Manufacture", 2))
+
+ # Serial nos should be same as transferred Serial nos
+ self.assertEqual(get_serial_nos(manufacture_ste_doc2.items[0].serial_no), serial_nos_list[1:3])
+ self.assertEqual(manufacture_ste_doc2.items[0].qty, 2)
+
+ def test_backflushed_serial_no_batch_raw_materials_based_on_transferred(self):
+ frappe.db.set_value(
+ "Manufacturing Settings",
+ None,
+ "backflush_raw_materials_based_on",
+ "Material Transferred for Manufacture",
+ )
+
+ sn_batch_item = "Test Batch Serial No WebCam"
+ fg_item = "Test FG Item with Serial & Batch No Raw Materials"
+
+ ste_doc = test_stock_entry.make_stock_entry(
+ item_code=sn_batch_item, target="Stores - _TC", qty=2, basic_rate=100, do_not_save=True
+ )
+
+ ste_doc.append(
+ "items",
+ {
+ "item_code": sn_batch_item,
+ "item_name": sn_batch_item,
+ "description": sn_batch_item,
+ "basic_rate": 100,
+ "t_warehouse": "Stores - _TC",
+ "qty": 2,
+ "uom": "Nos",
+ "stock_uom": "Nos",
+ "conversion_factor": 1,
+ },
+ )
+
+ # Inward raw materials in Stores warehouse
+ ste_doc.insert()
+ ste_doc.submit()
+
+ batch_dict = {row.batch_no: get_serial_nos(row.serial_no) for row in ste_doc.items}
+ batches = list(batch_dict.keys())
+
+ wo_doc = make_wo_order_test_record(production_item=fg_item, qty=4)
+ transferred_ste_doc = frappe.get_doc(
+ make_stock_entry(wo_doc.name, "Material Transfer for Manufacture", 4)
+ )
+
+ transferred_ste_doc.items[0].qty = 2
+ transferred_ste_doc.items[0].batch_no = batches[0]
+ transferred_ste_doc.items[0].serial_no = "\n".join(batch_dict.get(batches[0]))
+
+ new_row = copy.deepcopy(transferred_ste_doc.items[0])
+ new_row.name = ""
+ new_row.batch_no = batches[1]
+ new_row.serial_no = "\n".join(batch_dict.get(batches[1]))
+
+ # Transferred two batches from Stores to WIP Warehouse
+ transferred_ste_doc.append("items", new_row)
+ transferred_ste_doc.submit()
+
+ # First Manufacture stock entry
+ manufacture_ste_doc1 = frappe.get_doc(make_stock_entry(wo_doc.name, "Manufacture", 1))
+
+ # Batch no & Serial Nos should be same as transferred Batch no & Serial Nos
+ batch_no = manufacture_ste_doc1.items[0].batch_no
+ self.assertEqual(
+ get_serial_nos(manufacture_ste_doc1.items[0].serial_no)[0], batch_dict.get(batch_no)[0]
+ )
+ self.assertEqual(manufacture_ste_doc1.items[0].qty, 1)
+
+ manufacture_ste_doc1.submit()
+
+ # Second Manufacture stock entry
+ manufacture_ste_doc2 = frappe.get_doc(make_stock_entry(wo_doc.name, "Manufacture", 2))
+
+ # Batch no & Serial Nos should be same as transferred Batch no & Serial Nos
+ batch_no = manufacture_ste_doc2.items[0].batch_no
+ self.assertEqual(
+ get_serial_nos(manufacture_ste_doc2.items[0].serial_no)[0], batch_dict.get(batch_no)[1]
+ )
+ self.assertEqual(manufacture_ste_doc2.items[0].qty, 1)
+
+ batch_no = manufacture_ste_doc2.items[1].batch_no
+ self.assertEqual(
+ get_serial_nos(manufacture_ste_doc2.items[1].serial_no)[0], batch_dict.get(batch_no)[0]
+ )
+ self.assertEqual(manufacture_ste_doc2.items[1].qty, 1)
+
+
+def prepare_data_for_backflush_based_on_materials_transferred():
+ batch_item_doc = make_item(
+ "Test Batch MCC Keyboard",
+ {
+ "is_stock_item": 1,
+ "has_batch_no": 1,
+ "create_new_batch": 1,
+ "batch_number_series": "TBMK.#####",
+ "valuation_rate": 100,
+ "stock_uom": "Nos",
+ },
+ )
+
+ item = make_item(
+ "Test FG Item with Batch Raw Materials",
+ {
+ "is_stock_item": 1,
+ },
+ )
+
+ make_bom(item=item.name, source_warehouse="Stores - _TC", raw_materials=[batch_item_doc.name])
+
+ sn_item_doc = make_item(
+ "Test Serial No BTT Headphone",
+ {
+ "is_stock_item": 1,
+ "has_serial_no": 1,
+ "serial_no_series": "TSBH.#####",
+ "valuation_rate": 100,
+ "stock_uom": "Nos",
+ },
+ )
+
+ item = make_item(
+ "Test FG Item with Serial No Raw Materials",
+ {
+ "is_stock_item": 1,
+ },
+ )
+
+ make_bom(item=item.name, source_warehouse="Stores - _TC", raw_materials=[sn_item_doc.name])
+
+ sn_batch_item_doc = make_item(
+ "Test Batch Serial No WebCam",
+ {
+ "is_stock_item": 1,
+ "has_batch_no": 1,
+ "create_new_batch": 1,
+ "batch_number_series": "TBSW.#####",
+ "has_serial_no": 1,
+ "serial_no_series": "TBSWC.#####",
+ "valuation_rate": 100,
+ "stock_uom": "Nos",
+ },
+ )
+
+ item = make_item(
+ "Test FG Item with Serial & Batch No Raw Materials",
+ {
+ "is_stock_item": 1,
+ },
+ )
+
+ make_bom(item=item.name, source_warehouse="Stores - _TC", raw_materials=[sn_batch_item_doc.name])
+
def update_job_card(job_card, jc_qty=None):
employee = frappe.db.get_value("Employee", {"status": "Active"}, "name")
diff --git a/erpnext/stock/doctype/stock_entry/stock_entry.py b/erpnext/stock/doctype/stock_entry/stock_entry.py
index 293c2e5..08ce83b 100644
--- a/erpnext/stock/doctype/stock_entry/stock_entry.py
+++ b/erpnext/stock/doctype/stock_entry/stock_entry.py
@@ -596,21 +596,6 @@
title=_("Insufficient Stock"),
)
- def set_serial_nos(self, work_order):
- previous_se = frappe.db.get_value(
- "Stock Entry",
- {"work_order": work_order, "purpose": "Material Transfer for Manufacture"},
- "name",
- )
-
- for d in self.get("items"):
- transferred_serial_no = frappe.db.get_value(
- "Stock Entry Detail", {"parent": previous_se, "item_code": d.item_code}, "serial_no"
- )
-
- if transferred_serial_no:
- d.serial_no = transferred_serial_no
-
@frappe.whitelist()
def get_stock_and_rate(self):
"""
@@ -1321,7 +1306,7 @@
and not self.pro_doc.skip_transfer
and self.flags.backflush_based_on == "Material Transferred for Manufacture"
):
- self.get_transfered_raw_materials()
+ self.add_transfered_raw_materials_in_items()
elif (
self.work_order
@@ -1365,7 +1350,6 @@
# fetch the serial_no of the first stock entry for the second stock entry
if self.work_order and self.purpose == "Manufacture":
- self.set_serial_nos(self.work_order)
work_order = frappe.get_doc("Work Order", self.work_order)
add_additional_cost(self, work_order)
@@ -1655,7 +1639,7 @@
}
)
- def get_transfered_raw_materials(self):
+ def add_transfered_raw_materials_in_items(self) -> None:
available_materials = get_available_materials(self.work_order)
wo_data = frappe.db.get_value(
@@ -1666,9 +1650,11 @@
)
for key, row in available_materials.items():
- qty = (flt(row.qty) * flt(self.fg_completed_qty)) / (
- flt(wo_data.trans_qty) - flt(wo_data.produced_qty)
- )
+ remaining_qty_to_produce = flt(wo_data.trans_qty) - flt(wo_data.produced_qty)
+ if remaining_qty_to_produce <= 0:
+ continue
+
+ qty = (flt(row.qty) * flt(self.fg_completed_qty)) / remaining_qty_to_produce
item = row.item_details
if cint(frappe.get_cached_value("UOM", item.stock_uom, "must_be_whole_number")):
@@ -1676,7 +1662,7 @@
if row.batch_details:
for batch_no, batch_qty in row.batch_details.items():
- if qty <= 0:
+ if qty <= 0 or batch_qty <= 0:
continue
if batch_qty > qty:
@@ -1690,7 +1676,7 @@
else:
self.update_item_in_stock_entry_detail(row, item, qty)
- def update_item_in_stock_entry_detail(self, row, item, qty):
+ def update_item_in_stock_entry_detail(self, row, item, qty) -> None:
ste_item_details = {
"from_warehouse": item.warehouse,
"to_warehouse": "",
@@ -1704,11 +1690,28 @@
"original_item": item.original_item,
}
- if item.serial_no:
- ste_item_details["serial_no"] = "\n".join(item.serial_no[0 : cint(qty)])
+ if row.serial_nos:
+ serial_nos = row.serial_nos
+ if item.batch_no:
+ serial_nos = self.get_serial_nos_based_on_transferred_batch(item.batch_no, row.serial_nos)
+
+ serial_nos = serial_nos[0 : cint(qty)]
+ ste_item_details["serial_no"] = "\n".join(serial_nos)
+
+ # remove consumed serial nos from list
+ for sn in serial_nos:
+ row.serial_nos.remove(sn)
self.add_to_stock_entry_detail({item.item_code: ste_item_details})
+ @staticmethod
+ def get_serial_nos_based_on_transferred_batch(batch_no, serial_nos) -> list:
+ serial_nos = frappe.get_all(
+ "Serial No", filters={"batch_no": batch_no, "name": ("in", serial_nos)}, order_by="creation"
+ )
+
+ return [d.name for d in serial_nos]
+
def get_pending_raw_materials(self, backflush_based_on=None):
"""
issue (item quantity) that is pending to issue or desire to transfer,
@@ -2489,6 +2492,7 @@
if row.serial_no:
item_data.serial_nos.extend(get_serial_nos(row.serial_no))
+ item_data.serial_nos.sort()
else:
# Consume raw material qty in case of 'Manufacture' or 'Material Consumption for Manufacture'
@@ -2537,4 +2541,4 @@
)
)
.orderby(stock_entry.creation, stock_entry_detail.item_code, stock_entry_detail.idx)
- ).run(as_dict=1, debug=1)
+ ).run(as_dict=1)