fix(stock): take disassembly source columns from one posted line (#57710)

* fix(stock): take disassembly source columns from one posted line

get_items_from_manufacture_stock_entry collapses a work order's Manufacture
entries to one row per item and wrapped fifteen Stock Entry Detail columns in
independent Max() to satisfy Postgres' strict GROUP BY. Those columns describe
a line, not an item, and three sets have to stay together:

  uom only means something beside its conversion_factor
  batch_no and serial_no only beside their warehouse
  is_finished_item decides whether the row is the output or an input

Aggregated separately they can be drawn from different lines. Two Manufacture
entries consuming the same item in Nos and in Box return ("Nos", 5) -- a pair
that was never posted, and one that does not describe the summed quantity.

Keep the sums (and the qty-weighted basic_rate) in the aggregate, and read the
descriptive columns off a single real line: the earliest by Stock Entry
creation then idx. That is what MariaDB returned in practice, it is
deterministic, and it is identical on both engines. Same representative-row
shape already used by BOM Stock Analysis and the sub-assembly queries.

* test(manufacturing): cover disassembly source-row coherence

Two Manufacture entries consume the same raw material in different UOMs, so
the max uom and the max conversion factor come from different lines. Asserts
the returned pair is one that was actually posted. Fails on the previous
per-column Max() with ('Nos', 5.0) not found in {('Nos', 1.0), ('Box', 5.0)}.

* fix(stock): aggregate disassembly quantities in stock UOM

erpnext/manufacturing/doctype/work_order/test_work_order.py

@@ -2889,6 +2889,99 @@ class TestWorkOrder(ERPNextTestSuite):
 			f"BOM-path disassembly must apply process_loss_per; expected 18, got {bom_scrap_row.qty}",
 		)
 
+	def test_disassembly_mixed_uom_rows_are_aggregated_in_stock_uom(self):
+		"""Quantities and rates from different row UOMs must be aggregated in stock UOM."""
+		from erpnext.stock.doctype.stock_entry.services.disassemble import DisassembleStockEntry
+		from erpnext.stock.doctype.stock_entry.test_stock_entry import (
+			make_stock_entry as make_stock_entry_test_record,
+		)
+
+		raw_item_doc = make_item(
+			"Test Raw for Disassembly Coherence", {"is_stock_item": 1, "stock_uom": "Nos"}
+		)
+		box_uom = next((row for row in raw_item_doc.uoms if row.uom == "Box"), None)
+		if box_uom:
+			box_uom.conversion_factor = 5
+		else:
+			raw_item_doc.append("uoms", {"uom": "Box", "conversion_factor": 5})
+		raw_item_doc.save()
+		raw_item = raw_item_doc.name
+		fg_item = make_item("Test FG for Disassembly Coherence", {"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=10, 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,
+		)
+
+		transfer = frappe.get_doc(make_stock_entry(wo.name, "Material Transfer for Manufacture", wo.qty))
+		for item in transfer.items:
+			item.s_warehouse = wo.wip_warehouse
+		transfer.save()
+		transfer.submit()
+
+		first = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 5))
+		first.submit()
+		second = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 5))
+		second.submit()
+		wo.reload()
+
+		first_row = next(row for row in first.items if row.item_code == raw_item)
+		second_row = next(row for row in second.items if row.item_code == raw_item)
+		first_stock_qty = flt(first_row.transfer_qty)
+		frappe.db.set_value(
+			"Stock Entry Detail",
+			first_row.name,
+			{
+				"uom": "Box",
+				"conversion_factor": 5,
+				"qty": first_stock_qty / 5,
+				"transfer_qty": first_stock_qty,
+				"basic_rate": 100,
+			},
+			update_modified=False,
+		)
+		frappe.db.set_value("Stock Entry Detail", second_row.name, "basic_rate", 200, update_modified=False)
+
+		posted_rows = frappe.get_all(
+			"Stock Entry Detail",
+			filters={"parent": ("in", [first.name, second.name]), "item_code": raw_item},
+			fields=["qty", "transfer_qty", "uom", "conversion_factor", "basic_rate"],
+		)
+		self.assertEqual(len({row.uom for row in posted_rows}), 2)
+		self.assertTrue(
+			all(flt(row.qty) * flt(row.conversion_factor) == flt(row.transfer_qty) for row in posted_rows)
+		)
+
+		service = DisassembleStockEntry(frappe._dict(work_order=wo.name, source_stock_entry=None))
+		source_row = next(
+			row for row in service.get_items_from_manufacture_stock_entry() if row.item_code == raw_item
+		)
+
+		expected_stock_qty = sum(flt(row.transfer_qty) for row in posted_rows)
+		expected_rate = (
+			sum(flt(row.basic_rate) * flt(row.transfer_qty) for row in posted_rows) / expected_stock_qty
+		)
+		self.assertEqual(source_row.uom, source_row.stock_uom)
+		self.assertEqual(flt(source_row.conversion_factor), 1.0)
+		self.assertEqual(flt(source_row.qty), expected_stock_qty)
+		self.assertEqual(flt(source_row.transfer_qty), expected_stock_qty)
+		self.assertAlmostEqual(flt(source_row.basic_rate), expected_rate, places=6)
+
+		disassemble_qty = 4
+		disassembly = frappe.get_doc(make_stock_entry(wo.name, "Disassemble", disassemble_qty))
+		disassembly.save()
+		disassembly_row = next(row for row in disassembly.items if row.item_code == raw_item)
+		expected_disassembly_qty = expected_stock_qty * disassemble_qty / flt(wo.produced_qty)
+		self.assertEqual(disassembly_row.uom, disassembly_row.stock_uom)
+		self.assertEqual(flt(disassembly_row.conversion_factor), 1.0)
+		self.assertEqual(flt(disassembly_row.transfer_qty), expected_disassembly_qty)
+		disassembly.submit()
+
 	def test_disassembly_with_additional_rm_not_in_bom(self):
 		"""
 		Test that SE-linked disassembly includes additional raw materials

erpnext/stock/doctype/stock_entry/services/disassemble.py

@@ -2,7 +2,7 @@ from collections import defaultdict
 
 import frappe
 from frappe import _
-from frappe.query_builder.functions import Max, Min, NullIf, Sum
+from frappe.query_builder.functions import Min, NullIf, Sum
 from frappe.utils import flt
 
 from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
@@ -348,35 +348,16 @@ class DisassembleStockEntry(BaseStockEntry):
 				.run(as_dict=True)
 			)
 
-		# Aggregating across all Manufacture entries of the work order, one row per item_code.
-		# The non-grouped columns are constant per item_code in practice (an item plays one role with
-		# one uom/warehouse across the WO's manufacture entries); Max() keeps the GROUP BY valid on
-		# postgres while returning the value MySQL picked arbitrarily, preserving the one-row-per-item
-		# shape the disassembly expects.
-		return (
+		# Aggregate in stock UOM: qty is expressed in each row's selected UOM and cannot be added
+		# when manufacture entries use different UOMs for the same item. basic_rate is also per
+		# stock UOM, so weight it by transfer_qty. Manufacture rows always carry positive stock
+		# qty, so NullIf only guards a theoretical /0.
+		rows = (
 			query.select(
-				Sum(SED.qty).as_("qty"),
-				Sum(SED.transfer_qty).as_("transfer_qty"),
 				SED.item_code,
-				Max(SED.item_name).as_("item_name"),
-				Max(SED.description).as_("description"),
-				Max(SED.stock_uom).as_("stock_uom"),
-				Max(SED.uom).as_("uom"),
-				# qty-weighted average so consolidating an item across manufacture entries at different
-				# valuation rates values the summed qty correctly (Max would bias the rate high).
-				# Manufacture rows always carry positive qty, so NullIf only guards a theoretical /0.
-				(Sum(SED.basic_rate * SED.qty) / NullIf(Sum(SED.qty), 0)).as_("basic_rate"),
-				Max(SED.conversion_factor).as_("conversion_factor"),
-				Max(SED.is_finished_item).as_("is_finished_item"),
-				Max(SED.secondary_item_type).as_("secondary_item_type"),
-				Max(SED.is_legacy_scrap_item).as_("is_legacy_scrap_item"),
-				Max(SED.bom_secondary_item).as_("bom_secondary_item"),
-				Max(SED.batch_no).as_("batch_no"),
-				Max(SED.serial_no).as_("serial_no"),
-				Max(SED.use_serial_batch_fields).as_("use_serial_batch_fields"),
-				Max(SED.s_warehouse).as_("s_warehouse"),
-				Max(SED.t_warehouse).as_("t_warehouse"),
-				Max(SED.bom_no).as_("bom_no"),
+				Sum(SED.transfer_qty).as_("qty"),
+				Sum(SED.transfer_qty).as_("transfer_qty"),
+				(Sum(SED.basic_rate * SED.transfer_qty) / NullIf(Sum(SED.transfer_qty), 0)).as_("basic_rate"),
 			)
 			.where(SE.purpose == "Manufacture")
 			.where(SE.work_order == self.doc.work_order)
@@ -385,6 +366,61 @@ class DisassembleStockEntry(BaseStockEntry):
 			.run(as_dict=True)
 		)
 
+		representative = self.get_representative_manufacture_rows()
+		for row in rows:
+			row.update(representative.get(row.item_code) or {})
+			row.uom = row.stock_uom
+			row.conversion_factor = 1
+
+		return rows
+
+	def get_representative_manufacture_rows(self):
+		"""Earliest posted line per item across the work order's Manufacture entries.
+
+		The disassembly wants one row per item, but some descriptive columns describe a line, not
+		an item: batch_no and serial_no only mean something beside their warehouse, and
+		is_finished_item decides whether the row is the output or an input. Aggregating each column
+		on its own can pair values from different lines into a row that was never posted, so take
+		the columns from a single real line instead. UOM is normalized separately to stock UOM.
+		"""
+		SE = frappe.qb.DocType("Stock Entry")
+		SED = frappe.qb.DocType("Stock Entry Detail")
+
+		lines = (
+			frappe.qb.from_(SED)
+			.join(SE)
+			.on(SED.parent == SE.name)
+			.select(
+				SED.item_code,
+				SED.item_name,
+				SED.description,
+				SED.stock_uom,
+				SED.is_finished_item,
+				SED.secondary_item_type,
+				SED.is_legacy_scrap_item,
+				SED.bom_secondary_item,
+				SED.batch_no,
+				SED.serial_no,
+				SED.use_serial_batch_fields,
+				SED.s_warehouse,
+				SED.t_warehouse,
+				SED.bom_no,
+			)
+			.where(
+				(SE.docstatus == 1) & (SE.purpose == "Manufacture") & (SE.work_order == self.doc.work_order)
+			)
+			.orderby(SE.creation)
+			.orderby(SE.name)
+			.orderby(SED.idx)
+			.run(as_dict=True)
+		)
+
+		representative = {}
+		for line in lines:
+			representative.setdefault(line.item_code, line)
+
+		return representative
+
 	def on_submit(self):
 		self.set_serial_batch_for_disassembly()
 		self.update_disassembled_order()