fix: keep source rate on re-fetch when maintain same rate is enabled (backport #57479) (#57792)

* fix: keep source rate on re-fetch when maintain same rate is enabled (#57479)

* fix: type-annotate get_item_details arguments

---------

Co-authored-by: test <test@test.com>

erpnext/stock/get_item_details.py

@@ -43,6 +43,28 @@ purchase_doctypes = [
 
 NOT_APPLICABLE_TAX = "N/A"
 
+# For each transaction, the child-row link field(s) that point to the source
+# document item, mapped to that source item doctype. When "maintain same rate" is
+# on, a mapped row keeps the persisted source pricing (read straight from that row),
+# so an unsaved edit on the target row can never lock in a non-source rate.
+maintain_same_rate_source_fields = {
+	"Purchase Order": {"supplier_quotation_item": "Supplier Quotation Item"},
+	"Purchase Receipt": {"purchase_order_item": "Purchase Order Item"},
+	"Purchase Invoice": {"po_detail": "Purchase Order Item", "pr_detail": "Purchase Receipt Item"},
+	"Sales Order": {"quotation_item": "Quotation Item"},
+	"Delivery Note": {"so_detail": "Sales Order Item", "si_detail": "Sales Invoice Item"},
+	"Sales Invoice": {"so_detail": "Sales Order Item", "dn_detail": "Delivery Note Item"},
+}
+
+LOCKED_RATE_FIELDS = [
+	"price_list_rate",
+	"rate",
+	"discount_percentage",
+	"discount_amount",
+	"margin_type",
+	"margin_rate_or_amount",
+]
+
 
 def _preprocess_ctx(ctx):
 	if not ctx.price_list:
@@ -58,7 +80,12 @@ def _preprocess_ctx(ctx):
 
 @frappe.whitelist()
 @erpnext.normalize_ctx_input(ItemDetailsCtx)
-def get_item_details(ctx, doc=None, for_validate=False, overwrite_warehouse=True) -> ItemDetails:
+def get_item_details(
+	ctx: ItemDetailsCtx,
+	doc: Document | str | None = None,
+	for_validate: bool | None = False,
+	overwrite_warehouse: bool = True,
+) -> ItemDetails:
 	"""
 	ctx = {
 	        "item_code": "",
@@ -120,16 +147,20 @@ def get_item_details(ctx, doc=None, for_validate=False, overwrite_warehouse=True
 	if ctx.doctype in ["Purchase Order", "Purchase Receipt", "Purchase Invoice"]:
 		ctx.customer = None
 
-	out.update(get_price_list_rate(ctx, item))
+	source_row = get_rate_locked_source_row(ctx, doc)
+	if source_row:
+		lock_source_rate(out, source_row)
+	else:
+		out.update(get_price_list_rate(ctx, item))
 
-	if (
+		if (
-		not out.price_list_rate
+			not out.price_list_rate
-		and ctx.transaction_type == "selling"
+			and ctx.transaction_type == "selling"
-		and frappe.get_single_value("Selling Settings", "fallback_to_default_price_list")
+			and frappe.get_single_value("Selling Settings", "fallback_to_default_price_list")
-	):
+		):
-		fallback_args = ctx.copy()
+			fallback_args = ctx.copy()
-		fallback_args.price_list = frappe.get_single_value("Selling Settings", "selling_price_list")
+			fallback_args.price_list = frappe.get_single_value("Selling Settings", "selling_price_list")
-		out.update(get_price_list_rate(fallback_args, item))
+			out.update(get_price_list_rate(fallback_args, item))
 
 	ctx.customer = current_customer
 
@@ -144,9 +175,8 @@ def get_item_details(ctx, doc=None, for_validate=False, overwrite_warehouse=True
 		if ctx.get(key) is None:
 			ctx[key] = value
 
-	data = get_pricing_rule_for_item(ctx, doc=doc, for_validate=for_validate)
+	if not source_row:
-
+		out.update(get_pricing_rule_for_item(ctx, doc=doc, for_validate=for_validate))
-	out.update(data)
 
 	if (
 		frappe.get_single_value("Stock Settings", "auto_create_serial_and_batch_bundle_for_outward")
@@ -176,6 +206,61 @@ def remove_standard_fields(out: ItemDetails):
 	return out
 
 
+def get_rate_locked_source_row(ctx: ItemDetailsCtx, doc) -> frappe._dict | None:
+	"""Return the persisted source-document row a mapped target row is locked to.
+
+	The rate is read from the linked source row in the database (not the mutable
+	target row), so a re-fetch always restores the source pricing the maintain-same-
+	rate validator checks against, even after an unsaved edit on the target row.
+	"""
+	if isinstance(doc, str):
+		doc = json.loads(doc)
+
+	source_fields = maintain_same_rate_source_fields.get(ctx.parenttype or ctx.doctype)
+	if not source_fields or not doc or ctx.get("is_return") or not maintain_same_rate_enabled(ctx):
+		return None
+
+	row = next((d for d in doc.get("items") or [] if d.get("name") == ctx.child_docname), None)
+	if not row:
+		return None
+
+	for link_field, source_doctype in source_fields.items():
+		if source_name := row.get(link_field):
+			# a direct read would bypass permissions; only return source pricing to a
+			# caller allowed to read the source document
+			source = frappe.db.get_value(
+				source_doctype, source_name, [*LOCKED_RATE_FIELDS, "parent", "parenttype"], as_dict=True
+			)
+			if source and frappe.has_permission(source.parenttype, doc=source.parent):
+				return source
+			return None
+	return None
+
+
+def maintain_same_rate_enabled(ctx: ItemDetailsCtx) -> bool:
+	if (ctx.parenttype or ctx.doctype) in purchase_doctypes:
+		if ctx.get("is_internal_supplier"):
+			return False
+		return bool(cint(frappe.get_cached_value("Buying Settings", "None", "maintain_same_rate")))
+
+	if ctx.get("is_internal_customer"):
+		return False
+	return bool(cint(frappe.get_cached_value("Selling Settings", "None", "maintain_same_sales_rate")))
+
+
+def lock_source_rate(out: frappe._dict, source_row) -> None:
+	"""Copy the source row's whole pricing block onto out so a mapped row keeps its
+	exact rate. Pricing rules are skipped for these rows, so nothing re-derives it and
+	the manual discount or margin that made rate differ from price_list_rate survives.
+	"""
+	out.price_list_rate = flt(source_row.get("price_list_rate")) or flt(source_row.get("rate"))
+	out.rate = flt(source_row.get("rate"))
+	out.discount_percentage = flt(source_row.get("discount_percentage"))
+	out.discount_amount = flt(source_row.get("discount_amount"))
+	out.margin_type = source_row.get("margin_type")
+	out.margin_rate_or_amount = flt(source_row.get("margin_rate_or_amount"))
+
+
 def set_valuation_rate(out: ItemDetails | dict, ctx: ItemDetailsCtx):
 	if frappe.db.exists("Product Bundle", {"name": ctx.item_code, "disabled": 0}, cache=True):
 		valuation_rate = 0.0
@@ -1614,14 +1699,21 @@ def apply_price_list(ctx, as_doc=False, doc=None):
 
 def apply_price_list_on_item(ctx, doc=None):
 	item_doc = frappe.get_cached_doc("Item", ctx.item_code)
-	item_details = get_price_list_rate(ctx, item_doc)
+
+	source_row = get_rate_locked_source_row(ctx, doc)
+	if source_row:
+		item_details = frappe._dict()
+		lock_source_rate(item_details, source_row)
+	else:
+		item_details = get_price_list_rate(ctx, item_doc)
 
 	ctx.conversion_factor = flt(ctx.conversion_factor) or get_conversion_factor(ctx.item_code, ctx.uom).get(
 		"conversion_factor", 1
 	)
 	ctx.stock_qty = flt(ctx.qty) * flt(ctx.conversion_factor)
 
-	item_details.update(get_pricing_rule_for_item(ctx, doc=doc))
+	if not source_row:
+		item_details.update(get_pricing_rule_for_item(ctx, doc=doc))
 
 	return item_details
 

erpnext/stock/tests/test_get_item_details.py

@@ -123,3 +123,336 @@ class TestGetItemDetail(ERPNextTestSuite):
 		dn.save()
 		self.assertEqual(dn.items[0].batch_no, "BATCH01")
 		self.assertEqual(dn.items[0].rate, 50)
+
+	def test_maintain_same_rate_keeps_source_rate_on_refetch(self):
+		"""#57436: with "maintain same rate" on, re-fetching a PR row mapped from a
+		PO must keep the PO rate instead of pulling a newer, higher Item Price.
+
+		The rate is validated on save, so it can never persist changed; assert the
+		fetched rate directly to prove the newer Item Price is never picked up.
+		"""
+		from erpnext.buying.doctype.purchase_order.purchase_order import make_purchase_receipt
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+		from erpnext.stock.doctype.item.test_item import make_item
+
+		def set_maintain_same_rate(value):
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", value)
+			frappe.clear_cache(doctype="Buying Settings")
+
+		set_maintain_same_rate(1)
+
+		item_code = make_item(properties={"is_stock_item": 1}).name
+		po = create_purchase_order(item_code=item_code, qty=1, rate=100)
+
+		# The PO may auto-insert an Item Price at 100; bump it to the newer, higher rate.
+		item_price = frappe.db.get_value(
+			"Item Price", {"item_code": item_code, "price_list": "Standard Buying"}
+		)
+		if item_price:
+			frappe.db.set_value("Item Price", item_price, "price_list_rate", 120)
+		else:
+			frappe.get_doc(
+				{
+					"doctype": "Item Price",
+					"price_list": "Standard Buying",
+					"item_code": item_code,
+					"price_list_rate": 120,
+				}
+			).insert()
+
+		pr = make_purchase_receipt(po.name)
+		pr.insert()
+
+		def fetch_price_list_rate():
+			ctx = frappe._dict(
+				{
+					"item_code": item_code,
+					"doctype": "Purchase Receipt",
+					"name": pr.name,
+					"company": pr.company,
+					"supplier": pr.supplier,
+					"currency": pr.currency,
+					"conversion_rate": 1.0,
+					"price_list": "Standard Buying",
+					"price_list_currency": pr.currency,
+					"plc_conversion_rate": 1.0,
+					"warehouse": pr.items[0].warehouse,
+					"uom": pr.items[0].uom,
+					"stock_uom": pr.items[0].stock_uom,
+					"qty": pr.items[0].qty,
+					"child_doctype": pr.items[0].doctype,
+					"child_docname": pr.items[0].name,
+					"is_return": 0,
+					"is_internal_supplier": 0,
+					"ignore_pricing_rule": 1,
+				}
+			)
+			return get_item_details(ctx, pr).get("price_list_rate")
+
+		# Rate stays at the PO rate; the newer Item Price (120) is not fetched.
+		self.assertEqual(fetch_price_list_rate(), 100)
+
+		# Control: without the setting the newer Item Price would be fetched.
+		set_maintain_same_rate(0)
+		self.assertEqual(fetch_price_list_rate(), 120)
+
+	def test_maintain_same_rate_survives_refetch_with_discount(self):
+		"""A mapped Purchase Receipt row that carries a source discount (rate != price
+		list rate) must keep its rate when the row is re-fetched, so maintain-same-rate
+		lets the document save. process_item_selection runs the same recompute the desk
+		mirrors, so it covers the "discount discarded on refresh" concern end to end.
+		"""
+		from frappe.utils import flt
+
+		from erpnext.buying.doctype.purchase_order.purchase_order import make_purchase_receipt
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		item, price_list = "_Test Item", "_Test Buying Price List"
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		original_action = frappe.db.get_single_value("Buying Settings", "maintain_same_rate_action")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate_action", "Stop")
+		frappe.clear_cache(doctype="Buying Settings")
+
+		try:
+			for label, adjustment in (
+				("percentage", {"discount_percentage": 10}),
+				("amount", {"discount_amount": 10}),
+			):
+				with self.subTest(discount=label):
+					# a controlled discounted PO: list rate 100, effective rate 90
+					frappe.flags.dont_fetch_price_list_rate = True
+					po = create_purchase_order(item_code=item, qty=1, do_not_save=True)
+					po.buying_price_list = price_list
+					po.items[0].price_list_rate = 100
+					po.items[0].update(adjustment)
+					po.items[0].rate = 90
+					po.insert()
+					po.submit()
+					frappe.flags.dont_fetch_price_list_rate = False
+
+					# a newer Item Price must not leak onto the mapped row on re-fetch
+					item_price = frappe.db.get_value(
+						"Item Price", {"item_code": item, "price_list": price_list}
+					)
+					if item_price:
+						frappe.db.set_value("Item Price", item_price, "price_list_rate", 250)
+
+					pr = make_purchase_receipt(po.name)
+					pr.insert()
+					pr.process_item_selection(item_idx=pr.items[0].idx)
+
+					self.assertEqual(flt(pr.items[0].rate), 90)
+					pr.save()  # must not raise the maintain-same-rate check
+		finally:
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate_action", original_action)
+			frappe.clear_cache(doctype="Buying Settings")
+			frappe.flags.dont_fetch_price_list_rate = False
+
+	def test_apply_price_list_keeps_source_rate_when_maintain_same_rate(self):
+		"""#57436: the bulk apply_price_list path (price list / party / conversion rate
+		change) must also keep the source rate on mapped rows, not just re-fetch of a
+		single row. Here a PR row carries its PO rate (175) while the current price list
+		rate is 100; the bulk apply must keep 175.
+		"""
+		from frappe.utils import flt, nowdate
+
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+		from erpnext.stock.get_item_details import apply_price_list
+
+		item_code = "_Test Item"
+		price_list = "_Test Buying Price List"
+
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.clear_cache(doctype="Buying Settings")
+
+		try:
+			po = create_purchase_order(item_code=item_code, rate=175, qty=1)
+
+			row_name = "pr-row-1"
+			pr_doc = {
+				"doctype": "Purchase Receipt",
+				"items": [
+					{
+						"name": row_name,
+						"item_code": item_code,
+						"purchase_order_item": po.items[0].name,
+						"price_list_rate": 175,
+						"rate": 175,
+					}
+				],
+			}
+			ctx = frappe._dict(
+				doctype="Purchase Receipt",
+				supplier=po.supplier,
+				company=po.company,
+				currency=po.currency,
+				conversion_rate=1.0,
+				price_list=price_list,
+				plc_conversion_rate=1.0,
+				transaction_date=nowdate(),
+				items=[
+					frappe._dict(
+						doctype="Purchase Receipt Item",
+						parenttype="Purchase Receipt",
+						item_code=item_code,
+						child_docname=row_name,
+						qty=1,
+						uom=po.items[0].uom,
+						stock_uom=po.items[0].stock_uom,
+						conversion_factor=1.0,
+					)
+				],
+			)
+
+			result = apply_price_list(ctx, doc=pr_doc)
+			self.assertEqual(flt(result["children"][0].get("price_list_rate")), 175)
+		finally:
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.clear_cache(doctype="Buying Settings")
+
+	def test_maintain_same_rate_keeps_source_discount_on_refetch(self):
+		"""A mapped source row with a discount has rate != price_list_rate. Re-fetch must
+		return the source's rate and discount, not just the pre-discount price, or the
+		recomputed rate diverges from the reference and fails maintain-same-rate on save.
+		"""
+		from frappe.utils import flt
+
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		item_code = "_Test Item"
+		price_list = "_Test Buying Price List"
+
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.clear_cache(doctype="Buying Settings")
+
+		try:
+			# source PO carries the discount: list rate 100, 10% off, effective rate 90
+			frappe.flags.dont_fetch_price_list_rate = True
+			po = create_purchase_order(item_code=item_code, qty=1, do_not_save=True)
+			po.buying_price_list = price_list
+			po.items[0].price_list_rate = 100
+			po.items[0].discount_percentage = 10
+			po.items[0].rate = 90
+			po.insert()
+			po.submit()
+			frappe.flags.dont_fetch_price_list_rate = False
+
+			row_name = "pr-row-1"
+			pr_doc = {
+				"doctype": "Purchase Receipt",
+				"items": [
+					{"name": row_name, "item_code": item_code, "purchase_order_item": po.items[0].name}
+				],
+			}
+			ctx = frappe._dict(
+				item_code=item_code,
+				doctype="Purchase Receipt",
+				company=po.company,
+				supplier=po.supplier,
+				currency=po.currency,
+				conversion_rate=1.0,
+				price_list=price_list,
+				price_list_currency=po.currency,
+				plc_conversion_rate=1.0,
+				warehouse="_Test Warehouse - _TC",
+				uom=po.items[0].uom,
+				stock_uom=po.items[0].stock_uom,
+				qty=1,
+				child_docname=row_name,
+				is_return=0,
+				is_internal_supplier=0,
+				ignore_pricing_rule=1,
+			)
+
+			out = get_item_details(ctx, pr_doc)
+			self.assertEqual(flt(out.get("price_list_rate")), 100)
+			self.assertEqual(flt(out.get("rate")), 90)
+			self.assertEqual(flt(out.get("discount_percentage")), 10)
+		finally:
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.clear_cache(doctype="Buying Settings")
+			frappe.flags.dont_fetch_price_list_rate = False
+
+	def test_refetch_restores_source_rate_after_target_edit(self):
+		"""Editing a mapped row's rate then re-fetching must restore the persisted source
+		rate (read from the linked row), not lock in the edit, so the document still saves.
+		"""
+		from frappe.utils import flt
+
+		from erpnext.buying.doctype.purchase_order.purchase_order import make_purchase_receipt
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		item = "_Test Item"
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		original_action = frappe.db.get_single_value("Buying Settings", "maintain_same_rate_action")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate_action", "Stop")
+		frappe.clear_cache(doctype="Buying Settings")
+
+		try:
+			po = create_purchase_order(item_code=item, qty=1, rate=90)
+			pr = make_purchase_receipt(po.name)
+			pr.insert()
+
+			# user edits the mapped row to a non-source rate
+			pr.items[0].price_list_rate = 200
+			pr.items[0].rate = 200
+
+			# a re-fetch must restore the persisted source (PO) rate, not keep the edit
+			pr.process_item_selection(item_idx=pr.items[0].idx)
+			self.assertEqual(flt(pr.items[0].rate), 90)
+			pr.save()  # must not raise the maintain-same-rate check
+		finally:
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate_action", original_action)
+			frappe.clear_cache(doctype="Buying Settings")
+
+	def test_rate_lock_source_lookup_checks_permission(self):
+		"""The lock reads source pricing via a direct DB read, so it must not disclose a
+		source document's pricing to a caller who cannot read that document.
+		"""
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+		from erpnext.stock.get_item_details import get_rate_locked_source_row
+
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.clear_cache(doctype="Buying Settings")
+
+		role, email = "_Test Role Without PO Access", "_test_rate_lock_probe@example.com"
+		try:
+			po = create_purchase_order(item_code="_Test Item", qty=1, rate=90)
+			pr_doc = {
+				"doctype": "Purchase Receipt",
+				"items": [{"name": "r1", "item_code": "_Test Item", "purchase_order_item": po.items[0].name}],
+			}
+			ctx = frappe._dict(doctype="Purchase Receipt", child_docname="r1")
+
+			# an authorized caller receives the source row
+			self.assertIsNotNone(get_rate_locked_source_row(ctx.copy(), dict(pr_doc)))
+
+			if not frappe.db.exists("Role", role):
+				frappe.get_doc({"doctype": "Role", "role_name": role, "desk_access": 1}).insert(
+					ignore_permissions=True
+				)
+			if not frappe.db.exists("User", email):
+				frappe.get_doc(
+					{
+						"doctype": "User",
+						"email": email,
+						"first_name": "Probe",
+						"send_welcome_email": 0,
+						"roles": [{"role": role}],
+					}
+				).insert(ignore_permissions=True)
+
+			frappe.set_user(email)
+			# a caller who cannot read the Purchase Order gets nothing
+			self.assertIsNone(get_rate_locked_source_row(ctx.copy(), dict(pr_doc)))
+		finally:
+			frappe.set_user("Administrator")
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.clear_cache(doctype="Buying Settings")

erpnext/utilities/transaction_base.py

@@ -548,7 +548,9 @@ class TransactionBase(StatusUpdater):
 		from erpnext.stock.get_item_details import apply_price_list
 
 		args = {
-			"items": [x.as_dict() for x in self.items],
+			# pass child_docname so the maintain-same-rate lock in apply_price_list can
+			# match each row, consistent with the desk (JS) callers
+			"items": [{**x.as_dict(), "child_docname": x.name} for x in self.items],
 			"customer": self.customer or self.party_name,
 			"quotation_to": self.quotation_to,
 			"customer_group": self.customer_group,