mirror of
https://github.com/frappe/erpnext.git
synced 2026-08-13 06:31:48 +00:00
fix(stock): convert get_used_alternative_items to query builder
get_used_alternative_items built its WHERE with f-string interpolation of subcontract_order / subcontract_order_field / work_order (a SQL-injection risk) and used a raw implicit comma cross-join. Convert to frappe.qb with an inner_join on sted.parent == ste.name and parameterised conditions. The raw SELECT listed sted.conversion_factor twice; the qb version selects it once. Engine-portable and MariaDB-identical. Surgical re-apply: the rest of stock_entry.py (the services/ package layout and other develop-only logic) is untouched. Adds a test that substitutes an alternative item in a work order's transfer entry and asserts get_used_alternative_items returns the mapping, on MariaDB and Postgres. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Mihir Kandoi
@@ -167,6 +167,106 @@ class TestItemAlternative(ERPNextTestSuite):
|
||||
self.assertEqual(status, True)
|
||||
ste1.submit()
|
||||
|
||||
def test_get_used_alternative_items_returns_substitution(self):
|
||||
# get_used_alternative_items (raw SQL -> frappe.qb) returns the alternative items substituted
|
||||
# into a work order's transfer entries, keyed by the original item. Exercises the converted
|
||||
# query on both engines.
|
||||
from erpnext.stock.doctype.stock_entry.stock_entry import get_used_alternative_items
|
||||
|
||||
create_stock_reconciliation(
|
||||
item_code="Alternate Item For A RW 1", warehouse="_Test Warehouse - _TC", qty=5, rate=2000
|
||||
)
|
||||
create_stock_reconciliation(
|
||||
item_code="Test FG A RW 2", warehouse="_Test Warehouse - _TC", qty=5, rate=2000
|
||||
)
|
||||
pro_order = make_wo_order_test_record(
|
||||
production_item="Test Finished Goods - A",
|
||||
qty=5,
|
||||
source_warehouse="_Test Warehouse - _TC",
|
||||
wip_warehouse="Test Supplier Warehouse - _TC",
|
||||
)
|
||||
|
||||
ste = frappe.get_doc(make_stock_entry(pro_order.name, "Material Transfer for Manufacture", 5))
|
||||
ste.insert()
|
||||
for item in ste.items:
|
||||
if item.item_code == "Test FG A RW 1":
|
||||
item.item_code = "Alternate Item For A RW 1"
|
||||
item.item_name = "Alternate Item For A RW 1"
|
||||
item.description = "Alternate Item For A RW 1"
|
||||
item.original_item = "Test FG A RW 1"
|
||||
ste.submit()
|
||||
|
||||
used = get_used_alternative_items(work_order=pro_order.name)
|
||||
self.assertIn("Test FG A RW 1", used)
|
||||
self.assertEqual(used["Test FG A RW 1"].item_code, "Alternate Item For A RW 1")
|
||||
|
||||
def test_get_used_alternative_items_for_subcontract_order(self):
|
||||
# Covers the subcontract_order branch of get_used_alternative_items (including the dynamic
|
||||
# subcontract_order_field column) on both engines.
|
||||
from erpnext.stock.doctype.stock_entry.stock_entry import get_used_alternative_items
|
||||
|
||||
set_backflush_based_on("BOM")
|
||||
create_stock_reconciliation(
|
||||
item_code="Alternate Item For A RW 1", warehouse="_Test Warehouse - _TC", qty=5, rate=2000
|
||||
)
|
||||
create_stock_reconciliation(
|
||||
item_code="Test FG A RW 2", warehouse="_Test Warehouse - _TC", qty=5, rate=2000
|
||||
)
|
||||
supplier_warehouse = "Test Supplier Warehouse - _TC"
|
||||
make_service_item("Subcontracted Service Item 1")
|
||||
service_items = [
|
||||
{
|
||||
"warehouse": "_Test Warehouse - _TC",
|
||||
"item_code": "Subcontracted Service Item 1",
|
||||
"qty": 5,
|
||||
"rate": 3000,
|
||||
"fg_item": "Test Finished Goods - A",
|
||||
"fg_item_qty": 5,
|
||||
},
|
||||
]
|
||||
sco = get_subcontracting_order(service_items=service_items, supplier_warehouse=supplier_warehouse)
|
||||
rm_items = [
|
||||
{
|
||||
"item_code": "Test Finished Goods - A",
|
||||
"rm_item_code": "Test FG A RW 1",
|
||||
"item_name": "Test FG A RW 1",
|
||||
"qty": 5,
|
||||
"warehouse": "_Test Warehouse - _TC",
|
||||
"rate": 2000,
|
||||
"amount": 10000,
|
||||
"stock_uom": "Nos",
|
||||
},
|
||||
{
|
||||
"item_code": "Test Finished Goods - A",
|
||||
"rm_item_code": "Test FG A RW 2",
|
||||
"item_name": "Test FG A RW 2",
|
||||
"qty": 5,
|
||||
"warehouse": "_Test Warehouse - _TC",
|
||||
"rate": 2000,
|
||||
"amount": 10000,
|
||||
"stock_uom": "Nos",
|
||||
},
|
||||
]
|
||||
|
||||
se = frappe.get_doc(make_rm_stock_entry(sco.name, rm_items))
|
||||
se.to_warehouse = supplier_warehouse
|
||||
se.insert()
|
||||
for item in se.items:
|
||||
if item.item_code == "Test FG A RW 1":
|
||||
item.item_code = "Alternate Item For A RW 1"
|
||||
item.item_name = "Alternate Item For A RW 1"
|
||||
item.description = "Alternate Item For A RW 1"
|
||||
item.original_item = "Test FG A RW 1"
|
||||
se.save()
|
||||
se.submit()
|
||||
|
||||
used = get_used_alternative_items(
|
||||
subcontract_order=sco.name, subcontract_order_field="subcontracting_order"
|
||||
)
|
||||
self.assertIn("Test FG A RW 1", used)
|
||||
self.assertEqual(used["Test FG A RW 1"].item_code, "Alternate Item For A RW 1")
|
||||
set_backflush_based_on("Material Transferred for Subcontract")
|
||||
|
||||
def test_get_alternative_items_both_directions_and_dedup(self):
|
||||
"""get_alternative_items must return forward alternatives, reverse-only
|
||||
two_way alternatives, exclude one-way reverse rows, and dedupe an item
|
||||
|
||||
@@ -1623,29 +1623,42 @@ def get_remaining_operating_cost(work_order=None, bom_no=None):
|
||||
def get_used_alternative_items(
|
||||
subcontract_order=None, subcontract_order_field="subcontracting_order", work_order=None
|
||||
):
|
||||
cond = ""
|
||||
ste = frappe.qb.DocType("Stock Entry")
|
||||
sted = frappe.qb.DocType("Stock Entry Detail")
|
||||
|
||||
query = (
|
||||
frappe.qb.from_(ste)
|
||||
.inner_join(sted)
|
||||
.on(sted.parent == ste.name)
|
||||
.select(
|
||||
sted.original_item,
|
||||
sted.uom,
|
||||
sted.conversion_factor,
|
||||
sted.item_code,
|
||||
sted.item_name,
|
||||
sted.stock_uom,
|
||||
sted.description,
|
||||
)
|
||||
.where((ste.docstatus == 1) & (sted.original_item != sted.item_code))
|
||||
)
|
||||
|
||||
if subcontract_order:
|
||||
cond = f"and ste.purpose = 'Send to Subcontractor' and ste.{subcontract_order_field} = '{subcontract_order}'"
|
||||
# subcontract_order_field is interpolated as a column identifier; restrict it to the two known
|
||||
# Stock Entry link fields so an unexpected value can't reference an arbitrary column.
|
||||
if subcontract_order_field not in ("subcontracting_order", "subcontracting_inward_order"):
|
||||
frappe.throw(_("Invalid subcontract order field: {0}").format(subcontract_order_field))
|
||||
query = query.where(
|
||||
(ste.purpose == "Send to Subcontractor") & (ste[subcontract_order_field] == subcontract_order)
|
||||
)
|
||||
elif work_order:
|
||||
cond = f"and ste.purpose = 'Material Transfer for Manufacture' and ste.work_order = '{work_order}'"
|
||||
|
||||
if not cond:
|
||||
query = query.where(
|
||||
(ste.purpose == "Material Transfer for Manufacture") & (ste.work_order == work_order)
|
||||
)
|
||||
else:
|
||||
return {}
|
||||
|
||||
used_alternative_items = {}
|
||||
data = frappe.db.sql(
|
||||
f""" select sted.original_item, sted.uom, sted.conversion_factor,
|
||||
sted.item_code, sted.item_name, sted.conversion_factor,sted.stock_uom, sted.description
|
||||
from
|
||||
`tabStock Entry` ste, `tabStock Entry Detail` sted
|
||||
where
|
||||
sted.parent = ste.name and ste.docstatus = 1 and sted.original_item != sted.item_code
|
||||
{cond} """,
|
||||
as_dict=1,
|
||||
)
|
||||
|
||||
for d in data:
|
||||
for d in query.run(as_dict=1):
|
||||
used_alternative_items[d.original_item] = d
|
||||
|
||||
return used_alternative_items
|
||||
|
||||
Reference in New Issue
Block a user