fix: Bin reserved qty for production for extra material transfer

(cherry picked from commit f5378b6573)

# Conflicts:
#	erpnext/manufacturing/doctype/work_order/test_work_order.py
This commit is contained in:
Rohit Waghchaure
2026-01-23 19:35:47 +05:30
committed by Mergify
parent 0e3d276348
commit b5d8477354
2 changed files with 432 additions and 0 deletions

View File

@@ -3186,6 +3186,435 @@ class TestWorkOrder(FrappeTestCase):
allow_overproduction("overproduction_percentage_for_work_order", 0)
<<<<<<< HEAD
=======
def test_reserved_serial_batch(self):
raw_materials = []
for item_code, properties in {
"Test Reserved FG Item": {"is_stock_item": 1},
"Test Reserved Serial Item": {"has_serial_no": 1, "serial_no_series": "TSNN-RSI-.####"},
"Test Reserved Batch Item": {
"has_batch_no": 1,
"batch_number_series": "BCH-RBI-.####",
"create_new_batch": 1,
},
"Test Reserved Serial Batch Item": {
"has_serial_no": 1,
"serial_no_series": "TSNB-RSBI-.####",
"has_batch_no": 1,
"batch_number_series": "BCH-RSBI-.####",
"create_new_batch": 1,
},
}.items():
make_item(item_code, properties=properties)
if item_code != "Test Reserved FG Item":
raw_materials.append(item_code)
test_stock_entry.make_stock_entry(
item_code=item_code,
target="Stores - _TC",
qty=5,
basic_rate=100,
)
original_auto_reserve = frappe.db.get_single_value("Stock Settings", "auto_reserve_serial_and_batch")
original_backflush = frappe.db.get_single_value(
"Manufacturing Settings", "backflush_raw_materials_based_on"
)
frappe.db.set_single_value(
"Manufacturing Settings",
"backflush_raw_materials_based_on",
"Material Transferred for Manufacture",
)
frappe.db.set_single_value("Stock Settings", "auto_reserve_serial_and_batch", 1)
make_bom(
item="Test Reserved FG Item",
source_warehouse="Stores - _TC",
raw_materials=raw_materials,
)
wo = make_wo_order_test_record(
item="Test Reserved FG Item",
qty=5,
source_warehouse="Stores - _TC",
reserve_stock=1,
)
_reserved_item = get_reserved_entries(wo.name)
for key, value in _reserved_item.items():
self.assertEqual(key[1], "Stores - _TC")
self.assertEqual(value.reserved_qty, 5)
if value.serial_nos:
self.assertEqual(len(value.serial_nos), 5)
if value.batch_nos:
self.assertEqual(sum(value.batch_nos.values()), 5)
# Transfer 5 qty
mt_stock_entry = frappe.get_doc(make_stock_entry(wo.name, "Material Transfer for Manufacture", 5))
mt_stock_entry.submit()
for row in mt_stock_entry.items:
value = _reserved_item[(row.item_code, row.s_warehouse)]
self.assertEqual(row.qty, value.reserved_qty)
if value.serial_nos:
serial_nos = get_serial_nos_from_bundle(row.serial_and_batch_bundle)
self.assertEqual(sorted(serial_nos), sorted(value.serial_nos))
if value.batch_nos:
self.assertTrue(row.batch_no in value.batch_nos)
_before_reserved_item = get_reserved_entries(wo.name, mt_stock_entry.items[0].t_warehouse)
# Manufacture 2 qty
fg_stock_entry = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 2))
fg_stock_entry.submit()
for row in fg_stock_entry.items:
if not row.s_warehouse:
continue
value = _before_reserved_item[(row.item_code, row.s_warehouse)]
if row.serial_no:
serial_nos = get_serial_nos_from_bundle(row.serial_and_batch_bundle)
for sn in serial_nos:
self.assertTrue(sn in value.serial_nos)
value.serial_nos.remove(sn)
if row.batch_no:
self.assertTrue(row.batch_no in value.batch_nos)
value.batch_nos[row.batch_no] -= row.qty
if row.serial_no:
sns = get_serial_nos_from_bundle(row.serial_and_batch_bundle)
for sn in sns:
self.assertTrue(sn in value.serial_batches[row.batch_no])
value.serial_batches[row.batch_no].remove(sn)
# Manufacture 3 qty
fg_stock_entry = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 3))
fg_stock_entry.submit()
for row in fg_stock_entry.items:
if not row.s_warehouse:
continue
value = _before_reserved_item[(row.item_code, row.s_warehouse)]
if row.serial_no:
serial_nos = get_serial_nos_from_bundle(row.serial_and_batch_bundle)
self.assertEqual(sorted(serial_nos), sorted(value.serial_nos))
if row.batch_no:
self.assertTrue(row.batch_no in value.batch_nos)
self.assertEqual(value.batch_nos[row.batch_no], row.qty)
if row.serial_no:
sns = get_serial_nos_from_bundle(row.serial_and_batch_bundle)
self.assertEqual(sorted(sns), sorted(value.serial_batches[row.batch_no]))
frappe.db.set_single_value(
"Manufacturing Settings", "backflush_raw_materials_based_on", original_backflush
)
frappe.db.set_single_value("Stock Settings", "auto_reserve_serial_and_batch", original_auto_reserve)
def test_phantom_bom_item_not_in_additional_cost(self):
"""Test that phantom BOMs are not added to additional costs,
but regular non-stock items in the FG BOM are added."""
from erpnext.stock.doctype.item.test_item import make_item
# Create items:
# - FG Item (stock item)
# - Phantom sub-assembly (non-stock item to be phantom)
# - Phantom RM (stock item - component of phantom BOM)
# - Packing Material (non-stock item - directly in FG BOM)
# - Regular RM (stock item - directly in FG BOM)
fg_item = make_item(
"Test FG Item For Phantom Non Stock",
{"is_stock_item": 1, "valuation_rate": 100},
).name
phantom_item = make_item(
"Test Phantom Sub Assembly Non Stock",
{"is_stock_item": 0, "valuation_rate": 0},
).name
phantom_rm = make_item(
"Test Phantom RM Item",
{"is_stock_item": 1, "valuation_rate": 200},
).name
packing_material = make_item(
"Test Packing Material Non Stock",
{"is_stock_item": 0, "valuation_rate": 150},
).name
regular_rm = make_item(
"Test Regular RM Stock Item",
{"is_stock_item": 1, "valuation_rate": 100},
).name
# Create price list entries for non-stock items
price_list = "_Test Price List India"
for item_code, rate in [
(phantom_item, 500),
(phantom_rm, 200),
(packing_material, 150),
]:
if not frappe.db.get_value("Item Price", {"item_code": item_code, "price_list": price_list}):
frappe.get_doc(
{
"doctype": "Item Price",
"item_code": item_code,
"price_list_rate": rate,
"price_list": price_list,
}
).insert(ignore_permissions=True)
# Create Phantom BOM (for the phantom sub-assembly)
phantom_bom = frappe.get_doc(
{
"doctype": "BOM",
"item": phantom_item,
"is_default": 1,
"is_active": 1,
"is_phantom_bom": 1, # Mark as phantom BOM
"currency": "INR",
"quantity": 1,
"company": "_Test Company",
"rm_cost_as_per": "Price List",
"buying_price_list": price_list,
}
)
phantom_bom.append(
"items",
{
"item_code": phantom_rm,
"qty": 1,
"rate": 200,
},
)
phantom_bom.insert()
phantom_bom.submit()
# Create FG BOM with phantom item, packing material, and regular RM
fg_bom = frappe.get_doc(
{
"doctype": "BOM",
"item": fg_item,
"is_default": 1,
"is_active": 1,
"currency": "INR",
"quantity": 1,
"company": "_Test Company",
"rm_cost_as_per": "Price List",
"buying_price_list": price_list,
}
)
# Add phantom item (will be marked as is_phantom_item based on is_phantom_bom)
fg_bom.append(
"items",
{
"item_code": phantom_item,
"qty": 1,
"rate": 200,
"bom_no": phantom_bom.name,
},
)
# Add packing material (non-stock, directly in FG BOM)
fg_bom.append(
"items",
{
"item_code": packing_material,
"qty": 1,
"rate": 150,
},
)
# Add regular RM (stock item)
fg_bom.append(
"items",
{
"item_code": regular_rm,
"qty": 1,
"rate": 100,
},
)
fg_bom.insert()
fg_bom.submit()
# Ensure stock
test_stock_entry.make_stock_entry(
item_code=regular_rm,
target="_Test Warehouse - _TC",
qty=10,
basic_rate=100,
)
test_stock_entry.make_stock_entry(
item_code=phantom_rm,
target="_Test Warehouse - _TC",
qty=10,
basic_rate=200,
)
# Create work order
wo = make_wo_order_test_record(
production_item=fg_item,
bom_no=fg_bom.name,
qty=1,
source_warehouse="_Test Warehouse - _TC",
)
# Transfer materials
se_transfer = frappe.get_doc(make_stock_entry(wo.name, "Material Transfer for Manufacture", 1))
se_transfer.insert()
se_transfer.submit()
# Manufacture
se_manufacture = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 1))
se_manufacture.insert()
# Verify additional costs
self.assertTrue(se_manufacture.additional_costs, "Additional costs should not be empty")
total_additional_cost = sum(row.amount for row in se_manufacture.additional_costs)
self.assertEqual(
total_additional_cost,
150, # only packing material; phantom RM excluded
f"Additional cost should be 150 (packing material only), got {total_additional_cost}",
)
self.assertEqual(
se_manufacture.total_outgoing_value,
300, # 100 (regular RM) + 200 (phantom RM)
f"Total outgoing value should be 300, got {se_manufacture.total_outgoing_value}",
)
self.assertEqual(
se_manufacture.total_incoming_value,
450, # 300 (RM total) + 150 (packing material)
f"Total incoming value should be 450, got {se_manufacture.total_incoming_value}",
)
# Clean up
se_manufacture.submit()
se_manufacture.cancel()
se_transfer.cancel()
wo.reload()
wo.cancel()
fg_bom.cancel()
phantom_bom.cancel()
def test_phantom_bom_explosion(self):
from erpnext.manufacturing.doctype.bom.test_bom import create_tree_for_phantom_bom_tests
expected = create_tree_for_phantom_bom_tests()
wo = make_wo_order_test_record(item="Top Level Parent")
self.assertEqual([item.item_code for item in wo.required_items], expected)
def test_reserved_qty_for_pp_with_extra_material_transfer(self):
from erpnext.stock.doctype.stock_entry.test_stock_entry import (
make_stock_entry as make_stock_entry_test_record,
)
rm_item_code = make_item(
"_Test Reserved Qty PP Item",
{
"is_stock_item": 1,
},
).name
fg_item_code = make_item(
"_Test Reserved Qty PP FG Item",
{
"is_stock_item": 1,
},
).name
make_stock_entry_test_record(
item_code=rm_item_code, target="_Test Warehouse - _TC", qty=10, basic_rate=100
)
make_bom(
item=fg_item_code,
raw_materials=[rm_item_code],
)
wo_order = make_wo_order_test_record(
item=fg_item_code,
qty=1,
source_warehouse="_Test Warehouse - _TC",
skip_transfer=0,
target_warehouse="_Test Warehouse - _TC",
)
bin1_at_completion = get_bin(rm_item_code, "_Test Warehouse - _TC")
self.assertEqual(bin1_at_completion.reserved_qty_for_production, 1)
s = frappe.get_doc(make_stock_entry(wo_order.name, "Material Transfer for Manufacture", 1))
s.items[0].qty += 2 # extra material transfer
s.submit()
bin1_at_completion = get_bin(rm_item_code, "_Test Warehouse - _TC")
self.assertEqual(bin1_at_completion.reserved_qty_for_production, 0)
def get_reserved_entries(voucher_no, warehouse=None):
doctype = frappe.qb.DocType("Stock Reservation Entry")
sabb = frappe.qb.DocType("Serial and Batch Entry")
query = (
frappe.qb.from_(doctype)
.left_join(sabb)
.on(doctype.name == sabb.parent)
.select(
doctype.name,
doctype.item_code,
doctype.warehouse,
doctype.reserved_qty,
sabb.serial_no,
sabb.batch_no,
sabb.qty,
sabb.delivered_qty,
)
.where((doctype.voucher_no == voucher_no) & (doctype.docstatus == 1))
)
if warehouse:
query = query.where(doctype.warehouse == warehouse)
reservation_entries = query.run(as_dict=True)
_reserved_item = frappe._dict({})
for entry in reservation_entries:
key = (entry.item_code, entry.warehouse)
if key not in _reserved_item:
_reserved_item[key] = frappe._dict(
{
"reserved_qty": 0,
"serial_nos": [],
"batch_nos": defaultdict(int),
"serial_batches": defaultdict(list),
}
)
_reserved_item[key].reserved_qty += entry.qty
if entry.batch_no:
_reserved_item[key].batch_nos[entry.batch_no] += entry.qty
if entry.serial_no:
_reserved_item[key].serial_batches[entry.batch_no].append(entry.serial_no)
if entry.serial_no:
_reserved_item[key].serial_nos.append(entry.serial_no)
return _reserved_item
>>>>>>> f5378b6573 (fix: Bin reserved qty for production for extra material transfer)
def make_stock_in_entries_and_get_batches(rm_item, source_warehouse, wip_warehouse):
from erpnext.stock.doctype.stock_entry.test_stock_entry import (

View File

@@ -1786,6 +1786,9 @@ def get_reserved_qty_for_production(
qty_field = wo_item.required_qty
else:
qty_field = Case()
qty_field = qty_field.when(
((wo.skip_transfer == 0) & (wo_item.transferred_qty > wo_item.required_qty)), 0.0
)
qty_field = qty_field.when(wo.skip_transfer == 0, wo_item.required_qty - wo_item.transferred_qty)
qty_field = qty_field.else_(wo_item.required_qty - wo_item.consumed_qty)