nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-08-12 06:01:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(subcontracting): dedup validate_manufacture consumption checks

Browse Source 
The `skip_transfer` and transfer branches of `validate_manufacture` ran the
same per-item validation loop — look the row up or throw "not a part of",
check overconsumption, guard against duplicates, record — differing only in
the data source (SCIO Received Item vs Work Order Item), the available-qty
basis, the source-warehouse check (skip_transfer only) and the message text.

Split each branch into a small method that builds a normalised
`{item_code: {consumed_qty, available_qty}}` lookup, and share the loop via
`_validate_customer_provided_consumption`. Branch-specific throw messages are
passed as callbacks so the user-facing strings (and their translations) are
unchanged, and the order in which checks fire is preserved. Also drops the
unused `name` column from the skip_transfer query.

Adds a test for the non-skip-transfer manufacture flow (Material Transfer for
Manufacture -> Manufacture), which exercises the Work Order branch that the
existing suite — all of whose manufacture tests set skip_transfer=1 — never
covered. Full subcontracting-inward suite passes on MariaDB and PostgreSQL.
This commit is contained in:
Mihir Kandoi
2026-06-22 08:09:09 +05:30
parent 23f1fc6235
commit b129daedc8
2 changed files with 142 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

227
erpnext/controllers/subcontracting_inward_controller.py
View File

