diff --git a/erpnext/stock/doctype/item_alternative/test_item_alternative.py b/erpnext/stock/doctype/item_alternative/test_item_alternative.py index c11ced53837..744a25b7445 100644 --- a/erpnext/stock/doctype/item_alternative/test_item_alternative.py +++ b/erpnext/stock/doctype/item_alternative/test_item_alternative.py @@ -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 diff --git a/erpnext/stock/doctype/stock_entry/stock_entry.py b/erpnext/stock/doctype/stock_entry/stock_entry.py index 146deb6ea9a..0ccd3a747b4 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry.py @@ -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