Merge pull request #56221 from mihir-kandoi/pg-stock-entry

fix(stock): convert get_used_alternative_items to query builder (Postgres)

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

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