nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-09-17 02:26:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #57930 from frappe/mergify/bp/version-16-hotfix/pr-57895

Browse Source 
fix: track semi finished goods production flow (backport #57895)
This commit is contained in:
Mihir Kandoi
2026-08-09 22:47:52 +05:30
committed by GitHub
parent b4fbbe51e8 5c38e43758
commit 6e4a7fba7e
10 changed files with 890 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
erpnext/manufacturing/doctype/bom/bom.py
View File

@@ -282,6 +282,7 @@ class BOM(WebsiteGenerator):
self.clear_inspection()
self.validate_main_item()
self.validate_currency()
self.set_operation_finished_goods()
self.set_materials_based_on_operation_bom()
self.set_conversion_rate()
self.set_plc_conversion_rate()
@@ -307,15 +308,42 @@ class BOM(WebsiteGenerator):
if self.docstatus == 1:
self.validate_raw_materials_of_operation()
def set_operation_finished_goods(self):
"""Fill each operation's FG item where it is unambiguous: the final operation produces
this BOM's item, an operation with a BOM produces that BOM's item. Runs before
set_materials_based_on_operation_bom so derived rows get their materials expanded."""
if not self.track_semi_finished_goods:
return
for row in self.operations:
if row.is_final_finished_good and not row.finished_good:
row.finished_good = self.item
elif row.bom_no and not row.finished_good:
row.finished_good = frappe.get_cached_value("BOM", row.bom_no, "item")
def validate_semi_finished_goods(self):
if not self.track_semi_finished_goods or not self.operations:
return
fg_items = []
for row in self.operations:
if not row.finished_good:
frappe.throw(
_(
"Row #{0}: FG / Semi FG Item is required for the operation {1} as 'Track Semi Finished Goods' is enabled."
).format(row.idx, bold(row.operation)),
)
if not row.is_final_finished_good:
continue
if row.finished_good != self.item:
frappe.throw(
_(
"Row #{0}: The operation {1} has 'Is Final Finished Good' checked, so its FG / Semi FG Item must be {2}."
).format(row.idx, bold(row.operation), bold(self.item)),
)
fg_items.append(row.finished_good)
if not fg_items:
@@ -826,7 +854,7 @@ class BOM(WebsiteGenerator):
self.add_materials_from_bom(row.finished_good, row.bom_no, row.idx, qty=row.finished_good_qty)
@frappe.whitelist()
def add_raw_materials(self, operation_row_id, items):
def add_raw_materials(self, operation_row_id: str | int, items: str | list[dict]) -> None:
if isinstance(items, str):
items = parse_json(items)
@@ -836,17 +864,10 @@ class BOM(WebsiteGenerator):
row.update(get_item_details(row.get("item_code")))
row.operation_row_id = operation_row_id
item_row = None
if row.name:
item_row = self.get_item_data(row.name)
item_row = self.get_item_data(row.item_code, operation_row_id)
if item_row:
item_row.update(
{
"item_code": row.get("item_code"),
"qty": row.get("qty"),
}
)
item_row.qty = row.get("qty")
else:
row.idx = None
row.name = None
@@ -867,9 +888,9 @@ class BOM(WebsiteGenerator):
return False
def get_item_data(self, name):
def get_item_data(self, item_code, operation_row_id):
for row in self.items:
if row.item_code == name:
if row.item_code == item_code and cint(row.operation_row_id) == cint(operation_row_id):
return row
@frappe.whitelist()

203
erpnext/manufacturing/doctype/bom/test_bom.py
View File

@@ -7,7 +7,7 @@ from functools import partial
import frappe
from frappe.tests import timeout
from frappe.utils import cstr, flt
from frappe.utils import cint, cstr, flt
from erpnext.controllers.tests.test_subcontracting_controller import (
set_backflush_based_on,
@@ -811,6 +811,207 @@ class TestBOM(ERPNextTestSuite):
for row in bom.items:
self.assertEqual(row.stock_uom, "Kg")
@timeout
def test_track_semi_finished_goods_requires_finished_good_on_operations(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.manufacturing.doctype.workstation.test_workstation import make_workstation
fg_item = make_item(properties={"is_stock_item": 1}).name
sfg_item = make_item(properties={"is_stock_item": 1}).name
rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100.0}).name
make_workstation({"workstation": "_Test SFG Workstation"})
for operation in ("_Test SFG Operation", "_Test SFG Final Operation"):
make_operation({"operation": operation, "workstation": "_Test SFG Workstation"})
bom = frappe.new_doc("BOM")
bom.company = "_Test Company"
bom.item = fg_item
bom.quantity = 1
bom.with_operations = 1
bom.track_semi_finished_goods = 1
bom.append(
"operations",
{
"operation": "_Test SFG Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
},
)
bom.append(
"operations",
{
"operation": "_Test SFG Final Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"is_final_finished_good": 1,
},
)
bom.append("items", {"item_code": rm_item, "qty": 1, "operation_row_id": 1})
bom.append("items", {"item_code": sfg_item, "qty": 1, "operation_row_id": 2})
# the first operation produces nothing derivable: no FG item, no BOM to take it from
self.assertRaises(frappe.ValidationError, bom.insert)
bom.operations[0].finished_good = sfg_item
bom.insert()
# the final operation's FG item is derived from the BOM's own item
self.assertEqual(bom.operations[1].finished_good, fg_item)
@timeout
def test_add_raw_materials_when_item_is_used_by_another_operation(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.manufacturing.doctype.workstation.test_workstation import make_workstation
fg_item = make_item(properties={"is_stock_item": 1}).name
sfg_item = make_item(properties={"is_stock_item": 1}).name
rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100.0}).name
make_workstation({"workstation": "_Test SFG Workstation"})
for operation in ("_Test SFG Operation", "_Test SFG Final Operation"):
make_operation({"operation": operation, "workstation": "_Test SFG Workstation"})
bom = frappe.new_doc("BOM")
bom.company = "_Test Company"
bom.item = fg_item
bom.quantity = 1
bom.with_operations = 1
bom.track_semi_finished_goods = 1
bom.append(
"operations",
{
"operation": "_Test SFG Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"finished_good": sfg_item,
},
)
bom.append(
"operations",
{
"operation": "_Test SFG Final Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"is_final_finished_good": 1,
},
)
bom.append("items", {"item_code": rm_item, "qty": 1, "operation_row_id": 1})
bom.append("items", {"item_code": sfg_item, "qty": 1, "operation_row_id": 2})
bom.insert()
def rows_for(item_code, operation_row_id):
return [
row
for row in bom.items
if row.item_code == item_code and cint(row.operation_row_id) == operation_row_id
]
# the item already used by operation 1 gets its own new row under operation 2
bom.add_raw_materials(2, [{"item_code": rm_item, "qty": 3}])
self.assertEqual(len(rows_for(rm_item, 2)), 1)
self.assertEqual(flt(rows_for(rm_item, 2)[0].qty), 3.0)
self.assertEqual(flt(rows_for(rm_item, 1)[0].qty), 1.0)
# adding it again for the same operation updates the row instead of stacking another
bom.add_raw_materials(2, [{"item_code": rm_item, "qty": 5}])
self.assertEqual(len(rows_for(rm_item, 2)), 1)
self.assertEqual(flt(rows_for(rm_item, 2)[0].qty), 5.0)
@timeout
def test_operation_bom_materials_expand_on_single_pass_submit(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.manufacturing.doctype.workstation.test_workstation import make_workstation
fg_item = make_item(properties={"is_stock_item": 1}).name
sfg_item = make_item(properties={"is_stock_item": 1}).name
rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100.0}).name
make_workstation({"workstation": "_Test SFG Workstation"})
for operation in ("_Test SFG Operation", "_Test SFG Final Operation"):
make_operation({"operation": operation, "workstation": "_Test SFG Workstation"})
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg_item, quantity=1)
sfg_bom.append("items", {"item_code": rm_item, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
bom = frappe.new_doc("BOM")
bom.company = "_Test Company"
bom.item = fg_item
bom.quantity = 1
bom.with_operations = 1
bom.track_semi_finished_goods = 1
bom.append(
"operations",
{
"operation": "_Test SFG Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"bom_no": sfg_bom.name,
},
)
bom.append(
"operations",
{
"operation": "_Test SFG Final Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"is_final_finished_good": 1,
},
)
bom.append("items", {"item_code": sfg_item, "qty": 1, "operation_row_id": 2})
bom.submit()
self.assertEqual(bom.docstatus, 1)
self.assertEqual(bom.operations[0].finished_good, sfg_item)
self.assertTrue(
any(row.item_code == rm_item and cint(row.operation_row_id) == 1 for row in bom.items)
)
@timeout
def test_final_operation_must_produce_the_bom_item(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.manufacturing.doctype.workstation.test_workstation import make_workstation
fg_item = make_item(properties={"is_stock_item": 1}).name
sfg_item = make_item(properties={"is_stock_item": 1}).name
rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100.0}).name
make_workstation({"workstation": "_Test SFG Workstation"})
for operation in ("_Test SFG Operation", "_Test SFG Final Operation"):
make_operation({"operation": operation, "workstation": "_Test SFG Workstation"})
bom = frappe.new_doc("BOM")
bom.company = "_Test Company"
bom.item = fg_item
bom.quantity = 1
bom.with_operations = 1
bom.track_semi_finished_goods = 1
bom.append(
"operations",
{
"operation": "_Test SFG Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"finished_good": sfg_item,
},
)
bom.append(
"operations",
{
"operation": "_Test SFG Final Operation",
"workstation": "_Test SFG Workstation",
"time_in_mins": 30,
"is_final_finished_good": 1,
"finished_good": sfg_item,
},
)
bom.append("items", {"item_code": rm_item, "qty": 1, "operation_row_id": 1})
bom.append("items", {"item_code": sfg_item, "qty": 1, "operation_row_id": 2})
# the final operation claims to produce the semi FG, not this BOM's item
self.assertRaises(frappe.ValidationError, bom.insert)
bom.operations[1].finished_good = fg_item
bom.insert()
def get_default_bom(item_code="_Test FG Item 2"):
return frappe.db.get_value("BOM", {"item": item_code, "is_active": 1, "is_default": 1})

3
erpnext/manufacturing/doctype/bom_operation/bom_operation.json
View File

@@ -213,6 +213,7 @@
"fieldtype": "Link",
"in_list_view": 1,
"label": "FG / Semi FG Item",
"mandatory_depends_on": "eval:parent.track_semi_finished_goods === 1",
"options": "Item"
},
{
@@ -307,7 +308,7 @@
"index_web_pages_for_search": 1,
"istable": 1,
"links": [],
"modified": "2026-05-25 17:15:42.044630",
"modified": "2026-08-08 12:00:00.000000",
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "BOM Operation",

31
erpnext/manufacturing/doctype/job_card/job_card.py
View File

@@ -845,6 +845,9 @@ class JobCard(Document):
)
def validate_transfer_qty(self):
if self.track_semi_finished_goods and self.skip_material_transfer:
return
if (
not self.finished_good
and not self.is_corrective_job_card
@@ -1072,6 +1075,9 @@ class JobCard(Document):
wo.calculate_operating_cost()
wo.set_actual_dates()
if wo.track_semi_finished_goods:
wo.set_process_loss_qty()
if time_data:
wo.status = "In Process"
@@ -1659,25 +1665,24 @@ class JobCard(Document):
)
)
def get_consumed_process_loss(self):
table = frappe.qb.DocType("Stock Entry")
query = (
frappe.qb.from_(table)
.select(Sum(table.process_loss_qty))
.where((table.purpose == "Manufacture") & (table.job_card == self.name) & (table.docstatus == 1))
)
return query.run()[0][0] or 0
@frappe.whitelist()
def make_stock_entry_for_semi_fg_item(self, auto_submit: bool = False):
def get_consumed_process_loss():
table = frappe.qb.DocType("Stock Entry")
query = (
frappe.qb.from_(table)
.select(Sum(table.process_loss_qty))
.where(
(table.purpose == "Manufacture") & (table.job_card == self.name) & (table.docstatus == 1)
)
)
return query.run()[0][0] or 0
from erpnext.stock.doctype.stock_entry_type.stock_entry_type import ManufactureEntry
consumed_process_loss = self.get_consumed_process_loss()
ste = ManufactureEntry(
{
"for_quantity": self.get_qty_to_produce() - self.manufactured_qty,
"process_loss_qty": max(self.process_loss_qty - get_consumed_process_loss(), 0),
"for_quantity": self.get_qty_to_produce() - self.manufactured_qty - consumed_process_loss,
"process_loss_qty": max(self.process_loss_qty - consumed_process_loss, 0),
"job_card": self.name,
"skip_material_transfer": self.skip_material_transfer,
"backflush_from_wip_warehouse": self.backflush_from_wip_warehouse,

520
erpnext/manufacturing/doctype/job_card/test_job_card.py
View File

@@ -1334,6 +1334,212 @@ class TestJobCard(ERPNextTestSuite):
8,
)
def test_semi_fg_process_loss_rolls_up_to_work_order(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm = make_item("Process Loss Rollup RM 1", {"is_stock_item": 1}).name
fg = make_item("Process Loss Rollup FG 1", {"is_stock_item": 1}).name
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
fg_bom.append("items", {"item_code": rm, "qty": 1, "operation_row_id": 1})
operation = {
"operation": "Process Loss Rollup Op A",
"workstation": "_Test Workstation A",
"finished_good": fg,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"sequence_id": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
make_workstation(operation)
make_operation(operation)
fg_bom.append("operations", operation)
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg,
qty=10,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
do_not_save=True,
)
work_order.operations[0].time_in_mins = 60
work_order.save()
work_order.submit()
make_stock_entry(item_code=rm, target=warehouse, qty=100, basic_rate=100)
job_card = frappe.get_doc(
"Job Card",
frappe.get_all(
"Job Card",
filters={"work_order": work_order.name},
order_by="sequence_id, creation",
limit=1,
pluck="name",
)[0],
)
job_card.append("time_logs", {"from_time": "2024-05-01 08:00:00"})
job_card.save()
job_card.complete_job_card(
qty=8,
for_quantity=10,
pending_qty=0,
process_loss_qty=2,
end_time="2024-05-01 09:00:00",
)
job_card.reload()
self.assertEqual(flt(job_card.process_loss_qty), 2)
job_card.submit()
frappe.get_doc(job_card.make_stock_entry_for_semi_fg_item()).submit()
self.assertEqual(
flt(
frappe.db.get_value("Work Order Operation", work_order.operations[0].name, "process_loss_qty")
),
2,
)
work_order.reload()
self.assertEqual(flt(work_order.produced_qty), 8)
self.assertEqual(flt(work_order.process_loss_qty), 2)
self.assertEqual(work_order.status, "Completed")
def test_semi_fg_process_loss_of_an_intermediate_operation_rolls_up_to_work_order(self):
"""Loss booked by an earlier operation shrinks what the final operation can produce,
so it has to show up on the work order even though the final operation loses nothing."""
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm = make_item("Intermediate Loss RM 1", {"is_stock_item": 1}).name
sfg = make_item("Intermediate Loss SFG 1", {"is_stock_item": 1}).name
fg = make_item("Intermediate Loss FG 1", {"is_stock_item": 1}).name
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg, quantity=1)
sfg_bom.append("items", {"item_code": rm, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
operations = [
{
"operation": "Intermediate Loss Op A",
"finished_good": sfg,
"bom_no": sfg_bom.name,
"sequence_id": 1,
},
{
"operation": "Intermediate Loss Op B",
"finished_good": fg,
"is_final_finished_good": 1,
"sequence_id": 2,
},
]
for row in operations:
row.update(
{
"workstation": "_Test Workstation A",
"finished_good_qty": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
)
make_workstation(row)
make_operation(row)
fg_bom.append("operations", row)
fg_bom.append("items", {"item_code": sfg, "qty": 1, "operation_row_id": 2})
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg,
qty=10,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
do_not_save=True,
)
for row in work_order.operations:
row.time_in_mins = 60
work_order.save()
work_order.submit()
make_stock_entry(item_code=rm, target=warehouse, qty=100, basic_rate=100)
def get_job_card(operation):
return frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card",
{"work_order": work_order.name, "operation": operation, "docstatus": 0},
"name",
),
)
jc_a = get_job_card("Intermediate Loss Op A")
jc_a.append("time_logs", {"from_time": "2024-06-01 08:00:00"})
jc_a.save()
jc_a.complete_job_card(
qty=8, for_quantity=10, pending_qty=0, process_loss_qty=2, end_time="2024-06-01 09:00:00"
)
jc_a.reload()
jc_a.submit()
frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item()).submit()
work_order.reload()
self.assertEqual(flt(work_order.process_loss_qty), 2)
jc_b = get_job_card("Intermediate Loss Op B")
jc_b.for_quantity = 8
for row in jc_b.items:
row.required_qty = 8
jc_b.append(
"time_logs",
{"from_time": "2024-06-02 08:00:00", "to_time": "2024-06-02 09:00:00", "completed_qty": 8},
)
jc_b.save()
jc_b.submit()
frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item()).submit()
work_order.reload()
self.assertEqual(flt(work_order.produced_qty), 8)
self.assertEqual(flt(work_order.process_loss_qty), 2)
self.assertEqual(work_order.status, "Completed")
def test_semi_fg_pending_qty_is_left_to_another_job_card(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
@@ -1699,6 +1905,299 @@ class TestJobCard(ERPNextTestSuite):
consumed_batches = get_batches_from_bundle(sfg_consume_row.serial_and_batch_bundle)
self.assertEqual(set(consumed_batches.keys()), set(produced_batches.keys()))
def test_manufacture_entry_process_loss_not_taken_from_previous_operation(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm1 = make_item("PL Scope RM 1", {"is_stock_item": 1}).name
rm2 = make_item("PL Scope RM 2", {"is_stock_item": 1}).name
sfg = make_item("PL Scope SFG 1", {"is_stock_item": 1}).name
fg1 = make_item("PL Scope FG 1", {"is_stock_item": 1}).name
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg, quantity=1)
sfg_bom.append("items", {"item_code": rm1, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg1,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
operation1 = {
"operation": "PL Scope Op A",
"workstation": "_Test Workstation A",
"finished_good": sfg,
"bom_no": sfg_bom.name,
"finished_good_qty": 1,
"sequence_id": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
operation2 = {
"operation": "PL Scope Op B",
"workstation": "_Test Workstation A",
"finished_good": fg1,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"sequence_id": 2,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
make_workstation(operation1)
make_operation(operation1)
make_operation(operation2)
fg_bom.append("operations", operation1)
fg_bom.append("operations", operation2)
fg_bom.append("items", {"item_code": rm2, "qty": 1})
fg_bom.append("items", {"item_code": sfg, "qty": 1, "operation_row_id": 2})
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg1,
qty=5,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
)
make_stock_entry(item_code=rm1, target=warehouse, qty=10, basic_rate=100)
make_stock_entry(item_code=rm2, target=warehouse, qty=10, basic_rate=100)
make_stock_entry(item_code=sfg, target=warehouse, qty=10, basic_rate=100)
jc_a = frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card", {"work_order": work_order.name, "operation": "PL Scope Op A"}, "name"
),
)
jc_a.append(
"time_logs",
{"from_time": "2024-01-01 08:00:00", "to_time": "2024-01-01 09:00:00", "completed_qty": 3},
)
jc_a.pending_qty = 0
jc_a.process_loss_qty = 2
jc_a.submit()
me_a = frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item())
me_a.submit()
self.assertEqual(flt(me_a.process_loss_qty), 2.0)
jc_b = frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card", {"work_order": work_order.name, "operation": "PL Scope Op B"}, "name"
),
)
jc_b.append(
"time_logs",
{"from_time": "2024-02-01 08:00:00", "to_time": "2024-02-01 09:00:00", "completed_qty": 3},
)
jc_b.pending_qty = 2
jc_b.submit()
me_b = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
# operation A's loss must not leak into operation B's entry
self.assertEqual(flt(me_b.process_loss_qty), 0.0)
fg_row = next(row for row in me_b.items if row.is_finished_item)
self.assertEqual(flt(fg_row.qty), 3.0)
me_b.submit()
def make_semi_fg_work_order(self, prefix, qty=5):
"""Two-operation semi FG work order: Op A makes the SFG from RM 1, final Op B
consumes it. Both operations skip material transfer; stock is pre-seeded."""
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
warehouse = "Stores - _TC"
rm1 = make_item(f"{prefix} RM 1", {"is_stock_item": 1}).name
rm2 = make_item(f"{prefix} RM 2", {"is_stock_item": 1}).name
sfg = make_item(f"{prefix} SFG 1", {"is_stock_item": 1}).name
fg1 = make_item(f"{prefix} FG 1", {"is_stock_item": 1}).name
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg, quantity=1)
sfg_bom.append("items", {"item_code": rm1, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg1,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
operation1 = {
"operation": f"{prefix} Op A",
"workstation": "_Test Workstation A",
"finished_good": sfg,
"bom_no": sfg_bom.name,
"finished_good_qty": 1,
"sequence_id": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
operation2 = {
"operation": f"{prefix} Op B",
"workstation": "_Test Workstation A",
"finished_good": fg1,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"sequence_id": 2,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
make_workstation(operation1)
make_operation(operation1)
make_operation(operation2)
fg_bom.append("operations", operation1)
fg_bom.append("operations", operation2)
fg_bom.append("items", {"item_code": rm2, "qty": 1})
fg_bom.append("items", {"item_code": sfg, "qty": 1, "operation_row_id": 2})
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg1,
qty=qty,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
)
for item_code in (rm1, rm2, sfg):
make_stock_entry(item_code=item_code, target=warehouse, qty=10, basic_rate=100)
return work_order
def get_semi_fg_job_card(self, work_order, operation):
return frappe.get_doc(
"Job Card",
frappe.db.get_value("Job Card", {"work_order": work_order.name, "operation": operation}, "name"),
)
def test_partial_manufacture_entry_then_finish(self):
work_order = self.make_semi_fg_work_order("PL Partial")
jc_a = self.get_semi_fg_job_card(work_order, "PL Partial Op A")
jc_a.append(
"time_logs",
{"from_time": "2024-01-01 08:00:00", "to_time": "2024-01-01 09:00:00", "completed_qty": 5},
)
jc_a.submit()
frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item()).submit()
jc_b = self.get_semi_fg_job_card(work_order, "PL Partial Op B")
jc_b.append(
"time_logs",
{"from_time": "2024-02-01 08:00:00", "to_time": "2024-02-01 09:00:00", "completed_qty": 3},
)
jc_b.pending_qty = 0
jc_b.process_loss_qty = 2
jc_b.submit()
# book 1 of the 3 finished units now; the full process loss goes with this first entry,
# so it accounts for 3 of 5 and its materials are trimmed to the same share
first = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
fg_row = next(row for row in first.items if row.is_finished_item)
fg_row.qty = 1
for row in first.items:
if row.s_warehouse and not row.is_finished_item:
row.qty = flt(row.qty) * 3 / 5
first.save()
first.submit()
# the follow-up entry must be generated net of the already-booked loss and still submit
jc_b.reload()
second = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
fg_row = next(row for row in second.items if row.is_finished_item)
self.assertEqual(flt(fg_row.qty), 2.0)
self.assertEqual(flt(second.process_loss_qty), 0.0)
second.submit()
jc_b.reload()
self.assertEqual(flt(jc_b.manufactured_qty), 3.0)
# across both entries, consumption adds up to the job card's requirement of 5, no more
consumed = frappe.get_all(
"Stock Entry Detail",
filters={"parent": ["in", [first.name, second.name]], "s_warehouse": ["is", "set"]},
fields=["item_code", {"SUM": "qty", "as": "qty"}],
group_by="item_code",
)
self.assertTrue(consumed)
for row in consumed:
self.assertEqual(flt(row.qty), 5.0, f"{row.item_code} mis-consumed across partial entries")
def test_update_after_submit_keeps_manufacture_entry_intact(self):
work_order = self.make_semi_fg_work_order("PL Update")
jc_a = self.get_semi_fg_job_card(work_order, "PL Update Op A")
jc_a.append(
"time_logs",
{"from_time": "2024-01-01 08:00:00", "to_time": "2024-01-01 09:00:00", "completed_qty": 3},
)
jc_a.pending_qty = 0
jc_a.process_loss_qty = 2
jc_a.submit()
entry = frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item())
entry.submit()
if not frappe.db.exists("Print Heading", "_Test SFG Heading"):
frappe.get_doc({"doctype": "Print Heading", "print_heading": "_Test SFG Heading"}).insert()
entry.reload()
entry.select_print_heading = "_Test SFG Heading"
entry.save()
entry.reload()
self.assertEqual(flt(entry.process_loss_qty), 2.0)
def test_stale_manufacture_draft_cannot_over_produce_without_operation_bom(self):
work_order = self.make_semi_fg_work_order("PL NoBom")
jc_a = self.get_semi_fg_job_card(work_order, "PL NoBom Op A")
jc_a.append(
"time_logs",
{"from_time": "2024-01-01 08:00:00", "to_time": "2024-01-01 09:00:00", "completed_qty": 5},
)
jc_a.submit()
frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item()).submit()
# Op B has no operation BOM, so its entries carry no For Quantity to validate against
jc_b = self.get_semi_fg_job_card(work_order, "PL NoBom Op B")
jc_b.append(
"time_logs",
{"from_time": "2024-02-01 08:00:00", "to_time": "2024-02-01 09:00:00", "completed_qty": 3},
)
jc_b.pending_qty = 0
jc_b.process_loss_qty = 2
jc_b.submit()
draft_one = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
draft_two = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
draft_one.submit()
stale = frappe.get_doc("Stock Entry", draft_two.name)
self.assertRaises(frappe.ValidationError, stale.submit)
def test_semi_fg_auto_pull_with_uom_conversion(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
@@ -2114,6 +2613,27 @@ class TestJobCard(ERPNextTestSuite):
self.assertEqual(s.additional_costs[2].amount, 480)
self.assertEqual(s.additional_costs[3].amount, 480)
def test_semi_fg_job_card_is_exempt_from_transfer_qty_check(self):
jc = frappe.new_doc("Job Card")
jc.track_semi_finished_goods = 1
jc.skip_material_transfer = 1
jc.for_quantity = 10
jc.transferred_qty = 0
jc.append("items", {"item_code": "_Test Item"})
jc.validate_transfer_qty()
# with transfer enabled, a legacy card without an FG item keeps the strict check
jc.skip_material_transfer = 0
self.assertRaises(frappe.ValidationError, jc.validate_transfer_qty)
jc.finished_good = "_Test Item"
jc.validate_transfer_qty()
jc.finished_good = None
jc.track_semi_finished_goods = 0
self.assertRaises(frappe.ValidationError, jc.validate_transfer_qty)
def test_completion_qty_split_must_add_up(self):
jc = frappe.new_doc("Job Card")
jc.for_quantity = 5

18
erpnext/manufacturing/doctype/work_order/test_work_order.py
View File

@@ -4839,6 +4839,24 @@ class TestWorkOrder(ERPNextTestSuite):
# generated qty (3.0 for 8 units) differs from the BOM-scaled qty (7.5 for 20 units)
self.assertEqual(flt(row.qty, 6), 3.0)
def test_wip_warehouse_required_when_tracking_semi_finished_goods(self):
wo = frappe.new_doc("Work Order")
wo.track_semi_finished_goods = 1
wo.skip_transfer = 0
wo.fg_warehouse = "_Test Warehouse 1 - _TC"
self.assertRaises(frappe.ValidationError, wo.validate_warehouse)
wo.wip_warehouse = "_Test Warehouse - _TC"
wo.validate_warehouse()
# the top-level target warehouse stays optional; operations may carry their own
wo.fg_warehouse = None
wo.validate_warehouse()
wo.track_semi_finished_goods = 0
self.assertRaises(frappe.ValidationError, wo.validate_warehouse)
def get_reserved_entries(voucher_no, warehouse=None):
doctype = frappe.qb.DocType("Stock Reservation Entry")

4
erpnext/manufacturing/doctype/work_order/work_order.json
View File

@@ -272,7 +272,7 @@
"fieldtype": "Link",
"label": "Work-in-Progress Warehouse",
"link_filters": "[[\"Warehouse\",\"disabled\",\"=\",0],[\"Warehouse\",\"is_group\",\"=\",0]]",
"mandatory_depends_on": "eval:(!doc.skip_transfer || doc.from_wip_warehouse) && !doc.track_semi_finished_goods",
"mandatory_depends_on": "eval:!doc.skip_transfer || doc.from_wip_warehouse",
"options": "Warehouse"
},
{
@@ -739,7 +739,7 @@
"image_field": "image",
"is_submittable": 1,
"links": [],
"modified": "2026-06-03 21:35:34.175667",
"modified": "2026-08-08 12:00:00.000000",
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "Work Order",

13
erpnext/manufacturing/doctype/work_order/work_order.py
View File

@@ -883,6 +883,12 @@ class WorkOrder(Document):
return flt(query.run()[0][0])
def set_process_loss_qty(self):
self.db_set("process_loss_qty", self._process_loss_qty())
def _process_loss_qty(self):
if self.track_semi_finished_goods:
return flt(sum(flt(row.process_loss_qty) for row in self.operations))
table = frappe.qb.DocType("Stock Entry")
process_loss_qty = (
frappe.qb.from_(table)
@@ -892,7 +898,7 @@ class WorkOrder(Document):
)
).run()[0][0]
self.db_set("process_loss_qty", flt(process_loss_qty))
return flt(process_loss_qty)
def update_production_plan_status(self):
production_plan = frappe.get_doc("Production Plan", self.production_plan)
@@ -915,12 +921,9 @@ class WorkOrder(Document):
production_plan.run_method("update_produced_pending_qty", produced_qty, self.production_plan_item)
def validate_warehouse(self):
if self.track_semi_finished_goods:
return
if not self.wip_warehouse and not self.skip_transfer:
frappe.throw(_("Work-in-Progress Warehouse is required before Submit"))
if not self.fg_warehouse:
if not self.fg_warehouse and not self.track_semi_finished_goods:
frappe.throw(_("Target Warehouse is required before Submit"))
def before_submit(self):

75
erpnext/stock/doctype/stock_entry/stock_entry.py
View File

@@ -296,6 +296,7 @@ class StockEntry(StockController, SubcontractingInwardController):
self.validate_batch()
self.validate_inspection()
self.validate_fg_completed_qty()
self.validate_job_card_pending_production()
self.validate_difference_account()
self.set_job_card_data()
self.validate_job_card_item()
@@ -3192,21 +3193,11 @@ class StockEntry(StockController, SubcontractingInwardController):
return
precision = self.precision("process_loss_qty")
if self.work_order:
data = frappe.get_all(
"Work Order Operation",
filters={"parent": self.work_order},
fields=[{"MAX": "process_loss_qty", "as": "process_loss_qty"}],
)
process_loss_qty = self.get_pending_process_loss_qty()
if process_loss_qty and flt(self.process_loss_qty, precision) != flt(process_loss_qty, precision):
self.process_loss_qty = flt(process_loss_qty, precision)
if data and data[0].process_loss_qty:
process_loss_qty = data[0].process_loss_qty
if flt(self.process_loss_qty, precision) != flt(process_loss_qty, precision):
self.process_loss_qty = flt(process_loss_qty, precision)
frappe.msgprint(
_("The Process Loss Qty has reset as per job cards Process Loss Qty"), alert=True
)
frappe.msgprint(_("The Process Loss Qty has reset as per job cards Process Loss Qty"), alert=True)
if not self.process_loss_percentage and not self.process_loss_qty:
self.process_loss_percentage = frappe.get_cached_value(
@@ -3222,6 +3213,62 @@ class StockEntry(StockController, SubcontractingInwardController):
(flt(self.process_loss_qty) / flt(self.fg_completed_qty)) * 100
)
def validate_job_card_pending_production(self):
"""A draft created before other entries were submitted must not book more than the job
card still has left; without this, a stale draft over-produces the finished good."""
if self.purpose != "Manufacture" or not self.job_card:
return
if self._action == "update_after_submit":
return
job_card = frappe.get_doc("Job Card", self.job_card)
if job_card.is_corrective_job_card or job_card.is_subcontracted:
return
precision = frappe.get_precision("Stock Entry Detail", "qty")
pending_qty = flt(
flt(job_card.for_quantity)
- flt(job_card.pending_qty)
- flt(job_card.manufactured_qty)
- flt(job_card.get_consumed_process_loss()),
precision,
)
finished_qty = flt(sum(flt(d.transfer_qty) for d in self.items if d.is_finished_item), precision)
entry_qty = flt(finished_qty + flt(self.process_loss_qty), precision)
if entry_qty > pending_qty:
item_code = job_card.finished_good or job_card.production_item
uom = frappe.get_cached_value("Item", item_code, "stock_uom")
frappe.throw(
_(
"The Job Card {0} has only {1} left to produce, but this entry books {2} ({3} finished goods and {4} process loss). Cancel or update its other manufacture entries first."
).format(
frappe.bold(self.job_card),
frappe.bold(f"{pending_qty} {uom}"),
frappe.bold(f"{entry_qty} {uom}"),
f"{finished_qty} {uom}",
f"{flt(self.process_loss_qty, precision)} {uom}",
)
)
def get_pending_process_loss_qty(self):
"""Loss this entry should still book: the job card's unbooked loss when the entry
belongs to one, else the largest operation loss on the work order (legacy flow)."""
if self.job_card:
job_card = frappe.get_doc("Job Card", self.job_card)
return max(flt(job_card.process_loss_qty) - flt(job_card.get_consumed_process_loss()), 0)
if self.work_order:
data = frappe.get_all(
"Work Order Operation",
filters={"parent": self.work_order},
fields=[{"MAX": "process_loss_qty", "as": "process_loss_qty"}],
)
return flt(data[0].process_loss_qty) if data else 0
return 0
def set_work_order_details(self):
if not getattr(self, "pro_doc", None):
self.pro_doc = frappe._dict()

28
erpnext/stock/doctype/stock_entry_type/stock_entry_type.py
View File

@@ -125,6 +125,8 @@ class ManufactureEntry:
if backflush_based_on != "BOM":
available_serial_batches = self.get_transferred_serial_batches()
production_share = self.get_production_share()
items_to_remove = []
for item_code, _dict in item_dict.items():
_dict.from_warehouse = self.source_wh.get(item_code) or self.wip_warehouse
_dict.to_warehouse = ""
@@ -138,11 +140,33 @@ class ManufactureEntry:
_dict.qty = calculated_qty
self.update_available_serial_batches(_dict, available_serial_batches)
elif self.skip_material_transfer:
set_previous_operation_serial_batch(self.stock_entry, _dict)
else:
remaining_qty = max(flt(_dict.qty) - flt(_dict.consumed_qty), 0)
_dict.qty = min(flt(_dict.qty) * production_share, remaining_qty)
if not _dict.qty:
items_to_remove.append(item_code)
continue
if self.skip_material_transfer:
set_previous_operation_serial_batch(self.stock_entry, _dict)
for item_code in items_to_remove:
item_dict.pop(item_code)
self.stock_entry.add_to_stock_entry_detail(item_dict)
def get_production_share(self):
"""Fraction of the job card's production this entry accounts for; raw materials are
generated proportionally so several partial entries never consume more than required."""
for_quantity, pending_qty = frappe.db.get_value(
"Job Card", self.job_card, ["for_quantity", "pending_qty"]
)
qty_to_produce = flt(for_quantity) - flt(pending_qty)
if not qty_to_produce:
return 1
return min(flt(self.for_quantity) / qty_to_produce, 1)
def parse_available_serial_batches(self, item_dict, available_serial_batches):
key = (item_dict.item_code, item_dict.from_warehouse)
if key not in available_serial_batches: