mirror of
https://github.com/frappe/erpnext.git
synced 2026-08-29 06:38:24 +00:00
* fix: keep source rate on re-fetch when maintain same rate is enabled (backport #57479) With "maintain same rate" on, re-fetching item details on a row mapped from a source document (e.g. a Purchase Order) pulled the latest Item Price, giving a rate the document can never be saved with. Skip the price list fetch for such rows and keep the source rate, both for a single-row re-fetch and the bulk apply_price_list path (price list / party / conversion rate change). The rate is read from the linked source row in the database (not the mutable target row) and permission-checked against the source document, so an unsaved edit can't lock in a different rate and a crafted request can't disclose another document's pricing. Fixes frappe/erpnext#57436 * fix: resolve linter findings in get_item_details Add missing type hints on the whitelisted get_item_details signature and rename maintain_same_rate_enabled's sole "args" parameter, both flagged by the semgrep security/code-quality rules. Also drops an extra blank line that ruff-format rejected. * fix: widen get_item_details doc type hint to include Document accounts_controller.py calls get_item_details(args, self, ...) during validate, passing the transaction Document itself, not a dict/JSON string. The narrower hint tripped Frappe's runtime argument type validation on every whitelisted call with a live Document, failing test-record creation across the suite.
This commit is contained in:
Jatin3128
@@ -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
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user