From 27625a6f1fe5e0a2c7e70b223e64bfc54118a255 Mon Sep 17 00:00:00 2001 From: pandiyan Date: Sat, 8 Aug 2026 23:04:28 +0530 Subject: [PATCH] test: work order process loss for semi finished goods Cover both shapes: a single operation that books the loss itself, and a chain where an earlier operation books it and the final operation loses nothing, so the sum over the operations is the only correct source. (cherry picked from commit 24de81f9faca2abac250bf580b36e03ea226d11e) # Conflicts: # erpnext/manufacturing/doctype/job_card/test_job_card.py --- .../doctype/job_card/test_job_card.py | 420 ++++++++++++++++++ 1 file changed, 420 insertions(+) diff --git a/erpnext/manufacturing/doctype/job_card/test_job_card.py b/erpnext/manufacturing/doctype/job_card/test_job_card.py index a0ea71aaab0..c1f48953e5b 100644 --- a/erpnext/manufacturing/doctype/job_card/test_job_card.py +++ b/erpnext/manufacturing/doctype/job_card/test_job_card.py @@ -1265,6 +1265,426 @@ class TestJobCard(ERPNextTestSuite): 8, ) +<<<<<<< HEAD +======= + 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 + + warehouse = "Stores - _TC" + rm = make_item("Pending Qty RM 1", {"is_stock_item": 1}).name + fg = make_item("Pending Qty 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": "Pending Qty 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=5, + 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-04-01 08:00:00"}) + job_card.save() + + job_card.complete_job_card( + qty=3, + for_quantity=5, + pending_qty=2, + process_loss_qty=0, + end_time="2024-04-01 09:00:00", + ) + + job_card.reload() + self.assertEqual(flt(job_card.for_quantity), 5) + self.assertEqual(flt(job_card.pending_qty), 2) + self.assertEqual(flt(job_card.process_loss_qty), 0) + + job_card.submit() + self.assertEqual(job_card.status, "To Manufacture") + + manufacturing_entry = frappe.get_doc(job_card.make_stock_entry_for_semi_fg_item()) + finished_item = next(row for row in manufacturing_entry.items if row.is_finished_item) + self.assertEqual(flt(finished_item.qty), 3) + manufacturing_entry.submit() + + job_card.reload() + 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 + + warehouse = "Stores - _TC" + rm1 = make_item("Sequence Check RM 1", {"is_stock_item": 1}).name + rm2 = make_item("Sequence Check RM 2", {"is_stock_item": 1}).name + sfg1 = make_item("Sequence Check SFG 1", {"is_stock_item": 1}).name + sfg2 = make_item("Sequence Check SFG 2", {"is_stock_item": 1}).name + fg = make_item("Sequence Check FG 1", {"is_stock_item": 1}).name + + semi_fg_boms = {} + for semi_fg_item, raw_material in ((sfg1, rm1), (sfg2, rm2)): + bom = frappe.new_doc("BOM", company="_Test Company", item=semi_fg_item, quantity=1) + bom.append("items", {"item_code": raw_material, "qty": 1}) + bom.insert() + bom.submit() + semi_fg_boms[semi_fg_item] = bom.name + + fg_bom = frappe.new_doc( + "BOM", + company="_Test Company", + item=fg, + quantity=1, + with_operations=1, + track_semi_finished_goods=1, + ) + + operations = [ + { + "operation": "Sequence Check Op A", + "finished_good": sfg1, + "bom_no": semi_fg_boms[sfg1], + "sequence_id": 1, + }, + { + "operation": "Sequence Check Op B", + "finished_good": sfg2, + "bom_no": semi_fg_boms[sfg2], + "sequence_id": 1, + }, + { + "operation": "Sequence Check Op C", + "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": sfg1, "qty": 1, "operation_row_id": 3}) + fg_bom.append("items", {"item_code": sfg2, "qty": 1, "operation_row_id": 3}) + fg_bom.insert() + fg_bom.submit() + + work_order = make_wo_order_test_record( + item=fg, + qty=5, + 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=rm1, target=warehouse, qty=10, basic_rate=100) + make_stock_entry(item_code=rm2, target=warehouse, qty=10, 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", + ), + ) + + def add_time_log(job_card, day, qty): + job_card.append( + "time_logs", + { + "from_time": f"2024-01-{day} 08:00:00", + "to_time": f"2024-01-{day} 09:00:00", + "completed_qty": qty, + }, + ) + + jc_a = get_job_card("Sequence Check Op A") + jc_a.for_quantity = 3 + add_time_log(jc_a, "01", 3) + jc_a.submit() + + jc_b = get_job_card("Sequence Check Op B") + add_time_log(jc_b, "02", jc_b.for_quantity) + jc_b.submit() + frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item()).submit() + + jc_c = get_job_card("Sequence Check Op C") + jc_c.for_quantity = 3 + add_time_log(jc_c, "03", 3) + self.assertRaises(OperationSequenceError, jc_c.save) + + frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item()).submit() + + jc_c.reload() + jc_c.for_quantity = 4 + add_time_log(jc_c, "03", 4) + self.assertRaises(OperationSequenceError, jc_c.save) + + jc_c.reload() + jc_c.for_quantity = 3 + add_time_log(jc_c, "03", 3) + jc_c.submit() + + self.assertEqual(jc_c.docstatus, 1) + +>>>>>>> 24de81f9fa (test: work order process loss for semi finished goods) def test_semi_fg_batch_auto_pull_on_manufacture(self): from erpnext.manufacturing.doctype.operation.test_operation import make_operation from erpnext.stock.doctype.item.test_item import make_item