nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-08-11 05:31:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #57895 from mihir-kandoi/fix-track-sfg-finished-good

Browse Source 
fix: track semi finished goods production flow
This commit is contained in:
Mihir Kandoi
2026-08-09 20:11:05 +05:30
committed by GitHub
parent 0bb322f4d1 24de81f9fa
commit fe09f45ca7
11 changed files with 916 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -314,6 +314,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()
@@ -340,15 +341,42 @@ class BOM(WebsiteGenerator):
self.set_fg_cost_allocation()
self.validate_total_cost_allocation()
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:
@@ -800,15 +828,10 @@ class BOM(WebsiteGenerator):
row.update(get_item_details(row.get("item_code")))
row.operation_row_id = operation_row_id
item_row = self.get_item_data(row.name) if row.name else None
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
@@ -827,9 +850,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,
@@ -919,6 +919,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",

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

@@ -891,6 +891,9 @@ class JobCard(Document):
frappe.msgprint(message, alert=True, indicator="orange")
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
@@ -1111,6 +1114,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"
@@ -1461,12 +1467,12 @@ class JobCard(Document):
)
if self.track_semi_finished_goods and previous_operations:
manufactured_qty = self.get_manufactured_qty_per_operation(
[row.name for row in previous_operations]
)
totals = self.get_manufactured_qty_per_operation([row.name for row in previous_operations])
for row in previous_operations:
row.manufactured_qty = flt(manufactured_qty.get(row.name))
operation_totals = totals.get(row.name)
row.manufactured_qty = flt(operation_totals and operation_totals.manufactured_qty)
row.process_loss_qty = flt(operation_totals and operation_totals.process_loss_qty)
return previous_operations
@@ -1475,7 +1481,11 @@ class JobCard(Document):
data = (
frappe.qb.from_(job_card)
.select(job_card.operation_id, Sum(job_card.manufactured_qty))
.select(
job_card.operation_id,
Sum(job_card.manufactured_qty).as_("manufactured_qty"),
Sum(job_card.process_loss_qty).as_("process_loss_qty"),
)
.where(
(job_card.work_order == self.work_order)
& (job_card.docstatus == 1)
@@ -1483,9 +1493,9 @@ class JobCard(Document):
& (job_card.operation_id.isin(operation_ids))
)
.groupby(job_card.operation_id)
).run()
).run(as_dict=True)
return dict(data)
return {row.operation_id: row for row in data}
def get_current_operation_completed_qty(self):
current_operation_qty = 0.0
@@ -1537,19 +1547,35 @@ class JobCard(Document):
OperationSequenceError,
)
if manufactured_qty < current_operation_qty:
if manufactured_qty >= current_operation_qty:
return
if manufactured_qty + flt(row.process_loss_qty) >= current_operation_qty:
frappe.throw(
_(
"The completed quantity {0} of an operation {1} cannot be greater than the manufactured quantity {2} of a previous operation {3}. Submit the manufacturing entry for the operation {3} first."
"The completed quantity {0} of an operation {1} cannot be greater than the manufactured quantity {2} of a previous operation {3}, as {4} was booked as process loss there."
).format(
bold(self.get_qty_with_uom(current_operation_qty)),
bold(self.operation),
bold(self.get_qty_with_uom(manufactured_qty, row.finished_good)),
bold(row.operation),
bold(self.get_qty_with_uom(flt(row.process_loss_qty), row.finished_good)),
),
OperationSequenceError,
)
frappe.throw(
_(
"The completed quantity {0} of an operation {1} cannot be greater than the manufactured quantity {2} of a previous operation {3}. Submit the manufacturing entry for the operation {3} first."
).format(
bold(self.get_qty_with_uom(current_operation_qty)),
bold(self.operation),
bold(self.get_qty_with_uom(manufactured_qty, row.finished_good)),
bold(row.operation),
),
OperationSequenceError,
)
def validate_work_order(self):
if self.is_work_order_closed():
frappe.throw(_("You cannot make any changes to Job Card since Work Order is closed."))
@@ -1801,10 +1827,11 @@ class JobCard(Document):
def build_manufacture_stock_entry(self):
from erpnext.stock.doctype.stock_entry_type.stock_entry_type import ManufactureEntry
consumed_process_loss = self.get_consumed_process_loss()
return ManufactureEntry(
{
"for_quantity": self.get_qty_to_produce() - self.manufactured_qty,
"process_loss_qty": max(self.process_loss_qty - self.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,

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

@@ -1447,6 +1447,204 @@ class TestJobCard(ERPNextTestSuite):
self.assertEqual(flt(job_card.manufactured_qty), 3)
self.assertEqual(job_card.status, "Completed")
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 = self.get_first_job_card(work_order.name)
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)
# Operation A handed over only 8 units, so the final operation works on 8.
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_sequence_needs_previous_operations_manufactured(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
@@ -1726,6 +1924,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
@@ -2302,3 +2793,43 @@ class TestJobCardLogic(ERPNextTestSuite):
self.assertTrue(jc.has_overlap(1, sequential))
self.assertFalse(jc.has_overlap(2, sequential))
self.assertTrue(jc.has_overlap(2, overlapping))
def test_previous_operation_shortfall_from_process_loss_gets_the_right_message(self):
jc = frappe.new_doc("Job Card")
jc.operation = "_Test Painting"
jc.stock_uom = "Nos"
row = frappe._dict(
operation="_Test Assembly", manufactured_qty=8, process_loss_qty=2, finished_good=None
)
with self.assertRaises(OperationSequenceError) as loss_error:
jc.validate_previous_operation_manufactured_qty(row, 10)
self.assertIn("process loss", str(loss_error.exception))
row.process_loss_qty = 0
with self.assertRaises(OperationSequenceError) as pending_error:
jc.validate_previous_operation_manufactured_qty(row, 10)
self.assertIn("Submit the manufacturing entry", str(pending_error.exception))
jc.validate_previous_operation_manufactured_qty(row, 8)
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)

8
erpnext/manufacturing/doctype/work_order/services/status.py
View File

@@ -291,6 +291,12 @@ class StatusService:
)
def set_process_loss_qty(self):
self.doc.db_set("process_loss_qty", self._process_loss_qty())
def _process_loss_qty(self):
if self.doc.track_semi_finished_goods:
return flt(sum(flt(row.process_loss_qty) for row in self.doc.operations))
table = frappe.qb.DocType("Stock Entry")
process_loss_qty = (
frappe.qb.from_(table)
@@ -302,7 +308,7 @@ class StatusService:
)
).run()[0][0]
self.doc.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.doc.production_plan)

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

@@ -5461,6 +5461,24 @@ class TestWorkOrder(ERPNextTestSuite):
self.assertEqual(flt(qty_by_item.get(item_a)), 10.0)
self.assertEqual(flt(qty_by_item.get(item_b)), 10.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",

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

@@ -601,12 +601,9 @@ class WorkOrder(Document):
)
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):

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

@@ -319,6 +319,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.validate_job_card_item()
self.set_purpose_for_stock_entry()
@@ -1452,23 +1453,15 @@ 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)
frappe.msgprint(
_("The Process Loss Qty has been reset as per the job card's Process Loss Qty"),
alert=True,
)
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 been reset as per the job card's 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(
"BOM", self.bom_no, "process_loss_percentage"
@@ -1483,6 +1476,60 @@ 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.get_qty_to_produce())
- 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:
uom = job_card.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 self.work_order:
# common validations

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

@@ -126,6 +126,7 @@ class ManufactureEntry:
if backflush_based_on != "BOM":
available_serial_batches = self.get_transferred_serial_batches()
production_share = self.get_production_share()
for item_code, _dict in item_dict.items():
_dict.s_warehouse = self.source_wh.get(item_code) or self.wip_warehouse
_dict.t_warehouse = ""
@@ -140,11 +141,29 @@ 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:
continue
if self.skip_material_transfer:
set_previous_operation_serial_batch(self.stock_entry, _dict)
self.stock_entry.append("items", _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: