diff --git a/erpnext/stock/get_item_details.py b/erpnext/stock/get_item_details.py index 53401106f32..abbac9a7710 100644 --- a/erpnext/stock/get_item_details.py +++ b/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 ( - not out.price_list_rate - and ctx.transaction_type == "selling" - and frappe.get_single_value("Selling Settings", "fallback_to_default_price_list") - ): - fallback_args = ctx.copy() - fallback_args.price_list = frappe.get_single_value("Selling Settings", "selling_price_list") - out.update(get_price_list_rate(fallback_args, item)) + if ( + not out.price_list_rate + and ctx.transaction_type == "selling" + and frappe.get_single_value("Selling Settings", "fallback_to_default_price_list") + ): + fallback_args = ctx.copy() + fallback_args.price_list = frappe.get_single_value("Selling Settings", "selling_price_list") + 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) - - out.update(data) + if not source_row: + out.update(get_pricing_rule_for_item(ctx, doc=doc, for_validate=for_validate)) 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 diff --git a/erpnext/stock/tests/test_get_item_details.py b/erpnext/stock/tests/test_get_item_details.py index 99d94008221..d790c673e20 100644 --- a/erpnext/stock/tests/test_get_item_details.py +++ b/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") diff --git a/erpnext/utilities/transaction_base.py b/erpnext/utilities/transaction_base.py index 3ff42ed6a09..e6cad737a6b 100644 --- a/erpnext/utilities/transaction_base.py +++ b/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,