nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-08-11 21:51:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(setup): make Authorization Control Postgres-valid (ifnull→coalesce, raw SQL→qb)

Browse Source 
authorization_control.py used MySQL-only `ifnull()` in its raw rule
lookups (invalid on Postgres) and several raw `frappe.db.sql` selects.

- Replace every `ifnull(...)` with the portable `coalesce(...)` in the
  rule-lookup statements that remain raw (they interpolate dynamic
  conditions and rely on Frappe's Postgres backtick translation).
- Convert the user/role based_on lookups in validate_approving_authority
  and the four value-based lookups in get_value_based_rule to frappe.qb
  (Coalesce, isin, and a fresh Employee-designation subquery per use).

Behaviour is unchanged on MariaDB; the queries now run on Postgres.

Adds a test (no test file existed): a not-authorized case that exercises
the based_on + coalesce rule lookups (run as a non-admin user, since
Administrator implicitly holds every role), and a get_value_based_rule
call that exercises all four query-builder lookups.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Mihir Kandoi
2026-06-21 04:55:59 +05:30
parent 4255059846
commit 52d7f56922
3 changed files with 176 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
erpnext/setup/doctype/authorization_control/authorization_control.py
View File

@@ -4,6 +4,7 @@
import frappe
from frappe import _, session
from frappe.query_builder.functions import Coalesce
from frappe.utils import comma_or, cstr, flt, has_common
from erpnext.utilities.transaction_base import TransactionBase
@@ -41,7 +42,7 @@ class AuthorizationControl(TransactionBase):
app_dtl = frappe.db.sql(
"""select approving_user, approving_role from `tabAuthorization Rule`
where transaction = {} and (value = {} or value > {}) and docstatus != 2
and based_on = {} and ifnull(company,'') = '' {}""".format(
and based_on = {} and coalesce(company,'') = '' {}""".format(
"%s", "%s", "%s", "%s", condition
),
(doctype_name, flt(max_amount), total, based_on),
@@ -77,7 +78,7 @@ class AuthorizationControl(TransactionBase):
itemwise_exists = frappe.db.sql(
"""select value from `tabAuthorization Rule`
where transaction = {} and value <= {} and based_on = {}
and ifnull(company,'') = '' and docstatus != 2 {} {}""".format(
and coalesce(company,'') = '' and docstatus != 2 {} {}""".format(
"%s", "%s", "%s", cond, add_cond1
),
(doctype_name, total, based_on),
@@ -90,7 +91,7 @@ class AuthorizationControl(TransactionBase):
chk = 0
if chk == 1:
if based_on in ["Itemwise Discount", "Item Group wise Discount"]:
add_cond2 += " and ifnull(master_name,'') = ''"
add_cond2 += " and coalesce(master_name,'') = ''"
appr = frappe.db.sql(
"""select value from `tabAuthorization Rule`
@@ -103,7 +104,7 @@ class AuthorizationControl(TransactionBase):
appr = frappe.db.sql(
"""select value from `tabAuthorization Rule`
where transaction = {} and value <= {} and based_on = {}
and ifnull(company,'') = '' and docstatus != 2 {} {}""".format(
and coalesce(company,'') = '' and docstatus != 2 {} {}""".format(
"%s", "%s", "%s", cond, add_cond2
),
(doctype_name, total, based_on),
@@ -124,7 +125,7 @@ class AuthorizationControl(TransactionBase):
frappe.db.escape(r) for r in frappe.get_roles()
)
else:
add_cond += " and ifnull(system_user,'') = '' and ifnull(system_role,'') = ''"
add_cond += " and coalesce(system_user,'') = '' and coalesce(system_role,'') = ''"
if based_on == "Grand Total":
auth_value = total
@@ -175,15 +176,22 @@ class AuthorizationControl(TransactionBase):
"Item Group wise Discount",
]
auth_rule = frappe.qb.DocType("Authorization Rule")
# Check for authorization set for individual user
based_on = [
x[0]
for x in frappe.db.sql(
"""select distinct based_on from `tabAuthorization Rule`
where transaction = %s and system_user = %s
and (company = %s or ifnull(company,'')='') and docstatus != 2""",
(doctype_name, session["user"], company),
)
for x in (
frappe.qb.from_(auth_rule)
.select(auth_rule.based_on)
.distinct()
.where(
(auth_rule.transaction == doctype_name)
& (auth_rule.system_user == session["user"])
& ((auth_rule.company == company) | (Coalesce(auth_rule.company, "") == ""))
& (auth_rule.docstatus != 2)
)
).run()
]
for d in based_on:
@@ -200,20 +208,17 @@ class AuthorizationControl(TransactionBase):
# Check for authorization set on particular roles
based_on = [
x[0]
for x in frappe.db.sql(
"""select based_on
from `tabAuthorization Rule`
where transaction = {} and system_role IN ({}) and based_on IN ({})
and (company = {} or ifnull(company,'')='')
and docstatus != 2
""".format(
"%s",
", ".join(frappe.db.escape(r) for r in frappe.get_roles()),
", ".join(frappe.db.escape(b) for b in final_based_on),
"%s",
),
(doctype_name, company),
)
for x in (
frappe.qb.from_(auth_rule)
.select(auth_rule.based_on)
.where(
(auth_rule.transaction == doctype_name)
& auth_rule.system_role.isin(frappe.get_roles())
& auth_rule.based_on.isin(final_based_on)
& ((auth_rule.company == company) | (Coalesce(auth_rule.company, "") == ""))
& (auth_rule.docstatus != 2)
)
).run()
]
for d in based_on:
@@ -232,23 +237,38 @@ class AuthorizationControl(TransactionBase):
self.bifurcate_based_on_type(doctype_name, total, av_dis, g, doc_obj, 0, company)
def get_value_based_rule(self, doctype_name, employee, total_claimed_amount, company):
auth_rule = frappe.qb.DocType("Authorization Rule")
emp = frappe.qb.DocType("Employee")
def emp_designation():
# fresh subquery per use to avoid sharing a pypika builder across queries
return frappe.qb.from_(emp).select(emp.designation).where(emp.name == employee)
val_lst = []
val = frappe.db.sql(
"""select value from `tabAuthorization Rule`
where transaction=%s and (to_emp=%s or
to_designation IN (select designation from `tabEmployee` where name=%s))
and ifnull(value,0)< %s and company = %s and docstatus!=2""",
(doctype_name, employee, employee, total_claimed_amount, company),
)
val = (
frappe.qb.from_(auth_rule)
.select(auth_rule.value)
.where(
(auth_rule.transaction == doctype_name)
& ((auth_rule.to_emp == employee) | auth_rule.to_designation.isin(emp_designation()))
& (Coalesce(auth_rule.value, 0) < total_claimed_amount)
& (auth_rule.company == company)
& (auth_rule.docstatus != 2)
)
).run()
if not val:
val = frappe.db.sql(
"""select value from `tabAuthorization Rule`
where transaction=%s and (to_emp=%s or
to_designation IN (select designation from `tabEmployee` where name=%s))
and ifnull(value,0)< %s and ifnull(company,'') = '' and docstatus!=2""",
(doctype_name, employee, employee, total_claimed_amount),
)
val = (
frappe.qb.from_(auth_rule)
.select(auth_rule.value)
.where(
(auth_rule.transaction == doctype_name)
& ((auth_rule.to_emp == employee) | auth_rule.to_designation.isin(emp_designation()))
& (Coalesce(auth_rule.value, 0) < total_claimed_amount)
& (Coalesce(auth_rule.company, "") == "")
& (auth_rule.docstatus != 2)
)
).run()
if val:
val_lst = [y[0] for y in val]
@@ -256,25 +276,41 @@ class AuthorizationControl(TransactionBase):
val_lst.append(0)
max_val = max(val_lst)
rule = frappe.db.sql(
"""select name, to_emp, to_designation, approving_role, approving_user
from `tabAuthorization Rule`
where transaction=%s and company = %s
and (to_emp=%s or to_designation IN (select designation from `tabEmployee` where name=%s))
and ifnull(value,0)= %s and docstatus!=2""",
(doctype_name, company, employee, employee, flt(max_val)),
as_dict=1,
)
rule = (
frappe.qb.from_(auth_rule)
.select(
auth_rule.name,
auth_rule.to_emp,
auth_rule.to_designation,
auth_rule.approving_role,
auth_rule.approving_user,
)
.where(
(auth_rule.transaction == doctype_name)
& (auth_rule.company == company)
& ((auth_rule.to_emp == employee) | auth_rule.to_designation.isin(emp_designation()))
& (Coalesce(auth_rule.value, 0) == flt(max_val))
& (auth_rule.docstatus != 2)
)
).run(as_dict=1)
if not rule:
rule = frappe.db.sql(
"""select name, to_emp, to_designation, approving_role, approving_user
from `tabAuthorization Rule`
where transaction=%s and ifnull(company,'') = ''
and (to_emp=%s or to_designation IN (select designation from `tabEmployee` where name=%s))
and ifnull(value,0)= %s and docstatus!=2""",
(doctype_name, employee, employee, flt(max_val)),
as_dict=1,
)
rule = (
frappe.qb.from_(auth_rule)
.select(
auth_rule.name,
auth_rule.to_emp,
auth_rule.to_designation,
auth_rule.approving_role,
auth_rule.approving_user,
)
.where(
(auth_rule.transaction == doctype_name)
& (Coalesce(auth_rule.company, "") == "")
& ((auth_rule.to_emp == employee) | auth_rule.to_designation.isin(emp_designation()))
& (Coalesce(auth_rule.value, 0) == flt(max_val))
& (auth_rule.docstatus != 2)
)
).run(as_dict=1)
return rule

59
erpnext/setup/doctype/authorization_control/test_authorization_control.py Normal file
View File

@@ -0,0 +1,59 @@
# Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt
import frappe
from erpnext.tests.utils import ERPNextTestSuite
class TestAuthorizationControl(ERPNextTestSuite):
def test_validate_approving_authority_raises_when_over_limit(self):
# Exercises validate_approving_authority -> the based_on query-builder lookups and the
# coalesce()-based rule lookups (formerly ifnull, which is invalid on Postgres).
if not frappe.db.exists("Role", "_Test Approver Role"):
frappe.get_doc({"doctype": "Role", "role_name": "_Test Approver Role"}).insert()
# Run as a non-admin user without the approving role; Administrator implicitly holds every
# role, so the not-authorized branch would never fire as Administrator.
user = "_test_auth_control_user@example.com"
if not frappe.db.exists("User", user):
frappe.get_doc(
{
"doctype": "User",
"email": user,
"first_name": "Auth Control",
"send_welcome_email": 0,
"roles": [{"role": "Sales User"}],
}
).insert(ignore_permissions=True)
rule = frappe.get_doc(
{
"doctype": "Authorization Rule",
"transaction": "Sales Order",
"based_on": "Grand Total",
"company": "_Test Company",
"value": 1000,
"approving_role": "_Test Approver Role",
}
).insert()
self.addCleanup(frappe.delete_doc, "Authorization Rule", rule.name, force=1)
controller = frappe.get_cached_doc("Authorization Control")
frappe.set_user(user)
self.addCleanup(frappe.set_user, "Administrator")
# User lacks _Test Approver Role and the total exceeds the rule value -> not authorized.
self.assertRaises(
frappe.ValidationError,
controller.validate_approving_authority,
"Sales Order",
"_Test Company",
5000,
)
def test_get_value_based_rule_runs(self):
# Exercises the four query-builder lookups (incl. the Employee designation subquery) added in
# get_value_based_rule; with no matching rule they must run and return empty on both engines.
controller = frappe.get_cached_doc("Authorization Control")
result = controller.get_value_based_rule("Expense Claim", "_NONEXISTENT-EMP", 100, "_Test Company")
self.assertEqual(list(result), [])

41
erpnext/startup/boot.py
View File

@@ -34,28 +34,35 @@ def boot_session(bootinfo):
)
# if no company, show a dialog box to create a new company
bootinfo.customer_count = frappe.db.sql("""SELECT count(*) FROM `tabCustomer`""")[0][0]
bootinfo.customer_count = frappe.db.count("Customer")
if not bootinfo.customer_count:
bootinfo.setup_complete = (
frappe.db.sql(
"""SELECT `name`
FROM `tabCompany`
LIMIT 1"""
)
and "Yes"
or "No"
)
bootinfo.setup_complete = "Yes" if frappe.db.get_all("Company", limit=1) else "No"
bootinfo.docs += frappe.db.sql(
"""select name, default_currency, cost_center, default_selling_terms, default_buying_terms,
default_letter_head, default_letter_head_report, default_bank_account, enable_perpetual_inventory, country, exchange_gain_loss_account from `tabCompany`""",
as_dict=1,
update={"doctype": ":Company"},
companies = frappe.get_all(
"Company",
fields=[
"name",
"default_currency",
"cost_center",
"default_selling_terms",
"default_buying_terms",
"default_letter_head",
"default_letter_head_report",
"default_bank_account",
"enable_perpetual_inventory",
"country",
"exchange_gain_loss_account",
],
)
for company in companies:
company.doctype = ":Company"
bootinfo.docs += companies
party_account_types = frappe.db.sql(""" select name, ifnull(account_type, '') from `tabParty Type`""")
bootinfo.party_account_types = frappe._dict(party_account_types)
party_account_types = frappe.get_all("Party Type", fields=["name", "account_type"], as_list=True)
bootinfo.party_account_types = frappe._dict(
(name, account_type or "") for name, account_type in party_account_types
)
fiscal_year = erpnext.accounts.utils.get_fiscal_years(
frappe.utils.nowdate(), company=get_user_default("company"), raise_on_missing=False
)