@@ -248,125 +248,132 @@ class SubcontractingInwardController:
]
if frappe.get_cached_value("Work Order", self.work_order, "skip_transfer"):
customer_warehouse = frappe.get_cached_value(
"Subcontracting Inward Order", self.subcontracting_inward_order, "customer_warehouse"
self._validate_manufacture_consumption_against_scio(items)
else:
self._validate_manufacture_consumption_against_work_order(items)
def _validate_manufacture_consumption_against_scio(self, items):
customer_warehouse = frappe.get_cached_value(
"Subcontracting Inward Order", self.subcontracting_inward_order, "customer_warehouse"
)
table = frappe.qb.DocType("Subcontracting Inward Order Received Item")
query = (
frappe.qb.from_(table)
.select(
table.rm_item_code,
table.consumed_qty,
(table.received_qty - table.returned_qty).as_("available_qty"),
)
table = frappe.qb.DocType("Subcontracting Inward Order Received Item")
query = (
frappe.qb.from_(table)
.select(
table.rm_item_code,
(table.received_qty - table.returned_qty).as_("total_qty"),
table.consumed_qty,
table.name,
)
.where(
(table.docstatus == 1)
& (table.parent == self.subcontracting_inward_order)
& (
table.reference_name
== frappe.get_cached_value(
"Work Order", self.work_order, "subcontracting_inward_order_item"
)
.where(
(table.docstatus == 1)
& (table.parent == self.subcontracting_inward_order)
& (
table.reference_name
== frappe.get_cached_value(
"Work Order", self.work_order, "subcontracting_inward_order_item"
)
& (table.rm_item_code.isin([item.item_code for item in items]))
)
& (table.rm_item_code.isin([item.item_code for item in items]))
)
rm_item_dict = frappe._dict(
{
d.rm_item_code: frappe._dict(
{"name": d.name, "total_qty": d.total_qty, "qty": d.consumed_qty}
)
for d in query.run(as_dict=True)
}
)
lookup = {
d.rm_item_code: frappe._dict(consumed_qty=d.consumed_qty, available_qty=d.available_qty)
for d in query.run(as_dict=True)
}
def on_missing(item):
frappe.throw(
_(
"Row #{0}: Customer Provided Item {1} is not a part of Subcontracting Inward Order {2}"
).format(
item.idx,
get_link_to_form("Item", item.item_code),
get_link_to_form("Subcontracting Inward Order", self.subcontracting_inward_order),
)
)
item_codes = []
for item in items:
if rm := rm_item_dict.get(item.item_code):
if rm.qty + item.transfer_qty > rm.total_qty:
frappe.throw(
_(
"Row #{0}: Customer Provided Item {1} exceeds quantity available through Subcontracting Inward Order"
).format(item.idx, get_link_to_form("Item", item.item_code))
)
elif item.s_warehouse != customer_warehouse:
frappe.throw(
_(
"Row #{0}: For Customer Provided Item {1}, Source Warehouse must be {2}"
).format(
item.idx,
get_link_to_form("Item", item.item_code),
get_link_to_form("Warehouse", customer_warehouse),
)
)
elif item.item_code in item_codes:
frappe.throw(
_(
"Row #{0}: Customer Provided Item {1} cannot be added multiple times in the Subcontracting Inward process."
).format(
item.idx,
get_link_to_form("Item", item.item_code),
)
)
else:
item_codes.append(item.item_code)
else:
def on_overconsumption(item):
frappe.throw(
_(
"Row #{0}: Customer Provided Item {1} exceeds quantity available through Subcontracting Inward Order"
).format(item.idx, get_link_to_form("Item", item.item_code))
)
def check_source_warehouse(item):
if item.s_warehouse != customer_warehouse:
frappe.throw(
_("Row #{0}: For Customer Provided Item {1}, Source Warehouse must be {2}").format(
item.idx,
get_link_to_form("Item", item.item_code),
get_link_to_form("Warehouse", customer_warehouse),
)
)
self._validate_customer_provided_consumption(
items, lookup, on_missing, on_overconsumption, check_source_warehouse
)
def _validate_manufacture_consumption_against_work_order(self, items):
work_order_items = frappe.get_all(
"Work Order Item",
{"parent": self.work_order, "docstatus": 1, "is_customer_provided_item": 1},
["item_code", "transferred_qty", "consumed_qty"],
)
lookup = {
wo_item.item_code: frappe._dict(
consumed_qty=wo_item.consumed_qty, available_qty=wo_item.transferred_qty
)
for wo_item in work_order_items
}
def on_missing(item):
frappe.throw(
_("Row #{0}: Customer Provided Item {1} is not a part of Work Order {2}").format(
item.idx,
get_link_to_form("Item", item.item_code),
get_link_to_form("Work Order", self.work_order),
)
)
def on_overconsumption(item):
frappe.throw(
_(
"Row #{0}: Overconsumption of Customer Provided Item {1} against Work Order {2} is not allowed in the Subcontracting Inward process."
).format(
item.idx,
get_link_to_form("Item", item.item_code),
get_link_to_form("Work Order", self.work_order),
)
)
self._validate_customer_provided_consumption(items, lookup, on_missing, on_overconsumption)
def _validate_customer_provided_consumption(
self, items, lookup, on_missing, on_overconsumption, extra_check=None
):
"""Shared per-item guard for the skip-transfer and transfer manufacture paths.
`lookup` maps item_code -> {consumed_qty, available_qty}; the branch-specific
throw messages are supplied as callbacks. `extra_check` runs an extra per-item
validation (the source-warehouse check on the skip-transfer path).
"""
seen = []
for item in items:
record = lookup.get(item.item_code)
if not record:
on_missing(item)
elif record.consumed_qty + item.transfer_qty > record.available_qty:
on_overconsumption(item)
else:
if extra_check:
extra_check(item)
if item.item_code in seen:
frappe.throw(
_(
"Row #{0}: Customer Provided Item {1} is not a part of Subcontracting Inward Order {2}"
).format(
item.idx,
get_link_to_form("Item", item.item_code),
get_link_to_form("Subcontracting Inward Order", self.subcontracting_inward_order),
)
)
else:
work_order_items = frappe.get_all(
"Work Order Item",
{"parent": self.work_order, "docstatus": 1, "is_customer_provided_item": 1},
["item_code", "transferred_qty", "consumed_qty"],
)
wo_item_dict = frappe._dict(
{
wo_item.item_code: frappe._dict(
{"transferred_qty": wo_item.transferred_qty, "consumed_qty": wo_item.consumed_qty}
)
for wo_item in work_order_items
}
)
item_codes = []
for item in items:
if wo_item := wo_item_dict.get(item.item_code):
if wo_item.consumed_qty + item.transfer_qty > wo_item.transferred_qty:
frappe.throw(
_(
"Row #{0}: Overconsumption of Customer Provided Item {1} against Work Order {2} is not allowed in the Subcontracting Inward process."
).format(
item.idx,
get_link_to_form("Item", item.item_code),
get_link_to_form("Work Order", self.work_order),
)
)
elif item.item_code in item_codes:
frappe.throw(
_(
"Row #{0}: Customer Provided Item {1} cannot be added multiple times in the Subcontracting Inward process."
).format(
item.idx,
get_link_to_form("Item", item.item_code),
)
)
else:
item_codes.append(item.item_code)
else:
frappe.throw(
_("Row #{0}: Customer Provided Item {1} is not a part of Work Order {2}").format(
item.idx,
get_link_to_form("Item", item.item_code),
get_link_to_form("Work Order", self.work_order),
)
"Row #{0}: Customer Provided Item {1} cannot be added multiple times in the Subcontracting Inward process."
).format(item.idx, get_link_to_form("Item", item.item_code))
)
seen.append(item.item_code)
def set_allow_zero_valuation_rate(self):
if self.subcontracting_inward_order:

25
erpnext/subcontracting/doctype/subcontracting_inward_order/test_subcontracting_inward_order.py
View File

@@ -330,6 +330,31 @@ class IntegrationTestSubcontractingInwardOrder(ERPNextTestSuite):
self.assertEqual(scio.items[0].delivered_qty, 2)
self.assertEqual(scio.items[0].returned_qty, 1)
def test_manufacture_consumption_validates_against_work_order(self):
"""Cover the non-skip-transfer manufacture path, where consumption is validated
against the Work Order's transferred quantity (the Work Order branch of
validate_manufacture)."""
so, scio = create_so_scio()
frappe.new_doc("Stock Entry").update(scio.make_rm_stock_entry_inward()).submit()
scio.reload()
wo = frappe.get_doc("Work Order", scio.make_work_order()[0])
wo.wip_warehouse = "Work In Progress - _TC"
next(
item for item in wo.required_items if item.item_code == "Self RM"
).source_warehouse = "Stores - _TC"
wo.submit()
frappe.new_doc("Stock Entry").update(
make_stock_entry_from_wo(wo.name, "Material Transfer for Manufacture")
).submit()
manufacture = frappe.new_doc("Stock Entry").update(make_stock_entry_from_wo(wo.name, "Manufacture"))
manufacture.submit()
scio.reload()
self.assertEqual(scio.items[0].produced_qty, 5)
@ERPNextTestSuite.change_settings("Selling Settings", {"allow_delivery_of_overproduced_qty": 1})
@ERPNextTestSuite.change_settings(
"Manufacturing Settings", {"overproduction_percentage_for_work_order": 20}