Merge pull request #54097 from frappe/mergify/bp/version-15-hotfix/pr-53964

fix: consistently disassemble based on source  > SE / WO / BOM (backport #53964)
This commit is contained in:
Smit Vora
2026-04-07 19:27:23 +05:30
committed by GitHub
6 changed files with 837 additions and 131 deletions

View File

@@ -4,7 +4,7 @@
import frappe
from frappe.tests.utils import FrappeTestCase, change_settings, timeout
from frappe.utils import add_days, add_months, add_to_date, cint, flt, now, today
from frappe.utils import add_days, add_months, add_to_date, cint, flt, now, nowdate, nowtime, today
from erpnext.manufacturing.doctype.job_card.job_card import JobCardCancelError
from erpnext.manufacturing.doctype.job_card.job_card import make_stock_entry as make_stock_entry_from_jc
@@ -2395,7 +2395,7 @@ class TestWorkOrder(FrappeTestCase):
stock_entry.submit()
def test_disassembly_order_with_qty_behavior(self):
def test_disassembly_order_with_qty_from_wo_behavior(self):
# Create raw material and FG item
raw_item = make_item("Test Raw for Disassembly", {"is_stock_item": 1}).name
fg_item = make_item("Test FG for Disassembly", {"is_stock_item": 1}).name
@@ -2435,27 +2435,9 @@ class TestWorkOrder(FrappeTestCase):
se_for_manufacture = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", wo.qty))
se_for_manufacture.submit()
# Simulate a disassembly stock entry
# Disassembly via WO required_items path (no source_stock_entry)
disassemble_qty = 4
stock_entry = frappe.get_doc(make_stock_entry(wo.name, "Disassemble", disassemble_qty))
stock_entry.append(
"items",
{
"item_code": fg_item,
"qty": disassemble_qty,
"s_warehouse": wo.fg_warehouse,
},
)
for bom_item in bom.items:
stock_entry.append(
"items",
{
"item_code": bom_item.item_code,
"qty": (bom_item.qty / bom.quantity) * disassemble_qty,
"t_warehouse": wo.source_warehouse,
},
)
wo.reload()
stock_entry.save()
@@ -2470,7 +2452,7 @@ class TestWorkOrder(FrappeTestCase):
f"Expected FG qty {disassemble_qty}, found {finished_good_entry.qty}",
)
# Assert raw materials
# Assert raw materials - qty scaled from WO required_items
for item in stock_entry.items:
if item.item_code == fg_item:
continue
@@ -2494,10 +2476,35 @@ class TestWorkOrder(FrappeTestCase):
f"Work Order disassembled_qty mismatch: expected {disassemble_qty}, got {wo.disassembled_qty}",
)
# Second disassembly: explicitly linked to manufacture SE — verifies SE-linked path
# (first disassembly auto-set source_stock_entry since there's only one manufacture entry)
disassemble_qty_2 = 2
stock_entry_2 = frappe.get_doc(
make_stock_entry(
wo.name, "Disassemble", disassemble_qty_2, source_stock_entry=se_for_manufacture.name
)
)
stock_entry_2.save()
stock_entry_2.submit()
# All rows must trace back to se_for_manufacture
for item in stock_entry_2.items:
self.assertEqual(item.against_stock_entry, se_for_manufacture.name)
self.assertTrue(item.ste_detail)
# RM qty scaled from the manufacture SE rows
rm_row = next((i for i in stock_entry_2.items if i.item_code == raw_item), None)
expected_rm_qty = (bom.items[0].qty / bom.quantity) * disassemble_qty_2
self.assertAlmostEqual(rm_row.qty, expected_rm_qty, places=3)
wo.reload()
self.assertEqual(wo.disassembled_qty, disassemble_qty + disassemble_qty_2)
def test_disassembly_with_multiple_manufacture_entries(self):
"""
Test that disassembly does not create duplicate items when manufacturing
is done in multiple batches (multiple manufacture stock entries).
is done in multiple batches (multiple manufacture stock entries), including
secondary/scrap items.
Scenario:
1. Create Work Order for 10 units
@@ -2506,11 +2513,17 @@ class TestWorkOrder(FrappeTestCase):
4. Create Disassembly for 4 units
5. Verify no duplicate items in the disassembly stock entry
"""
# Create RM and FG item
# Create RM, scrap and FG item
raw_item1 = make_item("Test Raw for Multi Batch Disassembly 1", {"is_stock_item": 1}).name
raw_item2 = make_item("Test Raw for Multi Batch Disassembly 2", {"is_stock_item": 1}).name
scrap_item = make_item("Test Scrap for Multi Batch Disassembly", {"is_stock_item": 1}).name
fg_item = make_item("Test FG for Multi Batch Disassembly", {"is_stock_item": 1}).name
bom = make_bom(item=fg_item, quantity=1, raw_materials=[raw_item1, raw_item2], rm_qty=2)
bom = make_bom(
item=fg_item, quantity=1, raw_materials=[raw_item1, raw_item2], rm_qty=2, do_not_submit=True
)
# add scrap item
bom.append("scrap_items", {"item_code": scrap_item, "stock_qty": 10})
bom.submit()
# Create WO
wo = make_wo_order_test_record(production_item=fg_item, qty=10, bom_no=bom.name, status="Not Started")
@@ -2585,7 +2598,7 @@ class TestWorkOrder(FrappeTestCase):
f"Found duplicate items in disassembly stock entry: {duplicates}",
)
expected_items = 3 # FG item + 2 raw materials
expected_items = 4 # FG item + 2 raw materials + 1 scrap item
self.assertEqual(
len(stock_entry.items),
expected_items,
@@ -2596,6 +2609,16 @@ class TestWorkOrder(FrappeTestCase):
fg_item_row = next((i for i in stock_entry.items if i.item_code == fg_item), None)
self.assertEqual(fg_item_row.qty, disassemble_qty)
# Scrap item: should be taken from scrap warehouse in disassembly
scrap_row = next((i for i in stock_entry.items if i.item_code == scrap_item), None)
self.assertIsNotNone(scrap_row)
self.assertEqual(scrap_row.is_scrap_item, 1)
self.assertTrue(scrap_row.s_warehouse)
self.assertFalse(scrap_row.t_warehouse)
self.assertEqual(scrap_row.s_warehouse, wo.scrap_warehouse)
# BOM has scrap_qty=10/FG, total produced = 10*10 = 100, disassemble 4/10 → 40
self.assertEqual(scrap_row.qty, 40)
# RM quantities
for bom_item in bom.items:
expected_qty = (bom_item.qty / bom.quantity) * disassemble_qty
@@ -2607,19 +2630,57 @@ class TestWorkOrder(FrappeTestCase):
msg=f"Raw material {bom_item.item_code} qty mismatch",
)
# -- BOM-path disassembly (no source_stock_entry, no work_order) --
make_stock_entry_test_record(
item_code=scrap_item,
purpose="Material Receipt",
target=wo.fg_warehouse,
qty=50,
basic_rate=10,
)
bom_disassemble_qty = 2
bom_se = frappe.get_doc(
{
"doctype": "Stock Entry",
"stock_entry_type": "Disassemble",
"purpose": "Disassemble",
"from_bom": 1,
"bom_no": bom.name,
"fg_completed_qty": bom_disassemble_qty,
"from_warehouse": wo.fg_warehouse,
"to_warehouse": wo.wip_warehouse,
"company": wo.company,
"posting_date": nowdate(),
"posting_time": nowtime(),
}
)
bom_se.get_items()
bom_se.save()
bom_se.submit()
bom_scrap_row = next((i for i in bom_se.items if i.item_code == scrap_item), None)
self.assertIsNotNone(bom_scrap_row, "Scrap item must appear in BOM-path disassembly")
# v15: BOM scrap_qty=10/FG, no process_loss_per field → qty = 10 * 2 = 20
self.assertEqual(
bom_scrap_row.qty,
20,
f"BOM-path disassembly scrap qty mismatch; expected 20, got {bom_scrap_row.qty}",
)
def test_disassembly_with_additional_rm_not_in_bom(self):
"""
Test that disassembly correctly handles additional raw materials that were
manually added during manufacturing (not part of the BOM).
Test that SE-linked disassembly includes additional raw materials
that were manually added during manufacturing (not part of the BOM).
Scenario:
1. Create Work Order for 10 units with 2 raw materials in BOM
2. Transfer raw materials for manufacture
3. Manufacture in 2 parts (3 units, then 7 units)
4. In each manufacture entry, manually add an extra consumable item
(not in BOM) in proportion to the manufactured qty
5. Create Disassembly for 4 units
6. Verify that the additional RM is included in disassembly with proportional qty
5. Disassemble 3 units linked to first manufacture entry
6. Verify additional RM is included with correct proportional qty from SE1
"""
from erpnext.stock.doctype.stock_entry.test_stock_entry import (
make_stock_entry as make_stock_entry_test_record,
@@ -2655,9 +2716,8 @@ class TestWorkOrder(FrappeTestCase):
se_for_material_transfer.save()
se_for_material_transfer.submit()
# First Manufacture Entry - 3 units
# First Manufacture Entry - 3 units with additional RM
se_manufacture1 = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 3))
# Additional RM
se_manufacture1.append(
"items",
{
@@ -2670,9 +2730,8 @@ class TestWorkOrder(FrappeTestCase):
se_manufacture1.save()
se_manufacture1.submit()
# Second Manufacture Entry - 7 units
# Second Manufacture Entry - 7 units with additional RM
se_manufacture2 = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 7))
# AAdditional RM
se_manufacture2.append(
"items",
{
@@ -2688,13 +2747,15 @@ class TestWorkOrder(FrappeTestCase):
wo.reload()
self.assertEqual(wo.produced_qty, 10)
# Disassembly for 4 units
disassemble_qty = 4
stock_entry = frappe.get_doc(make_stock_entry(wo.name, "Disassemble", disassemble_qty))
# Disassemble 3 units linked to first manufacture entry
disassemble_qty = 3
stock_entry = frappe.get_doc(
make_stock_entry(wo.name, "Disassemble", disassemble_qty, source_stock_entry=se_manufacture1.name)
)
stock_entry.save()
stock_entry.submit()
# No duplicate
# No duplicates
item_counts = {}
for item in stock_entry.items:
item_code = item.item_code
@@ -2707,16 +2768,15 @@ class TestWorkOrder(FrappeTestCase):
f"Found duplicate items in disassembly stock entry: {duplicates}",
)
# Additional RM qty
# Additional RM should be included — qty proportional to SE1 (3 units -> 3 additional RM)
additional_rm_row = next((i for i in stock_entry.items if i.item_code == additional_rm), None)
self.assertIsNotNone(
additional_rm_row,
f"Additional raw material {additional_rm} not found in disassembly",
)
# intentional full reversal as not part of BOM
# eg: dies or consumables used during manufacturing
expected_additional_rm_qty = 3 + 7
# SE1 had 3 additional RM for 3 manufactured units, disassembling all 3
expected_additional_rm_qty = 3
self.assertAlmostEqual(
additional_rm_row.qty,
expected_additional_rm_qty,
@@ -2724,7 +2784,7 @@ class TestWorkOrder(FrappeTestCase):
msg=f"Additional RM qty mismatch: expected {expected_additional_rm_qty}, got {additional_rm_row.qty}",
)
# RM qty
# BOM RM qty — scaled from SE1's rows
for bom_item in bom.items:
expected_qty = (bom_item.qty / bom.quantity) * disassemble_qty
rm_row = next((i for i in stock_entry.items if i.item_code == bom_item.item_code), None)
@@ -2740,6 +2800,7 @@ class TestWorkOrder(FrappeTestCase):
fg_item_row = next((i for i in stock_entry.items if i.item_code == fg_item), None)
self.assertEqual(fg_item_row.qty, disassemble_qty)
# FG + 2 BOM RM + 1 additional RM = 4 items
expected_items = 4
self.assertEqual(
len(stock_entry.items),
@@ -2747,6 +2808,282 @@ class TestWorkOrder(FrappeTestCase):
f"Expected {expected_items} items, found {len(stock_entry.items)}",
)
# Verify traceability
for item in stock_entry.items:
self.assertEqual(item.against_stock_entry, se_manufacture1.name)
self.assertTrue(item.ste_detail)
def test_disassembly_auto_sets_source_stock_entry(self):
from erpnext.stock.doctype.stock_entry.test_stock_entry import (
make_stock_entry as make_stock_entry_test_record,
)
raw_item = make_item("Test Raw Auto Set Disassembly", {"is_stock_item": 1}).name
fg_item = make_item("Test FG Auto Set Disassembly", {"is_stock_item": 1}).name
bom = make_bom(item=fg_item, quantity=1, raw_materials=[raw_item], rm_qty=2)
wo = make_wo_order_test_record(production_item=fg_item, qty=5, bom_no=bom.name, status="Not Started")
make_stock_entry_test_record(
item_code=raw_item, purpose="Material Receipt", target=wo.wip_warehouse, qty=50, basic_rate=100
)
se_transfer = frappe.get_doc(make_stock_entry(wo.name, "Material Transfer for Manufacture", wo.qty))
for item in se_transfer.items:
item.s_warehouse = wo.wip_warehouse
se_transfer.save()
se_transfer.submit()
se_manufacture = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", wo.qty))
se_manufacture.submit()
# Disassemble without specifying source_stock_entry
stock_entry = frappe.get_doc(make_stock_entry(wo.name, "Disassemble", 3))
stock_entry.save()
# source_stock_entry should be auto-set since only one manufacture entry
self.assertEqual(stock_entry.source_stock_entry, se_manufacture.name)
# All items should have against_stock_entry linked
for item in stock_entry.items:
self.assertEqual(item.against_stock_entry, se_manufacture.name)
self.assertTrue(item.ste_detail)
stock_entry.submit()
def test_disassembly_batch_tracked_items(self):
from erpnext.stock.doctype.batch.batch import make_batch
from erpnext.stock.doctype.stock_entry.test_stock_entry import (
make_stock_entry as make_stock_entry_test_record,
)
wip_wh = "_Test Warehouse - _TC"
rm_item = make_item(
"Test Batch RM for Disassembly SB",
{
"is_stock_item": 1,
"has_batch_no": 1,
"create_new_batch": 1,
"batch_number_series": "TBRD-RM-.###",
},
).name
fg_item = make_item(
"Test Batch FG for Disassembly SB",
{
"is_stock_item": 1,
"has_batch_no": 1,
"create_new_batch": 1,
"batch_number_series": "TBRD-FG-.###",
},
).name
bom = make_bom(item=fg_item, quantity=1, raw_materials=[rm_item], rm_qty=2)
wo = make_wo_order_test_record(
production_item=fg_item,
qty=6,
bom_no=bom.name,
skip_transfer=1,
source_warehouse=wip_wh,
status="Not Started",
)
# Two separate RM receipts → two distinct batches (batch_1, batch_2)
rm_receipt_1 = make_stock_entry_test_record(
item_code=rm_item, purpose="Material Receipt", target=wip_wh, qty=6, basic_rate=100
)
rm_batch_1 = get_batch_from_bundle(
frappe.db.get_value(
"Stock Entry Detail",
{"parent": rm_receipt_1.name, "item_code": rm_item},
"serial_and_batch_bundle",
)
)
rm_receipt_2 = make_stock_entry_test_record(
item_code=rm_item, purpose="Material Receipt", target=wip_wh, qty=6, basic_rate=100
)
rm_batch_2 = get_batch_from_bundle(
frappe.db.get_value(
"Stock Entry Detail",
{"parent": rm_receipt_2.name, "item_code": rm_item},
"serial_and_batch_bundle",
)
)
self.assertNotEqual(rm_batch_1, rm_batch_2, "Two receipts must create two distinct RM batches")
fg_batch_1 = make_batch(frappe._dict(item=fg_item))
fg_batch_2 = make_batch(frappe._dict(item=fg_item))
# Manufacture entry 1 — 3 FG using batch_1 RM/FG
se_manufacture_1 = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 3))
for row in se_manufacture_1.items:
if row.item_code == rm_item:
row.batch_no = rm_batch_1
row.use_serial_batch_fields = 1
elif row.item_code == fg_item:
row.batch_no = fg_batch_1
row.use_serial_batch_fields = 1
se_manufacture_1.save()
se_manufacture_1.submit()
# Manufacture entry 2 — 3 FG using batch_2 RM/FG
se_manufacture_2 = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 3))
for row in se_manufacture_2.items:
if row.item_code == rm_item:
row.batch_no = rm_batch_2
row.use_serial_batch_fields = 1
elif row.item_code == fg_item:
row.batch_no = fg_batch_2
row.use_serial_batch_fields = 1
se_manufacture_2.save()
se_manufacture_2.submit()
# Disassemble 2 units from SE_1 only — must use SE_1's batches, not SE_2's
disassemble_qty = 2
stock_entry = frappe.get_doc(
make_stock_entry(
wo.name, "Disassemble", disassemble_qty, source_stock_entry=se_manufacture_1.name
)
)
stock_entry.save()
stock_entry.submit()
# FG row: must use fg_batch_1 exclusively (fg_batch_2 must not appear)
fg_row = next((i for i in stock_entry.items if i.item_code == fg_item), None)
self.assertIsNotNone(fg_row)
self.assertTrue(fg_row.serial_and_batch_bundle, "FG row must have a serial_and_batch_bundle")
self.assertEqual(get_batch_from_bundle(fg_row.serial_and_batch_bundle), fg_batch_1)
self.assertNotEqual(get_batch_from_bundle(fg_row.serial_and_batch_bundle), fg_batch_2)
# RM row: must use rm_batch_1 exclusively (rm_batch_2 must not appear)
rm_row = next((i for i in stock_entry.items if i.item_code == rm_item), None)
self.assertIsNotNone(rm_row)
self.assertTrue(rm_row.serial_and_batch_bundle, "RM row must have a serial_and_batch_bundle")
self.assertEqual(get_batch_from_bundle(rm_row.serial_and_batch_bundle), rm_batch_1)
self.assertNotEqual(get_batch_from_bundle(rm_row.serial_and_batch_bundle), rm_batch_2)
# RM qty: 2 FG disassembled x 2 RM per FG = 4
self.assertAlmostEqual(rm_row.qty, 4.0, places=3)
def test_disassembly_serial_tracked_items(self):
from frappe.model.naming import make_autoname
from erpnext.stock.doctype.stock_entry.test_stock_entry import (
make_stock_entry as make_stock_entry_test_record,
)
wip_wh = "_Test Warehouse - _TC"
rm_item = make_item(
"Test Serial RM for Disassembly SB",
{"is_stock_item": 1, "has_serial_no": 1, "serial_no_series": "TSRD-RM-.####"},
).name
fg_item = make_item(
"Test Serial FG for Disassembly SB",
{"is_stock_item": 1, "has_serial_no": 1, "serial_no_series": "TSRD-FG-.####"},
).name
bom = make_bom(item=fg_item, quantity=1, raw_materials=[rm_item], rm_qty=2)
wo = make_wo_order_test_record(
production_item=fg_item,
qty=6,
bom_no=bom.name,
skip_transfer=1,
source_warehouse=wip_wh,
status="Not Started",
)
# Two separate RM receipts → two disjoint sets of serial numbers
rm_receipt_1 = make_stock_entry_test_record(
item_code=rm_item, purpose="Material Receipt", target=wip_wh, qty=6, basic_rate=100
)
rm_serials_1 = get_serial_nos_from_bundle(
frappe.db.get_value(
"Stock Entry Detail",
{"parent": rm_receipt_1.name, "item_code": rm_item},
"serial_and_batch_bundle",
)
)
self.assertEqual(len(rm_serials_1), 6)
rm_receipt_2 = make_stock_entry_test_record(
item_code=rm_item, purpose="Material Receipt", target=wip_wh, qty=6, basic_rate=100
)
rm_serials_2 = get_serial_nos_from_bundle(
frappe.db.get_value(
"Stock Entry Detail",
{"parent": rm_receipt_2.name, "item_code": rm_item},
"serial_and_batch_bundle",
)
)
self.assertEqual(len(rm_serials_2), 6)
self.assertFalse(
set(rm_serials_1) & set(rm_serials_2), "Two receipts must produce disjoint RM serial sets"
)
# Pre-generate two sets of FG serial numbers
series = frappe.db.get_value("Item", fg_item, "serial_no_series")
fg_serials_1 = [make_autoname(series) for _ in range(3)]
fg_serials_2 = [make_autoname(series) for _ in range(3)]
# Manufacture entry 1 — consumes rm_serials_1, produces fg_serials_1
se_manufacture_1 = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 3))
for row in se_manufacture_1.items:
if row.item_code == rm_item:
row.serial_no = "\n".join(rm_serials_1)
row.use_serial_batch_fields = 1
elif row.item_code == fg_item:
row.serial_no = "\n".join(fg_serials_1)
row.use_serial_batch_fields = 1
se_manufacture_1.save()
se_manufacture_1.submit()
# Manufacture entry 2 — consumes rm_serials_2, produces fg_serials_2
se_manufacture_2 = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 3))
for row in se_manufacture_2.items:
if row.item_code == rm_item:
row.serial_no = "\n".join(rm_serials_2)
row.use_serial_batch_fields = 1
elif row.item_code == fg_item:
row.serial_no = "\n".join(fg_serials_2)
row.use_serial_batch_fields = 1
se_manufacture_2.save()
se_manufacture_2.submit()
# Disassemble 2 units from SE_1 only — must use SE_1's serials, not SE_2's
disassemble_qty = 2
stock_entry = frappe.get_doc(
make_stock_entry(
wo.name, "Disassemble", disassemble_qty, source_stock_entry=se_manufacture_1.name
)
)
stock_entry.save()
stock_entry.submit()
# FG row: 2 serials consumed — must be subset of fg_serials_1, disjoint from fg_serials_2
fg_row = next((i for i in stock_entry.items if i.item_code == fg_item), None)
self.assertIsNotNone(fg_row)
self.assertTrue(fg_row.serial_and_batch_bundle, "FG row must have a serial_and_batch_bundle")
fg_dasm_serials = get_serial_nos_from_bundle(fg_row.serial_and_batch_bundle)
self.assertEqual(len(fg_dasm_serials), disassemble_qty)
self.assertTrue(set(fg_dasm_serials).issubset(set(fg_serials_1)))
self.assertFalse(
set(fg_dasm_serials) & set(fg_serials_2), "Disassembly must not use SE_2's FG serials"
)
# RM row: 4 serials returned (2 FG x 2 RM each) — subset of rm_serials_1, disjoint from rm_serials_2
rm_row = next((i for i in stock_entry.items if i.item_code == rm_item), None)
self.assertIsNotNone(rm_row)
self.assertTrue(rm_row.serial_and_batch_bundle, "RM row must have a serial_and_batch_bundle")
rm_dasm_serials = get_serial_nos_from_bundle(rm_row.serial_and_batch_bundle)
self.assertEqual(len(rm_dasm_serials), disassemble_qty * 2)
self.assertTrue(set(rm_dasm_serials).issubset(set(rm_serials_1)))
self.assertFalse(
set(rm_dasm_serials) & set(rm_serials_2), "Disassembly must not use SE_2's RM serials"
)
def test_components_alternate_item_for_bom_based_manufacture_entry(self):
frappe.db.set_single_value("Manufacturing Settings", "backflush_raw_materials_based_on", "BOM")
frappe.db.set_single_value("Manufacturing Settings", "validate_components_quantities_per_bom", 1)

View File

@@ -415,7 +415,7 @@ frappe.ui.form.on("Work Order", {
make_disassembly_order(frm) {
erpnext.work_order
.show_prompt_for_qty_input(frm, "Disassemble")
.show_disassembly_prompt(frm)
.then((data) => {
if (flt(data.qty) <= 0) {
frappe.msgprint(__("Disassemble Qty cannot be less than or equal to <b>0</b>."));
@@ -425,11 +425,14 @@ frappe.ui.form.on("Work Order", {
work_order_id: frm.doc.name,
purpose: "Disassemble",
qty: data.qty,
source_stock_entry: data.source_stock_entry,
});
})
.then((stock_entry) => {
frappe.model.sync(stock_entry);
frappe.set_route("Form", stock_entry.doctype, stock_entry.name);
if (stock_entry) {
frappe.model.sync(stock_entry);
frappe.set_route("Form", stock_entry.doctype, stock_entry.name);
}
});
},
@@ -879,6 +882,60 @@ erpnext.work_order = {
return flt(max, precision("qty"));
},
show_disassembly_prompt: function (frm) {
let max_qty = flt(frm.doc.produced_qty - frm.doc.disassembled_qty);
let fields = [
{
fieldtype: "Link",
label: __("Source Manufacture Entry"),
fieldname: "source_stock_entry",
options: "Stock Entry",
description: __("Optional. Select a specific manufacture entry to reverse."),
get_query: () => {
return {
filters: {
work_order: frm.doc.name,
purpose: "Manufacture",
docstatus: 1,
},
};
},
onchange: async function () {
if (!frm.disassembly_prompt) return;
let se_name = this.value;
let qty = max_qty;
if (se_name) {
qty = await frappe.xcall(
"erpnext.manufacturing.doctype.work_order.work_order.get_disassembly_available_qty",
{ stock_entry_name: se_name }
);
}
frm.disassembly_prompt.set_value("qty", qty);
frm.disassembly_prompt.fields_dict.qty.set_description(__("Max: {0}", [qty]));
},
},
{
fieldtype: "Float",
label: __("Qty for {0}", [__("Disassemble")]),
fieldname: "qty",
description: __("Max: {0}", [max_qty]),
default: max_qty,
},
];
return new Promise((resolve, reject) => {
frm.disassembly_prompt = frappe.prompt(
fields,
(data) => resolve(data),
__("Disassemble"),
__("Create")
);
});
},
show_prompt_for_qty_input: function (frm, purpose) {
let max = this.get_max_transferable_qty(frm, purpose);

View File

@@ -1485,7 +1485,13 @@ def set_work_order_ops(name):
@frappe.whitelist()
def make_stock_entry(work_order_id, purpose, qty=None, target_warehouse=None):
def make_stock_entry(
work_order_id: str,
purpose: str,
qty: float | None = None,
target_warehouse: str | None = None,
source_stock_entry: str | None = None,
):
work_order = frappe.get_doc("Work Order", work_order_id)
if not frappe.db.get_value("Warehouse", work_order.wip_warehouse, "is_group"):
wip_warehouse = work_order.wip_warehouse
@@ -1522,6 +1528,8 @@ def make_stock_entry(work_order_id, purpose, qty=None, target_warehouse=None):
if purpose == "Disassemble":
stock_entry.from_warehouse = work_order.fg_warehouse
stock_entry.to_warehouse = target_warehouse or work_order.source_warehouse
if source_stock_entry:
stock_entry.source_stock_entry = source_stock_entry
stock_entry.set_stock_entry_type()
stock_entry.get_items()
@@ -1532,6 +1540,28 @@ def make_stock_entry(work_order_id, purpose, qty=None, target_warehouse=None):
return stock_entry.as_dict()
@frappe.whitelist()
def get_disassembly_available_qty(stock_entry_name: str, current_se_name: str | None = None) -> float:
se = frappe.db.get_value("Stock Entry", stock_entry_name, ["fg_completed_qty"], as_dict=True)
if not se:
return 0.0
filters = {
"source_stock_entry": stock_entry_name,
"purpose": "Disassemble",
"docstatus": 1,
}
if current_se_name:
filters["name"] = ("!=", current_se_name)
already_disassembled = flt(
frappe.db.get_value("Stock Entry", filters, "sum(fg_completed_qty)", order_by=None)
)
return flt(se.fg_completed_qty) - already_disassembled
@frappe.whitelist()
def get_default_warehouse():
doc = frappe.get_cached_doc("Manufacturing Settings")

View File

@@ -36,6 +36,16 @@ frappe.ui.form.on("Stock Entry", {
};
});
frm.set_query("source_stock_entry", function () {
return {
filters: {
purpose: "Manufacture",
docstatus: 1,
work_order: frm.doc.work_order || undefined,
},
};
});
frm.set_query("source_warehouse_address", function () {
return {
query: "erpnext.controllers.queries.get_warehouse_address",
@@ -219,6 +229,30 @@ frappe.ui.form.on("Stock Entry", {
});
},
source_stock_entry: async function (frm) {
if (!frm.doc.source_stock_entry || frm.doc.purpose !== "Disassemble") return;
if (frm._via_source_stock_entry) {
frm.call({
doc: frm.doc,
method: "get_items",
callback: function (r) {
if (!r.exc) refresh_field("items");
},
});
frm._via_source_stock_entry = false;
return;
}
let available_qty = await frappe.xcall(
"erpnext.manufacturing.doctype.work_order.work_order.get_disassembly_available_qty",
{ stock_entry_name: frm.doc.source_stock_entry }
);
// triggers get_items() via its onchange
await frm.set_value("fg_completed_qty", available_qty);
},
outgoing_stock_entry: function (frm) {
frappe.call({
doc: frm.doc,
@@ -315,6 +349,59 @@ frappe.ui.form.on("Stock Entry", {
__("View")
);
}
if (frm.doc.purpose === "Manufacture") {
frm.add_custom_button(
__("Disassemble"),
async function () {
let available_qty = await frappe.xcall(
"erpnext.manufacturing.doctype.work_order.work_order.get_disassembly_available_qty",
{ stock_entry_name: frm.doc.name }
);
frappe.prompt(
{
fieldtype: "Float",
label: __("Qty to Disassemble"),
fieldname: "qty",
default: available_qty,
description: __("Max: {0}", [available_qty]),
},
async (data) => {
if (frm.doc.work_order) {
let stock_entry = await frappe.xcall(
"erpnext.manufacturing.doctype.work_order.work_order.make_stock_entry",
{
work_order_id: frm.doc.work_order,
purpose: "Disassemble",
qty: data.qty,
source_stock_entry: frm.doc.name,
}
);
if (stock_entry) {
frappe.model.sync(stock_entry);
frappe.set_route("Form", stock_entry.doctype, stock_entry.name);
}
} else {
let se = frappe.model.get_new_doc("Stock Entry");
se.company = frm.doc.company;
se.stock_entry_type = "Disassemble";
se.purpose = "Disassemble";
se.source_stock_entry = frm.doc.name;
se.from_bom = frm.doc.from_bom;
se.bom_no = frm.doc.bom_no;
se.fg_completed_qty = data.qty;
frm._via_source_stock_entry = true;
frappe.set_route("Form", "Stock Entry", se.name);
}
},
__("Disassemble"),
__("Create")
);
},
__("Create")
);
}
}
if (frm.doc.docstatus === 0) {
@@ -1279,8 +1366,11 @@ erpnext.stock.StockEntry = class StockEntry extends erpnext.stock.StockControlle
if (!this.frm.doc.fg_completed_qty || !this.frm.doc.bom_no)
frappe.throw(__("BOM and Manufacturing Quantity are required"));
if (this.frm.doc.work_order || this.frm.doc.bom_no) {
// if work order / bom is mentioned, get items
if (
this.frm.doc.work_order ||
this.frm.doc.bom_no ||
(this.frm.doc.purpose === "Disassemble" && this.frm.doc.source_stock_entry)
) {
return this.frm.call({
doc: me.frm.doc,
freeze: true,

View File

@@ -11,6 +11,7 @@
"naming_series",
"stock_entry_type",
"outgoing_stock_entry",
"source_stock_entry",
"purpose",
"add_to_transit",
"work_order",
@@ -120,6 +121,15 @@
"options": "Stock Entry",
"read_only": 1
},
{
"depends_on": "eval:doc.purpose == 'Disassemble'",
"fieldname": "source_stock_entry",
"fieldtype": "Link",
"label": "Source Stock Entry (Manufacture)",
"no_copy": 1,
"options": "Stock Entry",
"print_hide": 1
},
{
"bold": 1,
"fetch_from": "stock_entry_type.purpose",

View File

@@ -28,7 +28,6 @@ from erpnext.buying.utils import check_on_hold_or_closed_status
from erpnext.controllers.taxes_and_totals import init_landed_taxes_and_totals
from erpnext.manufacturing.doctype.bom.bom import (
add_additional_cost,
get_bom_items_as_dict,
get_op_cost_from_sub_assemblies,
get_scrap_items_from_sub_assemblies,
validate_bom_no,
@@ -143,6 +142,7 @@ class StockEntry(StockController):
select_print_heading: DF.Link | None
set_posting_time: DF.Check
source_address_display: DF.SmallText | None
source_stock_entry: DF.Link | None
source_warehouse_address: DF.Link | None
stock_entry_type: DF.Link
subcontracting_order: DF.Link | None
@@ -217,6 +217,7 @@ class StockEntry(StockController):
self.validate_warehouse()
self.validate_warehouse_of_sabb()
self.validate_work_order()
self.validate_source_stock_entry()
self.validate_bom()
self.set_process_loss_qty()
self.validate_purchase_order()
@@ -293,6 +294,56 @@ class StockEntry(StockController):
if self.purpose != "Disassemble":
return
if self.get("source_stock_entry"):
self._set_serial_batch_for_disassembly_from_stock_entry()
else:
self._set_serial_batch_for_disassembly_from_available_materials()
def _set_serial_batch_for_disassembly_from_stock_entry(self):
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
get_voucher_wise_serial_batch_from_bundle,
)
source_fg_qty = flt(frappe.db.get_value("Stock Entry", self.source_stock_entry, "fg_completed_qty"))
scale_factor = flt(self.fg_completed_qty) / source_fg_qty if source_fg_qty else 0
bundle_data = get_voucher_wise_serial_batch_from_bundle(voucher_no=[self.source_stock_entry])
source_rows_by_name = {r.name: r for r in self.get_items_from_manufacture_stock_entry()}
for row in self.items:
if not row.ste_detail:
continue
source_row = source_rows_by_name.get(row.ste_detail)
if not source_row:
continue
source_warehouse = source_row.s_warehouse or source_row.t_warehouse
key = (source_row.item_code, source_warehouse, self.source_stock_entry)
source_bundle = bundle_data.get(key, {})
batches = defaultdict(float)
serial_nos = []
if source_bundle.get("batch_nos"):
qty_remaining = row.transfer_qty
for batch_no, batch_qty in source_bundle["batch_nos"].items():
if qty_remaining <= 0:
break
alloc = min(abs(flt(batch_qty)) * scale_factor, qty_remaining)
batches[batch_no] = alloc
qty_remaining -= alloc
elif source_row.batch_no:
batches[source_row.batch_no] = row.transfer_qty
if source_bundle.get("serial_nos"):
serial_nos = get_serial_nos(source_bundle["serial_nos"])[: int(row.transfer_qty)]
elif source_row.serial_no:
serial_nos = get_serial_nos(source_row.serial_no)[: int(row.transfer_qty)]
self._set_serial_batch_bundle_for_disassembly_row(row, serial_nos, batches)
def _set_serial_batch_for_disassembly_from_available_materials(self):
available_materials = get_available_materials(self.work_order, self)
for row in self.items:
warehouse = row.s_warehouse or row.t_warehouse
@@ -318,34 +369,38 @@ class StockEntry(StockController):
if materials.serial_nos:
serial_nos = materials.serial_nos[: int(row.transfer_qty)]
if not serial_nos and not batches:
continue
self._set_serial_batch_bundle_for_disassembly_row(row, serial_nos, batches)
bundle_doc = SerialBatchCreation(
{
"item_code": row.item_code,
"warehouse": warehouse,
"posting_date": self.posting_date,
"posting_time": self.posting_time,
"voucher_type": self.doctype,
"voucher_no": self.name,
"voucher_detail_no": row.name,
"qty": row.transfer_qty,
"type_of_transaction": "Inward" if row.t_warehouse else "Outward",
"company": self.company,
"do_not_submit": True,
}
).make_serial_and_batch_bundle(serial_nos=serial_nos, batch_nos=batches)
def _set_serial_batch_bundle_for_disassembly_row(self, row, serial_nos, batches):
if not serial_nos and not batches:
return
row.serial_and_batch_bundle = bundle_doc.name
row.use_serial_batch_fields = 0
warehouse = row.s_warehouse or row.t_warehouse
bundle_doc = SerialBatchCreation(
{
"item_code": row.item_code,
"warehouse": warehouse,
"posting_date": self.posting_date,
"posting_time": self.posting_time,
"voucher_type": self.doctype,
"voucher_no": self.name,
"voucher_detail_no": row.name,
"qty": row.transfer_qty,
"type_of_transaction": "Inward" if row.t_warehouse else "Outward",
"company": self.company,
"do_not_submit": True,
}
).make_serial_and_batch_bundle(serial_nos=serial_nos, batch_nos=batches)
row.db_set(
{
"serial_and_batch_bundle": bundle_doc.name,
"use_serial_batch_fields": 0,
}
)
row.serial_and_batch_bundle = bundle_doc.name
row.use_serial_batch_fields = 0
row.db_set(
{
"serial_and_batch_bundle": bundle_doc.name,
"use_serial_batch_fields": 0,
}
)
def on_submit(self):
self.set_serial_batch_for_disassembly()
@@ -729,7 +784,7 @@ class StockEntry(StockController):
if self.purpose == "Disassemble":
if has_bom:
if d.is_finished_item:
if d.is_finished_item or d.is_scrap_item:
d.t_warehouse = None
if not d.s_warehouse:
frappe.throw(_("Source warehouse is mandatory for row {0}").format(d.idx))
@@ -766,6 +821,36 @@ class StockEntry(StockController):
elif self.purpose != "Material Transfer":
self.work_order = None
def validate_source_stock_entry(self):
if not self.get("source_stock_entry"):
return
if self.work_order:
source_wo = frappe.db.get_value("Stock Entry", self.source_stock_entry, "work_order")
if source_wo and source_wo != self.work_order:
frappe.throw(
_(
"Source Stock Entry {0} belongs to Work Order {1}, not {2}. Please use a manufacture entry from the same Work Order."
).format(self.source_stock_entry, source_wo, self.work_order),
title=_("Work Order Mismatch"),
)
from erpnext.manufacturing.doctype.work_order.work_order import get_disassembly_available_qty
available_qty = get_disassembly_available_qty(self.source_stock_entry, self.name)
if flt(self.fg_completed_qty) > available_qty:
frappe.throw(
_(
"Cannot disassemble {0} qty against Stock Entry {1}. Only {2} qty available to disassemble."
).format(
self.fg_completed_qty,
self.source_stock_entry,
available_qty,
),
title=_("Excess Disassembly"),
)
def check_if_operations_completed(self):
"""Check if Time Sheets are completed against before manufacturing to capture operating costs."""
prod_order = frappe.get_doc("Work Order", self.work_order)
@@ -1958,44 +2043,114 @@ class StockEntry(StockController):
)
def get_items_for_disassembly(self):
"""Get items for Disassembly Order"""
"""Get items for Disassembly Order.
Priority:
1. From a specific Manufacture Stock Entry (exact reversal)
2. From Work Order Manufacture Stock Entries (averaged reversal)
3. From BOM (standalone disassembly)
"""
# Auto-set source_stock_entry if WO has exactly one manufacture entry
if not self.get("source_stock_entry") and self.work_order:
manufacture_entries = frappe.get_all(
"Stock Entry",
filters={
"work_order": self.work_order,
"purpose": "Manufacture",
"docstatus": 1,
},
pluck="name",
limit_page_length=2,
)
if len(manufacture_entries) == 1:
self.source_stock_entry = manufacture_entries[0]
if self.get("source_stock_entry"):
return self._add_items_for_disassembly_from_stock_entry()
if self.work_order:
return self._add_items_for_disassembly_from_work_order()
return self._add_items_for_disassembly_from_bom()
def _add_items_for_disassembly_from_work_order(self):
items = self.get_items_from_manufacture_entry()
def _add_items_for_disassembly_from_stock_entry(self):
source_fg_qty = frappe.db.get_value("Stock Entry", self.source_stock_entry, "fg_completed_qty")
if not source_fg_qty:
frappe.throw(
_("Source Stock Entry {0} has no finished goods quantity").format(self.source_stock_entry)
)
s_warehouse = frappe.db.get_value("Work Order", self.work_order, "fg_warehouse")
disassemble_qty = flt(self.fg_completed_qty)
scale_factor = disassemble_qty / flt(source_fg_qty)
items_dict = get_bom_items_as_dict(
self.bom_no,
self.company,
self.fg_completed_qty,
fetch_exploded=self.use_multi_level_bom,
fetch_qty_in_stock_uom=False,
self._append_disassembly_row_from_source(
disassemble_qty=disassemble_qty,
scale_factor=scale_factor,
)
for row in items:
child_row = self.append("items", {})
for field, value in row.items():
if value is not None:
child_row.set(field, value)
def _add_items_for_disassembly_from_work_order(self):
wo_produced_qty = frappe.db.get_value("Work Order", self.work_order, "produced_qty")
# update qty and amount from BOM items
bom_items = items_dict.get(row.item_code)
if bom_items:
child_row.qty = bom_items.get("qty", child_row.qty)
child_row.amount = bom_items.get("amount", child_row.amount)
wo_produced_qty = flt(wo_produced_qty)
if wo_produced_qty <= 0:
frappe.throw(_("Work Order {0} has no produced qty").format(self.work_order))
if row.is_finished_item:
child_row.qty = self.fg_completed_qty
disassemble_qty = flt(self.fg_completed_qty)
if disassemble_qty <= 0:
frappe.throw(_("Disassemble Qty cannot be less than or equal to 0."))
child_row.s_warehouse = (self.from_warehouse or s_warehouse) if row.is_finished_item else ""
child_row.t_warehouse = row.s_warehouse
child_row.is_finished_item = 0 if row.is_finished_item else 1
scale_factor = disassemble_qty / wo_produced_qty
self._append_disassembly_row_from_source(
disassemble_qty=disassemble_qty,
scale_factor=scale_factor,
)
def _append_disassembly_row_from_source(self, disassemble_qty, scale_factor):
for source_row in self.get_items_from_manufacture_stock_entry():
if source_row.is_finished_item:
qty = disassemble_qty
s_warehouse = self.from_warehouse or source_row.t_warehouse
t_warehouse = ""
elif source_row.s_warehouse:
# RM: was consumed FROM s_warehouse -> return TO s_warehouse
qty = flt(source_row.qty * scale_factor)
s_warehouse = ""
t_warehouse = self.to_warehouse or source_row.s_warehouse
else:
# Scrap/secondary: was produced TO t_warehouse -> take FROM t_warehouse
qty = flt(source_row.qty * scale_factor)
s_warehouse = source_row.t_warehouse
t_warehouse = ""
item = {
"item_code": source_row.item_code,
"item_name": source_row.item_name,
"description": source_row.description,
"stock_uom": source_row.stock_uom,
"uom": source_row.uom,
"conversion_factor": source_row.conversion_factor,
"basic_rate": source_row.basic_rate,
"qty": qty,
"s_warehouse": s_warehouse,
"t_warehouse": t_warehouse,
"is_finished_item": source_row.is_finished_item,
"is_scrap_item": source_row.is_scrap_item,
"bom_no": source_row.bom_no,
# batch and serial bundles built on submit
"use_serial_batch_fields": 1 if (source_row.batch_no or source_row.serial_no) else 0,
}
if self.source_stock_entry:
item.update(
{
"against_stock_entry": self.source_stock_entry,
"ste_detail": source_row.name,
}
)
self.append("items", item)
def _add_items_for_disassembly_from_bom(self):
if not self.bom_no or not self.fg_completed_qty:
@@ -2011,37 +2166,64 @@ class StockEntry(StockController):
self.add_to_stock_entry_detail(item_dict)
# Scrap items (reverse: take scrap FROM scrap warehouse instead of producing TO it)
scrap_items = self.get_bom_scrap_material(self.fg_completed_qty)
if scrap_items:
scrap_warehouse = self.from_warehouse
if self.work_order:
wo_values = frappe.db.get_value(
"Work Order", self.work_order, ["scrap_warehouse", "fg_warehouse"], as_dict=True
)
scrap_warehouse = wo_values.scrap_warehouse or scrap_warehouse or wo_values.fg_warehouse
for item in scrap_items.values():
item["from_warehouse"] = scrap_warehouse
item["to_warehouse"] = ""
item["is_finished_item"] = 0
self.add_to_stock_entry_detail(scrap_items, bom_no=self.bom_no)
# Finished goods
self.load_items_from_bom()
def get_items_from_manufacture_entry(self):
return frappe.get_all(
"Stock Entry",
fields=[
"`tabStock Entry Detail`.`item_code`",
"`tabStock Entry Detail`.`item_name`",
"`tabStock Entry Detail`.`description`",
"sum(`tabStock Entry Detail`.qty) as qty",
"sum(`tabStock Entry Detail`.transfer_qty) as transfer_qty",
"`tabStock Entry Detail`.`stock_uom`",
"`tabStock Entry Detail`.`uom`",
"`tabStock Entry Detail`.`basic_rate`",
"`tabStock Entry Detail`.`conversion_factor`",
"`tabStock Entry Detail`.`is_finished_item`",
"`tabStock Entry Detail`.`batch_no`",
"`tabStock Entry Detail`.`serial_no`",
"`tabStock Entry Detail`.`s_warehouse`",
"`tabStock Entry Detail`.`t_warehouse`",
"`tabStock Entry Detail`.`use_serial_batch_fields`",
],
filters=[
["Stock Entry", "purpose", "=", "Manufacture"],
["Stock Entry", "work_order", "=", self.work_order],
["Stock Entry", "docstatus", "=", 1],
["Stock Entry Detail", "docstatus", "=", 1],
],
order_by="`tabStock Entry Detail`.`idx` desc, `tabStock Entry Detail`.`is_finished_item` desc",
group_by="`tabStock Entry Detail`.`item_code`",
def get_items_from_manufacture_stock_entry(self):
SE = frappe.qb.DocType("Stock Entry")
SED = frappe.qb.DocType("Stock Entry Detail")
query = frappe.qb.from_(SED).join(SE).on(SED.parent == SE.name).where(SE.docstatus == 1)
common_fields = [
SED.item_code,
SED.item_name,
SED.description,
SED.stock_uom,
SED.uom,
SED.basic_rate,
SED.conversion_factor,
SED.is_finished_item,
SED.is_scrap_item,
SED.batch_no,
SED.serial_no,
SED.use_serial_batch_fields,
SED.s_warehouse,
SED.t_warehouse,
SED.bom_no,
]
if self.source_stock_entry:
return (
query.select(SED.name, SED.qty, SED.transfer_qty, *common_fields)
.where(SE.name == self.source_stock_entry)
.orderby(SED.idx)
.run(as_dict=True)
)
return (
query.select(Sum(SED.qty).as_("qty"), Sum(SED.transfer_qty).as_("transfer_qty"), *common_fields)
.where(SE.purpose == "Manufacture")
.where(SE.work_order == self.work_order)
.groupby(SED.item_code)
.orderby(SED.idx)
.run(as_dict=True)
)
@frappe.whitelist()
@@ -2403,7 +2585,7 @@ class StockEntry(StockController):
return item_dict
def get_scrap_items_from_job_card(self):
if not self.pro_doc:
if not getattr(self, "pro_doc", None):
self.set_work_order_details()
if not self.pro_doc.operations: