diff --git a/erpnext/stock/get_item_details.py b/erpnext/stock/get_item_details.py index c9fa5fe6e16..6c1d658481a 100644 --- a/erpnext/stock/get_item_details.py +++ b/erpnext/stock/get_item_details.py @@ -7,6 +7,7 @@ import json import frappe from frappe import _, throw from frappe.model import child_table_fields, default_fields +from frappe.model.document import Document from frappe.model.meta import get_field_precision from frappe.model.utils import get_fetch_values from frappe.query_builder.functions import IfNull, Sum @@ -34,11 +35,34 @@ purchase_doctypes = [ "Purchase Invoice", ] +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", +] + NOT_APPLICABLE_TAX = "N/A" @frappe.whitelist() -def get_item_details(args, doc=None, for_validate=False, overwrite_warehouse=True): +def get_item_details( + args: dict | str, + doc: Document | dict | str | None = None, + for_validate: bool | str = False, + overwrite_warehouse: bool | str = True, +): """ args = { "item_code": "", @@ -100,16 +124,20 @@ def get_item_details(args, doc=None, for_validate=False, overwrite_warehouse=Tru if args.get("doctype") in ["Purchase Order", "Purchase Receipt", "Purchase Invoice"]: args.customer = None - out.update(get_price_list_rate(args, item)) + source_row = get_rate_locked_source_row(args, doc) + if source_row: + lock_source_rate(out, source_row) + else: + out.update(get_price_list_rate(args, item)) - if ( - not out.price_list_rate - and args.transaction_type == "selling" - and frappe.get_single_value("Selling Settings", "fallback_to_default_price_list") - ): - fallback_args = args.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 args.transaction_type == "selling" + and frappe.get_single_value("Selling Settings", "fallback_to_default_price_list") + ): + fallback_args = args.copy() + fallback_args.price_list = frappe.get_single_value("Selling Settings", "selling_price_list") + out.update(get_price_list_rate(fallback_args, item)) args.customer = current_customer @@ -124,9 +152,8 @@ def get_item_details(args, doc=None, for_validate=False, overwrite_warehouse=Tru if args.get(key) is None: args[key] = value - data = get_pricing_rule_for_item(args, doc=doc, for_validate=for_validate) - - out.update(data) + if not source_row: + out.update(get_pricing_rule_for_item(args, doc=doc, for_validate=for_validate)) if ( frappe.db.get_single_value("Stock Settings", "auto_create_serial_and_batch_bundle_for_outward") @@ -156,6 +183,52 @@ def remove_standard_fields(details): return details +def get_rate_locked_source_row(args, doc): + """Reads the source row from the DB, not the mutable target row, so an unsaved edit can't override the locked rate.""" + if isinstance(doc, str): + doc = json.loads(doc) + + source_fields = maintain_same_rate_source_fields.get(args.parenttype or args.doctype) + if not source_fields or not doc or args.get("is_return") or not maintain_same_rate_enabled(args): + return None + + row = next((d for d in doc.get("items") or [] if d.get("name") == args.child_docname), None) + if not row: + return None + + for link_field, source_doctype in source_fields.items(): + if source_name := row.get(link_field): + # don't leak another document's pricing to a caller without read access + 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(transaction_args): + if (transaction_args.parenttype or transaction_args.doctype) in purchase_doctypes: + if transaction_args.get("is_internal_supplier"): + return False + return bool(cint(frappe.get_cached_value("Buying Settings", "None", "maintain_same_rate"))) + + if transaction_args.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, source_row): + """Copies the full pricing block so a manual discount or margin on the source row 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, args): if frappe.db.exists("Product Bundle", {"name": args.item_code, "disabled": 0}, cache=True): valuation_rate = 0.0 @@ -1522,13 +1595,21 @@ def apply_price_list(args, as_doc=False, doc=None): def apply_price_list_on_item(args, doc=None): item_doc = frappe.db.get_value("Item", args.item_code, ["name", "variant_of"], as_dict=1) - item_details = get_price_list_rate(args, item_doc) + + source_row = get_rate_locked_source_row(args, doc) + if source_row: + item_details = frappe._dict() + lock_source_rate(item_details, source_row) + else: + item_details = get_price_list_rate(args, item_doc) args.conversion_factor = flt(args.conversion_factor) or get_conversion_factor( args.item_code, args.uom ).get("conversion_factor", 1) args.stock_qty = flt(args.qty) * flt(args.conversion_factor) - item_details.update(get_pricing_rule_for_item(args, doc=doc)) + + if not source_row: + item_details.update(get_pricing_rule_for_item(args, 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 d0b725b9179..398d7e603b5 100644 --- a/erpnext/stock/tests/test_get_item_details.py +++ b/erpnext/stock/tests/test_get_item_details.py @@ -177,3 +177,248 @@ class TestGetItemDetail(FrappeTestCase): 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(): + args = 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(args, 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_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, + } + ], + } + args = 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(args, 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} + ], + } + args = 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(args, 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_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}], + } + args = frappe._dict(doctype="Purchase Receipt", child_docname="r1") + + # an authorized caller receives the source row + self.assertIsNotNone(get_rate_locked_source_row(args.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(args.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")