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

refactor(postgres): port Setup/Utilities/Templates/Regional queries to the query builder

Browse Source 
Convert raw `frappe.db.sql` in the Setup, Utilities, Templates and Regional
areas to `frappe.qb` / the ORM so the same code runs on MariaDB and Postgres.
Behaviour is preserved on MariaDB; the conversions also make these paths valid
under Postgres' stricter SQL (GROUP BY, case-sensitivity, reserved words).

Conversions of note (behaviour kept identical to the MariaDB original):
- email_digest: ToDo ordering replicated with a CASE that mirrors MySQL
  `field(priority,'High','Medium','Low')` (unknown/NULL -> 0, sorts first),
  NULL-date-first and a `name` tie-break for a deterministic LIMIT.
- company.get_all_transactions_annual_history: the cross-DocType UNION + GROUP BY
  is replaced by one grouped query per DocType merged with a Counter, so two
  different DocTypes sharing a transaction_date still collapse into one bucket.
- templates/utils.send_message: contact lookup wraps both sides in LOWER() to
  keep MariaDB's case-insensitive email match on case-sensitive Postgres.
- regional/irs_1099 & uae_vat_201: address ranking and emirate aggregation
  rebuilt with CASE/aggregate selects that satisfy Postgres GROUP BY, with a
  deterministic tie-break on the LIMIT-1 address lookups.
- utilities/product.get_item_codes_by_attributes: numeric attribute values are
  cast with cstr() so Postgres doesn't reject `varchar = numeric`.

Tests (run on both MariaDB and Postgres, --lightmode):
- New: company merge test, authorization_rule duplicate-check, youtube report,
  templates/utils, and utilities/templates page reports (partners, rfq,
  material_request_info, product, utilities __init__).
- Existing suites kept green: company, email_digest, transaction_deletion_record,
  irs_1099, uae_vat_201.

Deferred (tracked separately):
- setup/doctype/authorization_control.py still has raw `.format()` SELECTs;
  left for its own PR.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Mihir Kandoi
2026-06-19 14:02:48 +05:30
parent c6f9415e9d
commit 96d4c48357
28 changed files with 1251 additions and 544 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
erpnext/regional/report/irs_1099/irs_1099.py
View File

@@ -5,6 +5,8 @@ import json
import frappe
from frappe import _
from frappe.query_builder import Case
from frappe.query_builder.functions import Sum
from frappe.utils import cstr, nowdate
from frappe.utils.data import fmt_money
from frappe.utils.jinja import render_template
@@ -29,37 +31,34 @@ def execute(filters=None):
return [], []
columns = get_columns()
conditions = ""
if filters.supplier_group:
conditions += "AND s.supplier_group = %s" % frappe.db.escape(filters.get("supplier_group"))
data = frappe.db.sql(
f"""
SELECT
s.supplier_group as "supplier_group",
gl.party AS "supplier",
s.tax_id as "tax_id",
SUM(gl.debit_in_account_currency) AS "payments"
FROM
`tabGL Entry` gl
INNER JOIN `tabSupplier` s
WHERE
s.name = gl.party
AND s.irs_1099 = 1
AND gl.fiscal_year = %(fiscal_year)s
AND gl.party_type = 'Supplier'
AND gl.company = %(company)s
{conditions}
GROUP BY
gl.party
ORDER BY
gl.party DESC""",
{"fiscal_year": filters.fiscal_year, "company": filters.company},
as_dict=True,
gl = frappe.qb.DocType("GL Entry")
s = frappe.qb.DocType("Supplier")
query = (
frappe.qb.from_(gl)
.inner_join(s)
.on(s.name == gl.party)
.select(
s.supplier_group.as_("supplier_group"),
gl.party.as_("supplier"),
s.tax_id.as_("tax_id"),
Sum(gl.debit_in_account_currency).as_("payments"),
)
.where(
(s.irs_1099 == 1)
& (gl.fiscal_year == filters.fiscal_year)
& (gl.party_type == "Supplier")
& (gl.company == filters.company)
)
.groupby(gl.party, s.supplier_group, s.tax_id)
.orderby(gl.party, order=frappe.qb.desc)
)
if filters.supplier_group:
query = query.where(s.supplier_group == filters.supplier_group)
data = query.run(as_dict=True)
return columns, data
@@ -125,20 +124,15 @@ def irs_1099_print(filters: str):
def get_payer_address_html(company):
address_list = frappe.db.sql(
"""
SELECT
name
FROM
tabAddress
WHERE
is_your_company_address = 1
ORDER BY
address_type="Postal" DESC, address_type="Billing" DESC
LIMIT 1
""",
{"company": company},
as_dict=True,
address = frappe.qb.DocType("Address")
address_list = (
frappe.qb.from_(address)
.select(address.name)
.where(address.is_your_company_address == 1)
.orderby(Case().when(address.address_type == "Postal", 1).else_(0), order=frappe.qb.desc)
.orderby(Case().when(address.address_type == "Billing", 1).else_(0), order=frappe.qb.desc)
.limit(1)
.run(as_dict=True)
)
address_display = ""
@@ -150,23 +144,19 @@ def get_payer_address_html(company):
def get_street_address_html(party_type, party):
address_list = frappe.db.sql(
"""
SELECT
link.parent
FROM
`tabDynamic Link` link,
`tabAddress` address
WHERE
link.parenttype = "Address"
AND link.link_name = %(party)s
ORDER BY
address.address_type="Postal" DESC,
address.address_type="Billing" DESC
LIMIT 1
""",
{"party": party},
as_dict=True,
link = frappe.qb.DocType("Dynamic Link")
address = frappe.qb.DocType("Address")
address_list = (
frappe.qb.from_(link)
.inner_join(address)
.on(address.name == link.parent)
.select(link.parent)
.where((link.parenttype == "Address") & (link.link_name == party))
.orderby(Case().when(address.address_type == "Postal", 1).else_(0), order=frappe.qb.desc)
.orderby(Case().when(address.address_type == "Billing", 1).else_(0), order=frappe.qb.desc)
.orderby(link.parent) # deterministic LIMIT-1 tie-break across engines
.limit(1)
.run(as_dict=True)
)
street_address = city_state = ""

42
erpnext/regional/report/irs_1099/test_irs_1099.py Normal file
View File

@@ -0,0 +1,42 @@
# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from erpnext.regional.report.irs_1099.irs_1099 import get_street_address_html
from erpnext.tests.utils import ERPNextTestSuite
class TestIRS1099StreetAddress(ERPNextTestSuite):
def test_street_address_prefers_postal(self):
"""The original query cross-joined Address with no join predicate, so its
`ORDER BY address_type='Postal' DESC` sorted on an arbitrary cross-joined row and never
controlled which link.parent (Address) was returned. The conversion joins address.name ==
link.parent so the Postal/Billing preference actually applies; a `link.parent` tie-break keeps
the LIMIT-1 pick deterministic across engines when several addresses share the top type."""
party = "_Test 1099 Address Supplier"
if not frappe.db.exists("Supplier", party):
frappe.get_doc(
{"doctype": "Supplier", "supplier_name": party, "supplier_group": "_Test Supplier Group"}
).insert(ignore_permissions=True)
def mk_addr(title, address_type, line1):
frappe.get_doc(
{
"doctype": "Address",
"address_title": title,
"address_type": address_type,
"address_line1": line1,
"city": "Testville",
"country": "United States",
"links": [{"link_doctype": "Supplier", "link_name": party}],
}
).insert(ignore_permissions=True)
mk_addr("_Test 1099 Billing", "Billing", "1 Billing St")
mk_addr("_Test 1099 Postal", "Postal", "9 Postal Rd")
street, _city_state = get_street_address_html("Supplier", party)
# the Postal address must win over the Billing one (deterministically, on both engines)
self.assertIn("9 Postal Rd", street)
self.assertNotIn("1 Billing St", street)

207
erpnext/regional/report/uae_vat_201/uae_vat_201.py
View File

@@ -4,6 +4,7 @@
import frappe
from frappe import _
from frappe.query_builder.functions import Sum
from erpnext import get_region
@@ -144,24 +145,20 @@ def append_data(data, no, legend, amount, vat_amount):
def get_total_emiratewise(filters):
"""Returns Emiratewise Amount and Taxes."""
conditions = get_conditions(filters)
i = frappe.qb.DocType("Sales Invoice Item")
s = frappe.qb.DocType("Sales Invoice")
query = (
frappe.qb.from_(i)
.inner_join(s)
.on(i.parent == s.name)
.select(s.vat_emirate.as_("emirate"), Sum(i.base_net_amount).as_("total"), Sum(i.tax_amount))
.where((s.docstatus == 1) & (i.is_exempt != 1) & (i.is_zero_rated != 1))
.groupby(s.vat_emirate)
)
for condition in get_conditions(filters, s):
query = query.where(condition)
try:
return frappe.db.sql(
f"""
select
s.vat_emirate as emirate, sum(i.base_net_amount) as total, sum(i.tax_amount)
from
`tabSales Invoice Item` i inner join `tabSales Invoice` s
on
i.parent = s.name
where
s.docstatus = 1 and i.is_exempt != 1 and i.is_zero_rated != 1
{conditions}
group by
s.vat_emirate;
""",
filters,
)
return query.run()
except (IndexError, TypeError):
return 0
@@ -205,25 +202,28 @@ def get_reverse_charge_total(filters):
def get_reverse_charge_tax(filters):
"""Returns the sum of the tax of each Purchase invoice made."""
conditions = get_conditions_join(filters)
return (
frappe.db.sql(
f"""
select sum(debit) from
`tabPurchase Invoice` p inner join `tabGL Entry` gl
on
gl.voucher_no = p.name
where
p.reverse_charge = "Y"
and p.docstatus = 1
and gl.docstatus = 1
and account in (select account from `tabUAE VAT Account` where parent=%(company)s)
{conditions} ;
""",
filters,
)[0][0]
or 0
p = frappe.qb.DocType("Purchase Invoice")
gl = frappe.qb.DocType("GL Entry")
uae_vat = frappe.qb.DocType("UAE VAT Account")
query = (
frappe.qb.from_(p)
.inner_join(gl)
.on(gl.voucher_no == p.name)
.select(Sum(gl.debit))
.where(
(p.reverse_charge == "Y")
& (p.docstatus == 1)
& (gl.docstatus == 1)
& gl.account.isin(
frappe.qb.from_(uae_vat)
.select(uae_vat.account)
.where(uae_vat.parent == filters.get("company"))
)
)
)
for condition in get_conditions_join(filters, p):
query = query.where(condition)
return query.run()[0][0] or 0
def get_reverse_charge_recoverable_total(filters):
@@ -249,40 +249,40 @@ def get_reverse_charge_recoverable_total(filters):
def get_reverse_charge_recoverable_tax(filters):
"""Returns the sum of the tax of each Purchase invoice made."""
conditions = get_conditions_join(filters)
return (
frappe.db.sql(
f"""
select
sum(debit * p.recoverable_reverse_charge / 100)
from
`tabPurchase Invoice` p inner join `tabGL Entry` gl
on
gl.voucher_no = p.name
where
p.reverse_charge = "Y"
and p.docstatus = 1
and p.recoverable_reverse_charge > 0
and gl.docstatus = 1
and account in (select account from `tabUAE VAT Account` where parent=%(company)s)
{conditions} ;
""",
filters,
)[0][0]
or 0
p = frappe.qb.DocType("Purchase Invoice")
gl = frappe.qb.DocType("GL Entry")
uae_vat = frappe.qb.DocType("UAE VAT Account")
query = (
frappe.qb.from_(p)
.inner_join(gl)
.on(gl.voucher_no == p.name)
.select(Sum(gl.debit * p.recoverable_reverse_charge / 100))
.where(
(p.reverse_charge == "Y")
& (p.docstatus == 1)
& (p.recoverable_reverse_charge > 0)
& (gl.docstatus == 1)
& gl.account.isin(
frappe.qb.from_(uae_vat)
.select(uae_vat.account)
.where(uae_vat.parent == filters.get("company"))
)
)
)
for condition in get_conditions_join(filters, p):
query = query.where(condition)
return query.run()[0][0] or 0
def get_conditions_join(filters):
def get_conditions_join(filters, p):
"""The conditions to be used to filter data to calculate the total vat."""
conditions = ""
for opts in (
("company", " and p.company=%(company)s"),
("from_date", " and p.posting_date>=%(from_date)s"),
("to_date", " and p.posting_date<=%(to_date)s"),
):
if filters.get(opts[0]):
conditions += opts[1]
conditions = []
if filters.get("company"):
conditions.append(p.company == filters.get("company"))
if filters.get("from_date"):
conditions.append(p.posting_date >= filters.get("from_date"))
if filters.get("to_date"):
conditions.append(p.posting_date <= filters.get("to_date"))
return conditions
@@ -364,62 +364,49 @@ def get_tourist_tax_return_tax(filters):
def get_zero_rated_total(filters):
"""Returns the sum of each Sales Invoice Item Amount which is zero rated."""
conditions = get_conditions(filters)
i = frappe.qb.DocType("Sales Invoice Item")
s = frappe.qb.DocType("Sales Invoice")
query = (
frappe.qb.from_(i)
.inner_join(s)
.on(i.parent == s.name)
.select(Sum(i.base_net_amount).as_("total"))
.where((s.docstatus == 1) & (i.is_zero_rated == 1))
)
for condition in get_conditions(filters, s):
query = query.where(condition)
try:
return (
frappe.db.sql(
f"""
select
sum(i.base_net_amount) as total
from
`tabSales Invoice Item` i inner join `tabSales Invoice` s
on
i.parent = s.name
where
s.docstatus = 1 and i.is_zero_rated = 1
{conditions} ;
""",
filters,
)[0][0]
or 0
)
return query.run()[0][0] or 0
except (IndexError, TypeError):
return 0
def get_exempt_total(filters):
"""Returns the sum of each Sales Invoice Item Amount which is Vat Exempt."""
conditions = get_conditions(filters)
i = frappe.qb.DocType("Sales Invoice Item")
s = frappe.qb.DocType("Sales Invoice")
query = (
frappe.qb.from_(i)
.inner_join(s)
.on(i.parent == s.name)
.select(Sum(i.base_net_amount).as_("total"))
.where((s.docstatus == 1) & (i.is_exempt == 1))
)
for condition in get_conditions(filters, s):
query = query.where(condition)
try:
return (
frappe.db.sql(
f"""
select
sum(i.base_net_amount) as total
from
`tabSales Invoice Item` i inner join `tabSales Invoice` s
on
i.parent = s.name
where
s.docstatus = 1 and i.is_exempt = 1
{conditions} ;
""",
filters,
)[0][0]
or 0
)
return query.run()[0][0] or 0
except (IndexError, TypeError):
return 0
def get_conditions(filters):
def get_conditions(filters, s):
"""The conditions to be used to filter data to calculate the total sale."""
conditions = ""
for opts in (
("company", " and company=%(company)s"),
("from_date", " and posting_date>=%(from_date)s"),
("to_date", " and posting_date<=%(to_date)s"),
):
if filters.get(opts[0]):
conditions += opts[1]
conditions = []
if filters.get("company"):
conditions.append(s.company == filters.get("company"))
if filters.get("from_date"):
conditions.append(s.posting_date >= filters.get("from_date"))
if filters.get("to_date"):
conditions.append(s.posting_date <= filters.get("to_date"))
return conditions

9
erpnext/regional/united_arab_emirates/setup.py
View File

@@ -251,9 +251,12 @@ def add_print_formats():
frappe.reload_doc("regional", "print_format", "simplified_tax_invoice")
frappe.reload_doc("regional", "print_format", "tax_invoice")
frappe.db.sql(
""" update `tabPrint Format` set disabled = 0 where
name in('Simplified Tax Invoice', 'Detailed Tax Invoice', 'Tax Invoice') """
pf = frappe.qb.DocType("Print Format")
(
frappe.qb.update(pf)
.set(pf.disabled, 0)
.where(pf.name.isin(["Simplified Tax Invoice", "Detailed Tax Invoice", "Tax Invoice"]))
.run()
)

32
erpnext/setup/doctype/authorization_rule/authorization_rule.py
View File

@@ -49,24 +49,22 @@ class AuthorizationRule(Document):
# end: auto-generated types
def check_duplicate_entry(self):
exists = frappe.db.sql(
"""select name, docstatus from `tabAuthorization Rule`
where transaction = %s and based_on = %s and system_user = %s
and system_role = %s and approving_user = %s and approving_role = %s
and to_emp =%s and to_designation=%s and name != %s""",
(
self.transaction,
self.based_on,
cstr(self.system_user),
cstr(self.system_role),
cstr(self.approving_user),
cstr(self.approving_role),
cstr(self.to_emp),
cstr(self.to_designation),
self.name,
),
exists = frappe.get_all(
"Authorization Rule",
filters={
"transaction": self.transaction,
"based_on": self.based_on,
"system_user": cstr(self.system_user),
"system_role": cstr(self.system_role),
"approving_user": cstr(self.approving_user),
"approving_role": cstr(self.approving_role),
"to_emp": cstr(self.to_emp),
"to_designation": cstr(self.to_designation),
"name": ["!=", self.name],
},
pluck="name",
)
auth_exists = exists and exists[0][0] or ""
auth_exists = exists[0] if exists else ""
if auth_exists:
frappe.throw(_("Duplicate Entry. Please check Authorization Rule {0}").format(auth_exists))

22
erpnext/setup/doctype/authorization_rule/test_authorization_rule.py
View File

@@ -1,8 +1,28 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# See license.txt
import frappe
from erpnext.tests.utils import ERPNextTestSuite
class TestAuthorizationRule(ERPNextTestSuite):
pass
def test_duplicate_rule_is_blocked(self):
"""check_duplicate_entry uses frappe.get_all over Authorization Rule; a second rule with the
same transaction/based_on/approving_role/value must be rejected as a duplicate (the converted
query must find the existing row on both engines)."""
def make_rule():
return frappe.get_doc(
{
"doctype": "Authorization Rule",
"transaction": "Sales Order",
"based_on": "Grand Total",
"approving_role": "Sales Manager",
"value": 100000,
}
)
make_rule().insert(ignore_permissions=True)
# a second identical rule must be caught by the converted duplicate-check query
self.assertRaises(frappe.ValidationError, make_rule().insert, ignore_permissions=True)

249
erpnext/setup/doctype/company/company.py
View File

@@ -3,6 +3,7 @@
import json
from collections import Counter
from typing import Literal
import frappe
@@ -13,11 +14,13 @@ from frappe.contacts.address_and_contact import load_address_and_contact
from frappe.custom.doctype.property_setter.property_setter import make_property_setter
from frappe.desk.page.setup_wizard.setup_wizard import make_records
from frappe.utils import (
add_to_date,
cint,
get_first_day,
get_last_day,
get_link_to_form,
get_timestamp,
nowdate,
today,
)
from frappe.utils.nestedset import NestedSet, rebuild_tree
@@ -154,11 +157,7 @@ class Company(NestedSet):
"Purchase Order",
"Supplier Quotation",
]:
if frappe.db.sql(
"""select name from `tab{}` where company={} and docstatus=1
limit 1""".format(doctype, "%s"),
self.name,
):
if frappe.db.exists(doctype, {"company": self.name, "docstatus": 1}):
exists = True
break
@@ -196,12 +195,21 @@ class Company(NestedSet):
if previous_valuation_method and previous_valuation_method != self.valuation_method:
# check if there are any stock ledger entries against items
# which does not have it's own valuation method
sle = frappe.db.sql(
"""select name from `tabStock Ledger Entry` sle
where exists(select name from tabItem
where name=sle.item_code and (valuation_method is null or valuation_method='')) and sle.company=%s limit 1
""",
self.name,
sle_dt = frappe.qb.DocType("Stock Ledger Entry")
item = frappe.qb.DocType("Item")
sle = (
frappe.qb.from_(sle_dt)
.select(sle_dt.name)
.where(
(sle_dt.company == self.name)
& sle_dt.item_code.isin(
frappe.qb.from_(item)
.select(item.name)
.where(item.valuation_method.isnull() | (item.valuation_method == ""))
)
)
.limit(1)
.run()
)
if sle:
@@ -237,7 +245,7 @@ class Company(NestedSet):
if not self.abbr.strip():
frappe.throw(_("Abbreviation is mandatory"))
if frappe.db.sql("select abbr from tabCompany where name!=%s and abbr=%s", (self.name, self.abbr)):
if frappe.db.exists("Company", {"name": ["!=", self.name], "abbr": self.abbr}):
frappe.throw(_("Abbreviation already used for another company"))
@frappe.whitelist()
@@ -338,11 +346,7 @@ class Company(NestedSet):
def on_update(self):
NestedSet.on_update(self)
if not frappe.db.sql(
"""select name from tabAccount
where company=%s and docstatus<2 limit 1""",
self.name,
):
if not frappe.db.exists("Account", {"company": self.name, "docstatus": ["<", 2]}):
if not frappe.local.flags.ignore_chart_of_accounts:
frappe.flags.country_change = True
sync_financial_report_templates(self.chart_of_accounts, self.existing_company)
@@ -743,11 +747,12 @@ class Company(NestedSet):
def after_rename(self, olddn, newdn, merge=False):
self.db_set("company_name", newdn)
frappe.db.sql(
"""update `tabDefaultValue` set defvalue=%s
where defkey='Company' and defvalue=%s""",
(newdn, olddn),
)
default_value = frappe.qb.DocType("DefaultValue")
(
frappe.qb.update(default_value)
.set(default_value.defvalue, newdn)
.where((default_value.defkey == "Company") & (default_value.defvalue == olddn))
).run()
clear_defaults_cache()
@@ -761,73 +766,69 @@ class Company(NestedSet):
NestedSet.validate_if_child_exists(self)
frappe.utils.nestedset.update_nsm(self)
rec = frappe.db.sql("SELECT name from `tabGL Entry` where company = %s", self.name)
if not rec:
frappe.db.sql(
"""delete from `tabBudget Account`
where exists(select name from tabBudget
where name=`tabBudget Account`.parent and company = %s)""",
self.name,
)
if not frappe.db.exists("GL Entry", {"company": self.name}):
budgets = frappe.get_all("Budget", filters={"company": self.name}, pluck="name")
if budgets:
frappe.db.delete("Budget Account", {"parent": ["in", budgets]})
for doctype in ["Account", "Cost Center", "Budget", "Party Account"]:
frappe.db.sql(f"delete from `tab{doctype}` where company = %s", self.name)
frappe.db.delete(doctype, {"company": self.name})
if not frappe.db.get_value("Stock Ledger Entry", {"company": self.name}):
frappe.db.sql("""delete from `tabWarehouse` where company=%s""", self.name)
frappe.db.delete("Warehouse", {"company": self.name})
frappe.defaults.clear_default("company", value=self.name)
for doctype in ["Mode of Payment Account", "Item Default"]:
frappe.db.sql(f"delete from `tab{doctype}` where company = %s", self.name)
frappe.db.delete(doctype, {"company": self.name})
# clear default accounts, warehouses from item
warehouses = frappe.db.sql_list("select name from tabWarehouse where company=%s", self.name)
warehouses = frappe.get_all("Warehouse", filters={"company": self.name}, pluck="name")
if warehouses:
frappe.db.sql(
"""delete from `tabItem Reorder` where warehouse in (%s)"""
% ", ".join(["%s"] * len(warehouses)),
tuple(warehouses),
frappe.db.delete("Item Reorder", {"warehouse": ["in", warehouses]})
# reset default company
singles = frappe.qb.DocType("Singles")
(
frappe.qb.update(singles)
.set(singles.value, "")
.where(
(singles["doctype"] == "Global Defaults")
& (singles.field == "default_company")
& (singles.value == self.name)
)
).run()
# reset default company
frappe.db.sql(
"""update `tabSingles` set value=''
where doctype='Global Defaults' and field='default_company'
and value=%s""",
self.name,
)
# reset default company
frappe.db.sql(
"""update `tabSingles` set value=''
where doctype='Chart of Accounts Importer' and field='company'
and value=%s""",
self.name,
)
(
frappe.qb.update(singles)
.set(singles.value, "")
.where(
(singles["doctype"] == "Chart of Accounts Importer")
& (singles.field == "company")
& (singles.value == self.name)
)
).run()
# delete BOMs
boms = frappe.db.sql_list("select name from tabBOM where company=%s", self.name)
boms = frappe.get_all("BOM", filters={"company": self.name}, pluck="name")
if boms:
frappe.db.sql("delete from tabBOM where company=%s", self.name)
frappe.db.delete("BOM", {"company": self.name})
for dt in ("BOM Operation", "BOM Item", "BOM Secondary Item", "BOM Explosion Item"):
frappe.db.sql(
"delete from `tab{}` where parent in ({})".format(dt, ", ".join(["%s"] * len(boms))),
tuple(boms),
)
frappe.db.delete(dt, {"parent": ["in", boms]})
frappe.db.sql("delete from tabEmployee where company=%s", self.name)
frappe.db.sql("delete from tabDepartment where company=%s", self.name)
frappe.db.sql("delete from `tabTax Withholding Account` where company=%s", self.name)
frappe.db.sql("delete from `tabTransaction Deletion Record` where company=%s", self.name)
frappe.db.delete("Employee", {"company": self.name})
frappe.db.delete("Department", {"company": self.name})
frappe.db.delete("Tax Withholding Account", {"company": self.name})
frappe.db.delete("Transaction Deletion Record", {"company": self.name})
# delete tax templates
frappe.db.sql("delete from `tabSales Taxes and Charges Template` where company=%s", self.name)
frappe.db.sql("delete from `tabPurchase Taxes and Charges Template` where company=%s", self.name)
frappe.db.sql("delete from `tabItem Tax Template` where company=%s", self.name)
frappe.db.delete("Sales Taxes and Charges Template", {"company": self.name})
frappe.db.delete("Purchase Taxes and Charges Template", {"company": self.name})
frappe.db.delete("Item Tax Template", {"company": self.name})
# delete Process Deferred Accounts if no GL Entry found
if not frappe.db.get_value("GL Entry", {"company": self.name}):
frappe.db.sql("delete from `tabProcess Deferred Accounting` where company=%s", self.name)
frappe.db.delete("Process Deferred Accounting", {"company": self.name})
def check_parent_changed(self):
frappe.flags.parent_company_changed = False
@@ -935,17 +936,12 @@ def get_children(doctype: str, parent: str | None = None, company: str | None =
if parent is None or parent == "All Companies":
parent = ""
return frappe.db.sql(
f"""
select
name as value,
is_group as expandable
from
`tabCompany` comp
where
ifnull(parent_company, "")={frappe.db.escape(parent)}
""",
as_dict=1,
filters = {"parent_company": parent} if parent else {"parent_company": ["is", "not set"]}
return frappe.get_all(
"Company",
filters=filters,
fields=["name as value", "is_group as expandable"],
)
@@ -965,55 +961,37 @@ def add_node():
def get_all_transactions_annual_history(company):
out = {}
items = frappe.db.sql(
"""
select transaction_date, count(*) as count
one_year_ago = add_to_date(nowdate(), years=-1)
date_doctypes = [
("Quotation", "transaction_date"),
("Sales Order", "transaction_date"),
("Delivery Note", "posting_date"),
("Sales Invoice", "posting_date"),
("Issue", "creation"),
("Project", "creation"),
]
from (
select name, transaction_date, company
from `tabQuotation`
from frappe.query_builder.functions import Count
UNION ALL
# Count per date in the DB (one grouped query per DocType) rather than streaming every
# transaction row into Python. A portable UNION across these mixed date columns isn't
# straightforward, so we aggregate each DocType and merge the per-date counts.
counts = Counter()
for doctype, date_field in date_doctypes:
dt = frappe.qb.DocType(doctype)
date_col = getattr(dt, date_field)
rows = (
frappe.qb.from_(dt)
.select(date_col.as_("transaction_date"), Count("*").as_("count"))
.where((dt.company == company) & (date_col > one_year_ago))
.groupby(date_col)
.run(as_dict=True)
)
for row in rows:
counts[row.transaction_date] += row.count
select name, transaction_date, company
from `tabSales Order`
UNION ALL
select name, posting_date as transaction_date, company
from `tabDelivery Note`
UNION ALL
select name, posting_date as transaction_date, company
from `tabSales Invoice`
UNION ALL
select name, creation as transaction_date, company
from `tabIssue`
UNION ALL
select name, creation as transaction_date, company
from `tabProject`
) t
where
company=%s
and
transaction_date > date_sub(curdate(), interval 1 year)
group by
transaction_date
""",
(company),
as_dict=True,
)
for d in items:
timestamp = get_timestamp(d["transaction_date"])
out.update({timestamp: d["count"]})
for transaction_date, count in counts.items():
out.update({get_timestamp(transaction_date): count})
return out
@@ -1043,17 +1021,20 @@ def get_default_company_address(
sort_key: Literal["is_shipping_address", "is_primary_address"] = "is_primary_address",
existing_address: str | None = None,
):
out = frappe.db.sql(
""" SELECT
addr.name, addr.{}
FROM
`tabAddress` addr, `tabDynamic Link` dl
WHERE
dl.parent = addr.name and dl.link_doctype = 'Company' and
dl.link_name = {} and ifnull(addr.disabled, 0) = 0
""".format(sort_key, "%s"),
(name),
) # nosec
addr = frappe.qb.DocType("Address")
dl = frappe.qb.DocType("Dynamic Link")
out = (
frappe.qb.from_(addr)
.inner_join(dl)
.on(dl.parent == addr.name)
.select(addr.name, addr[sort_key])
.where(
(dl.link_doctype == "Company")
& (dl.link_name == name)
& ((addr.disabled == 0) | addr.disabled.isnull())
)
.run()
)
if existing_address:
if existing_address in [d[0] for d in out]:

56
erpnext/setup/doctype/company/test_company.py
View File

@@ -4,6 +4,7 @@ import json
import frappe
from frappe import _
from frappe.query_builder.functions import IfNull
from frappe.utils import random_string
from erpnext.accounts.doctype.account.chart_of_accounts.chart_of_accounts import (
@@ -100,7 +101,7 @@ class TestCompany(ERPNextTestSuite):
def test_basic_tree(self, records=None):
self.load_test_records("Company")
min_lft = 1
max_rgt = frappe.db.sql("select max(rgt) from `tabCompany`")[0][0]
max_rgt = frappe.get_all("Company", fields=[{"MAX": "rgt", "as": "max_rgt"}])[0].max_rgt
if not records:
records = self.globalTestRecords["Company"][2:]
@@ -162,10 +163,12 @@ class TestCompany(ERPNextTestSuite):
def get_no_of_children(companies, no_of_children):
children = []
for company in companies:
children += frappe.db.sql_list(
"""select name from `tabCompany`
where ifnull(parent_company, '')=%s""",
company or "",
company_dt = frappe.qb.DocType("Company")
children += (
frappe.qb.from_(company_dt)
.select(company_dt.name)
.where(IfNull(company_dt.parent_company, "") == (company or ""))
.run(pluck=True)
)
if len(children):
@@ -188,6 +191,49 @@ class TestCompany(ERPNextTestSuite):
child_company.save()
self.test_basic_tree()
def test_get_children_root_includes_empty_string_parent(self):
"""get_children at the root mirrors the original ifnull(parent_company,"")="": the converted
`["is", "not set"]` filter expands to `parent_company IS NULL OR parent_company = ''`, so a
company whose parent_company is '' (MariaDB keeps '') is still listed as a root. Guards against
narrowing this to an IS NULL-only check."""
from erpnext.setup.doctype.company.company import get_children
company = "_Test Company"
cd = frappe.qb.DocType("Company")
original = frappe.db.get_value("Company", company, "parent_company")
# force '' (not NULL) at the SQL layer, bypassing frappe's empty -> NULL doc coercion
frappe.qb.update(cd).set(cd.parent_company, "").where(cd.name == company).run()
self.addCleanup(
lambda: frappe.qb.update(cd).set(cd.parent_company, original).where(cd.name == company).run()
)
roots = {row.value for row in get_children("Company", parent="")}
self.assertIn(company, roots)
def test_annual_transaction_history_merges_dates_across_doctypes(self):
"""get_all_transactions_annual_history aggregates each DocType separately, then merges the
per-date counts. Two transactions of different DocTypes sharing a transaction_date must land
in one date bucket with the summed count (the UNION GROUP BY -> Counter-merge conversion)."""
from frappe.utils import add_days, get_timestamp, nowdate
from erpnext.selling.doctype.quotation.test_quotation import make_quotation
from erpnext.selling.doctype.sales_order.test_sales_order import make_sales_order
from erpnext.setup.doctype.company.company import get_all_transactions_annual_history
company = "_Test Company"
txn_date = add_days(nowdate(), -30)
key = get_timestamp(txn_date)
before = get_all_transactions_annual_history(company).get(key, 0)
quotation = make_quotation(company=company, transaction_date=txn_date, do_not_submit=True)
self.addCleanup(frappe.delete_doc, "Quotation", quotation.name, force=True)
sales_order = make_sales_order(company=company, transaction_date=txn_date, do_not_submit=True)
self.addCleanup(frappe.delete_doc, "Sales Order", sales_order.name, force=True)
after = get_all_transactions_annual_history(company).get(key, 0)
self.assertEqual(after - before, 2)
def test_demo_data(self):
from erpnext.setup.demo import clear_demo_data, setup_demo_data

228
erpnext/setup/doctype/email_digest/email_digest.py
View File

@@ -9,6 +9,8 @@ import frappe.desk.notifications
from dateutil.relativedelta import relativedelta
from frappe import _
from frappe.core.doctype.user.user import STANDARD_USERS
from frappe.query_builder import Case
from frappe.query_builder.functions import Count, IfNull, Sum
from frappe.utils import (
add_to_date,
flt,
@@ -85,14 +87,11 @@ class EmailDigest(Document):
@frappe.whitelist()
def get_users(self):
"""get list of users"""
user_list = frappe.db.sql(
"""
select name, enabled from tabUser
where name not in ({})
and user_type != "Website User"
order by enabled desc, name asc""".format(", ".join(["%s"] * len(STANDARD_USERS))),
STANDARD_USERS,
as_dict=1,
user_list = frappe.get_all(
"User",
filters={"name": ["not in", STANDARD_USERS], "user_type": ["!=", "Website User"]},
fields=["name", "enabled"],
order_by="enabled desc, name asc",
)
if self.recipient_list:
@@ -107,13 +106,7 @@ class EmailDigest(Document):
@frappe.whitelist()
def send(self):
# send email only to enabled users
valid_users = [
p[0]
for p in frappe.db.sql(
"""select name from `tabUser`
where enabled=1"""
)
]
valid_users = frappe.get_all("User", filters={"enabled": 1}, pluck="name")
if self.recipients:
for row in self.recipients:
@@ -229,12 +222,24 @@ class EmailDigest(Document):
if not user_id:
user_id = frappe.session.user
todo_list = frappe.db.sql(
"""select *
from `tabToDo` where (owner=%s or assigned_by=%s) and status='Open'
order by field(priority, 'High', 'Medium', 'Low') asc, date asc limit 20""",
(user_id, user_id),
as_dict=True,
todo = frappe.qb.DocType("ToDo")
# matches MySQL field(priority,'High','Medium','Low'): unknown/empty/NULL -> 0 (sorts first)
priority_order = (
Case()
.when(todo.priority == "High", 1)
.when(todo.priority == "Medium", 2)
.when(todo.priority == "Low", 3)
.else_(0)
)
todo_list = (
frappe.qb.from_(todo)
.select(todo.star)
.where(((todo.owner == user_id) | (todo.assigned_by == user_id)) & (todo.status == "Open"))
.orderby(priority_order)
.orderby(IfNull(todo.date, "1000-01-01")) # NULL dates first, as MariaDB `date asc` did
.orderby(todo.name)
.limit(20)
.run(as_dict=True)
)
for t in todo_list:
@@ -247,10 +252,12 @@ class EmailDigest(Document):
if not user_id:
user_id = frappe.session.user
return frappe.db.sql(
"""select count(*) from `tabToDo`
where status='Open' and (owner=%s or assigned_by=%s)""",
(user_id, user_id),
todo = frappe.qb.DocType("ToDo")
return (
frappe.qb.from_(todo)
.select(Count("*"))
.where((todo.status == "Open") & ((todo.owner == user_id) | (todo.assigned_by == user_id)))
.run()
)[0][0]
def get_issue_list(self, user_id=None):
@@ -263,11 +270,12 @@ class EmailDigest(Document):
if not role_permissions.get("read"):
return None
issue_list = frappe.db.sql(
"""select *
from `tabIssue` where status in ("Replied","Open")
order by creation asc limit 10""",
as_dict=True,
issue_list = frappe.get_all(
"Issue",
filters={"status": ["in", ["Replied", "Open"]]},
fields=["*"],
order_by="creation asc",
limit=10,
)
for t in issue_list:
@@ -277,21 +285,19 @@ class EmailDigest(Document):
def get_issue_count(self):
"""Get count of Issue"""
return frappe.db.sql(
"""select count(*) from `tabIssue`
where status in ('Open','Replied') """
)[0][0]
return frappe.db.count("Issue", {"status": ["in", ["Open", "Replied"]]})
def get_project_list(self, user_id=None):
"""Get project list"""
if not user_id:
user_id = frappe.session.user
project_list = frappe.db.sql(
"""select *
from `tabProject` where status='Open' and project_type='External'
order by creation asc limit 10""",
as_dict=True,
project_list = frappe.get_all(
"Project",
filters={"status": "Open", "project_type": "External"},
fields=["*"],
order_by="creation asc",
limit=10,
)
for t in project_list:
@@ -301,10 +307,7 @@ class EmailDigest(Document):
def get_project_count(self):
"""Get count of Project"""
return frappe.db.sql(
"""select count(*) from `tabProject`
where status='Open' and project_type='External'"""
)[0][0]
return frappe.db.count("Project", {"status": "Open", "project_type": "External"})
def set_accounting_cards(self, context):
"""Create accounting cards if checked"""
@@ -485,12 +488,20 @@ class EmailDigest(Document):
def get_sales_orders_to_bill(self):
"""Get value not billed"""
value, count = frappe.db.sql(
"""select ifnull((sum(grand_total)) - (sum(grand_total*per_billed/100)),0),
count(*) from `tabSales Order`
where (transaction_date <= %(to_date)s) and billing_status != "Fully Billed" and company = %(company)s
and status not in ('Closed','Cancelled', 'Completed') """,
{"to_date": self.future_to_date, "company": self.company},
so = frappe.qb.DocType("Sales Order")
value, count = (
frappe.qb.from_(so)
.select(
IfNull(Sum(so.grand_total) - Sum(so.grand_total * so.per_billed / 100), 0),
Count("*"),
)
.where(
(so.transaction_date <= self.future_to_date)
& (so.billing_status != "Fully Billed")
& (so.company == self.company)
& so.status.notin(["Closed", "Cancelled", "Completed"])
)
.run()
)[0]
label = get_link_to_report(
@@ -511,12 +522,20 @@ class EmailDigest(Document):
def get_sales_orders_to_deliver(self):
"""Get value not delivered"""
value, count = frappe.db.sql(
"""select ifnull((sum(grand_total)) - (sum(grand_total*per_delivered/100)),0),
count(*) from `tabSales Order`
where (transaction_date <= %(to_date)s) and delivery_status != "Fully Delivered" and company = %(company)s
and status not in ('Closed','Cancelled', 'Completed') """,
{"to_date": self.future_to_date, "company": self.company},
so = frappe.qb.DocType("Sales Order")
value, count = (
frappe.qb.from_(so)
.select(
IfNull(Sum(so.grand_total) - Sum(so.grand_total * so.per_delivered / 100), 0),
Count("*"),
)
.where(
(so.transaction_date <= self.future_to_date)
& (so.delivery_status != "Fully Delivered")
& (so.company == self.company)
& so.status.notin(["Closed", "Cancelled", "Completed"])
)
.run()
)[0]
label = get_link_to_report(
@@ -537,12 +556,20 @@ class EmailDigest(Document):
def get_purchase_orders_to_receive(self):
"""Get value not received"""
value, count = frappe.db.sql(
"""select ifnull((sum(grand_total))-(sum(grand_total*per_received/100)),0),
count(*) from `tabPurchase Order`
where (transaction_date <= %(to_date)s) and per_received < 100 and company = %(company)s
and status not in ('Closed','Cancelled', 'Completed') """,
{"to_date": self.future_to_date, "company": self.company},
po = frappe.qb.DocType("Purchase Order")
value, count = (
frappe.qb.from_(po)
.select(
IfNull(Sum(po.grand_total) - Sum(po.grand_total * po.per_received / 100), 0),
Count("*"),
)
.where(
(po.transaction_date <= self.future_to_date)
& (po.per_received < 100)
& (po.company == self.company)
& po.status.notin(["Closed", "Cancelled", "Completed"])
)
.run()
)[0]
label = get_link_to_report(
@@ -563,12 +590,20 @@ class EmailDigest(Document):
def get_purchase_orders_to_bill(self):
"""Get purchase not billed"""
value, count = frappe.db.sql(
"""select ifnull((sum(grand_total)) - (sum(grand_total*per_billed/100)),0),
count(*) from `tabPurchase Order`
where (transaction_date <= %(to_date)s) and per_billed < 100 and company = %(company)s
and status not in ('Closed','Cancelled', 'Completed') """,
{"to_date": self.future_to_date, "company": self.company},
po = frappe.qb.DocType("Purchase Order")
value, count = (
frappe.qb.from_(po)
.select(
IfNull(Sum(po.grand_total) - Sum(po.grand_total * po.per_billed / 100), 0),
Count("*"),
)
.where(
(po.transaction_date <= self.future_to_date)
& (po.per_billed < 100)
& (po.company == self.company)
& po.status.notin(["Closed", "Cancelled", "Completed"])
)
.run()
)[0]
label = get_link_to_report(
@@ -707,13 +742,21 @@ class EmailDigest(Document):
return self.get_summary_of_pending_quotations("pending_quotations")
def get_summary_of_pending(self, doc_type, fieldname, getfield):
value, count, billed_value, delivered_value = frappe.db.sql(
"""select ifnull(sum(grand_total),0), count(*),
ifnull(sum(grand_total*per_billed/100),0), ifnull(sum(grand_total*{}/100),0) from `tab{}`
where (transaction_date <= %(to_date)s)
and status not in ('Closed','Cancelled', 'Completed')
and company = %(company)s """.format(getfield, doc_type),
{"to_date": self.future_to_date, "company": self.company},
doc = frappe.qb.DocType(doc_type)
value, count, billed_value, delivered_value = (
frappe.qb.from_(doc)
.select(
IfNull(Sum(doc.grand_total), 0),
Count("*"),
IfNull(Sum(doc.grand_total * doc.per_billed / 100), 0),
IfNull(Sum(doc.grand_total * doc[getfield] / 100), 0),
)
.where(
(doc.transaction_date <= self.future_to_date)
& doc.status.notin(["Closed", "Cancelled", "Completed"])
& (doc.company == self.company)
)
.run()
)[0]
return {
@@ -725,20 +768,27 @@ class EmailDigest(Document):
}
def get_summary_of_pending_quotations(self, fieldname):
value, count = frappe.db.sql(
"""select ifnull(sum(grand_total),0), count(*) from `tabQuotation`
where (transaction_date <= %(to_date)s)
and company = %(company)s
and status not in ('Ordered','Cancelled', 'Lost') """,
{"to_date": self.future_to_date, "company": self.company},
quotation = frappe.qb.DocType("Quotation")
value, count = (
frappe.qb.from_(quotation)
.select(IfNull(Sum(quotation.grand_total), 0), Count("*"))
.where(
(quotation.transaction_date <= self.future_to_date)
& (quotation.company == self.company)
& quotation.status.notin(["Ordered", "Cancelled", "Lost"])
)
.run()
)[0]
last_value = frappe.db.sql(
"""select ifnull(sum(grand_total),0) from `tabQuotation`
where (transaction_date <= %(to_date)s)
and company = %(company)s
and status not in ('Ordered','Cancelled', 'Lost') """,
{"to_date": self.past_to_date, "company": self.company},
last_value = (
frappe.qb.from_(quotation)
.select(IfNull(Sum(quotation.grand_total), 0))
.where(
(quotation.transaction_date <= self.past_to_date)
& (quotation.company == self.company)
& quotation.status.notin(["Ordered", "Cancelled", "Lost"])
)
.run()
)[0][0]
label = get_link_to_report(
@@ -898,10 +948,8 @@ class EmailDigest(Document):
def send():
now_date = now_datetime().date()
for ed in frappe.db.sql(
"""select name from `tabEmail Digest`
where enabled=1 and docstatus<2""",
as_list=1,
for ed in frappe.get_all(
"Email Digest", filters={"enabled": 1, "docstatus": ["<", 2]}, fields=["name"], as_list=True
):
ed_obj = frappe.get_doc("Email Digest", ed[0])
if now_date == ed_obj.get_next_sending():

39
erpnext/setup/doctype/email_digest/test_email_digest.py
View File

@@ -40,6 +40,45 @@ class TestEmailDigest(ERPNextTestSuite):
self.assertIn(po1.name, overdue_items)
self.assertNotIn(po2.name, overdue_items)
def test_get_todo_list_priority_and_date_ordering(self):
"""Original SQL ordered by `field(priority,'High','Medium','Low') asc, date asc`: MySQL
FIELD() returns 0 for empty/unknown priority (sorts FIRST under asc) and MariaDB sorts NULL
dates FIRST. The conversion preserves this: the priority CASE uses else_(0) (unknown/empty
priority sorts FIRST) and IfNull(date,'1000-01-01') keeps NULL dates FIRST, so the LIMIT-20
slice is identical on both engines. The two assertions below exercise both branches and would
fail if either sentinel were flipped to sort those rows last."""
user = "_test_todo_order@example.com"
if not frappe.db.exists("User", user):
frappe.get_doc(
{"doctype": "User", "email": user, "first_name": "Todo Order", "send_welcome_email": 0}
).insert(ignore_permissions=True)
def mk(desc, priority, date):
td = frappe.get_doc(
{
"doctype": "ToDo",
"description": desc,
"assigned_by": user,
"status": "Open",
"priority": "Medium",
}
).insert(ignore_permissions=True)
frappe.db.set_value("ToDo", td.name, {"priority": priority, "date": date}, update_modified=False)
return td.name
empty_pri = mk("empty-priority", "", "2020-01-01")
high_dated = mk("high-dated", "High", "2020-06-15")
high_nulldate = mk("high-nulldate", "High", None)
mk("low", "Low", "2020-03-01")
rows = frappe.new_doc("Email Digest").get_todo_list(user_id=user)
order = [r.name for r in rows]
# unknown/empty priority (FIELD()=0) must sort before High
self.assertLess(order.index(empty_pri), order.index(high_dated))
# within the High tier, a NULL date must sort before a real date (MariaDB NULLs-first)
self.assertLess(order.index(high_nulldate), order.index(high_dated))
def create_email_digest(**args):
args = frappe._dict(args)

8
erpnext/setup/doctype/transaction_deletion_record/transaction_deletion_record.py
View File

@@ -669,11 +669,9 @@ class TransactionDeletionRecord(Document):
self.enqueue_task(task="Delete Leads and Addresses")
return
frappe.db.sql(
"""delete from `tabBin` where warehouse in
(select name from tabWarehouse where company=%s)""",
self.company,
)
warehouses = frappe.get_all("Warehouse", filters={"company": self.company}, pluck="name")
if warehouses:
frappe.db.delete("Bin", {"warehouse": ["in", warehouses]})
self.db_set("delete_bin_data_status", "Completed")
self.enqueue_task(task="Delete Leads and Addresses")

6
erpnext/setup/install.py
View File

@@ -58,10 +58,8 @@ def set_single_defaults():
"Selling Settings",
"Stock Settings",
):
default_values = frappe.db.sql(
"""select fieldname, `default` from `tabDocField`
where parent=%s""",
dt,
default_values = frappe.get_all(
"DocField", filters={"parent": dt}, fields=["fieldname", "default"], as_list=True
)
if default_values:
try:

44
erpnext/templates/pages/material_request_info.py
View File

@@ -35,28 +35,30 @@ def get_context(context):
def get_more_items_info(items, material_request):
for item in items:
item.customer_provided = frappe.get_value("Item", item.item_code, "is_customer_provided_item")
item.work_orders = frappe.db.sql(
"""
select
wo.name, wo.status, wo_item.consumed_qty
from
`tabWork Order Item` wo_item, `tabWork Order` wo
where
wo_item.item_code=%s
and wo_item.consumed_qty=0
and wo_item.parent=wo.name
and wo.status not in ('Completed', 'Cancelled', 'Stopped')
order by
wo.name asc""",
item.item_code,
as_dict=1,
wo = frappe.qb.DocType("Work Order")
wo_item = frappe.qb.DocType("Work Order Item")
item.work_orders = (
frappe.qb.from_(wo_item)
.inner_join(wo)
.on(wo_item.parent == wo.name)
.select(wo.name, wo.status, wo_item.consumed_qty)
.where(
(wo_item.item_code == item.item_code)
& (wo_item.consumed_qty == 0)
& (wo.status.notin(["Completed", "Cancelled", "Stopped"]))
)
.orderby(wo.name)
.run(as_dict=1)
)
item.delivered_qty = flt(
frappe.db.sql(
"""select sum(transfer_qty)
from `tabStock Entry Detail` where material_request = %s
and item_code = %s and docstatus = 1""",
(material_request, item.item_code),
)[0][0]
frappe.get_all(
"Stock Entry Detail",
filters={
"material_request": material_request,
"item_code": item.item_code,
"docstatus": 1,
},
fields=[{"SUM": "transfer_qty", "as": "transfer_qty"}],
)[0].transfer_qty
)
return items

9
erpnext/templates/pages/partners.py
View File

@@ -8,10 +8,11 @@ page_title = "Partners"
def get_context(context):
partners = frappe.db.sql(
"""select * from `tabSales Partner`
where show_in_website=1 order by name asc""",
as_dict=True,
partners = frappe.get_all(
"Sales Partner",
filters={"show_in_website": 1},
fields=["*"],
order_by="name asc",
)
return {"partners": partners, "title": page_title}

28
erpnext/templates/pages/rfq.py
View File

@@ -31,10 +31,10 @@ def get_supplier():
def check_supplier_has_docname_access(supplier):
status = True
if frappe.form_dict.name not in frappe.db.sql_list(
"""select parent from `tabRequest for Quotation Supplier`
where supplier = %s""",
(supplier,),
if frappe.form_dict.name not in frappe.get_all(
"Request for Quotation Supplier",
filters={"supplier": supplier},
pluck="parent",
):
status = False
return status
@@ -59,15 +59,17 @@ def update_supplier_details(context):
def get_link_quotation(supplier, rfq):
quotation = frappe.db.sql(
""" select distinct `tabSupplier Quotation Item`.parent as name,
`tabSupplier Quotation`.status, `tabSupplier Quotation`.transaction_date from
`tabSupplier Quotation Item`, `tabSupplier Quotation` where `tabSupplier Quotation`.docstatus < 2 and
`tabSupplier Quotation Item`.request_for_quotation =%(name)s and
`tabSupplier Quotation Item`.parent = `tabSupplier Quotation`.name and
`tabSupplier Quotation`.supplier = %(supplier)s order by `tabSupplier Quotation`.creation desc""",
{"name": rfq, "supplier": supplier},
as_dict=1,
sqi = frappe.qb.DocType("Supplier Quotation Item")
sq = frappe.qb.DocType("Supplier Quotation")
quotation = (
frappe.qb.from_(sqi)
.inner_join(sq)
.on(sqi.parent == sq.name)
.select(sqi.parent.as_("name"), sq.status, sq.transaction_date, sq.creation)
.distinct()
.where((sq.docstatus < 2) & (sqi.request_for_quotation == rfq) & (sq.supplier == supplier))
.orderby(sq.creation, order=frappe.qb.desc)
.run(as_dict=1)
)
for data in quotation:

162
erpnext/templates/pages/test_material_request_info.py Normal file
View File

@@ -0,0 +1,162 @@
# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from frappe.utils import now, today
from erpnext.templates.pages.material_request_info import get_more_items_info
from erpnext.tests.utils import ERPNextTestSuite
class TestMaterialRequestInfo(ERPNextTestSuite):
"""Covers the two converted queries in ``get_more_items_info``: the query-builder
join that links Work Orders (via Work Order Item) to a Material Request's items,
and the ``SUM(transfer_qty)`` aggregate over submitted Stock Entry Detail rows that
feeds ``item.delivered_qty``.
"""
def setUp(self):
self.item_code = "_Test Item"
self.company = "_Test Company"
# A submitted Material Request that the page is rendered for.
self.material_request = self._make_material_request()
# A Work Order whose child Work Order Item references the same item.
# The converted query joins Work Order Item -> Work Order on this item.
self.work_order = self._make_linked_work_order(self.material_request.name)
def _make_material_request(self):
mr = frappe.new_doc("Material Request")
mr.material_request_type = "Manufacture"
mr.company = self.company
mr.append(
"items",
{
"item_code": self.item_code,
"qty": 5,
"uom": "_Test UOM",
"conversion_factor": 1,
"schedule_date": today(),
"warehouse": "_Test Warehouse - _TC",
},
)
mr.insert()
mr.submit()
return mr
def _make_linked_work_order(self, material_request):
"""Insert a Work Order + Work Order Item row directly.
We avoid the BOM-driven Work Order controller (heavy, needs a default
BOM and a submit cycle) because the query under test only reads the
columns we set here: ``name``, ``status`` and ``consumed_qty``.
"""
wo = frappe.new_doc("Work Order")
wo.production_item = self.item_code
wo.item_name = self.item_code
wo.qty = 5
wo.company = self.company
wo.fg_warehouse = "_Test Warehouse - _TC"
wo.wip_warehouse = "_Test Warehouse - _TC"
wo.planned_start_date = now()
wo.material_request = material_request
wo.status = "Not Started" # not in the excluded set
wo.bom_no = "TEST-BOM-MRI" # placeholder; never read by the query
wo.flags.ignore_validate = True
wo.flags.ignore_mandatory = True
wo.flags.name_set = True
wo.name = frappe.generate_hash("wo-mri", 12)
wo.db_insert()
wo_item = frappe.new_doc("Work Order Item")
wo_item.parent = wo.name
wo_item.parenttype = "Work Order"
wo_item.parentfield = "required_items"
wo_item.idx = 1
wo_item.item_code = self.item_code
wo_item.item_name = self.item_code
wo_item.required_qty = 5
wo_item.consumed_qty = 0 # the query filters on consumed_qty == 0
wo_item.flags.name_set = True
wo_item.name = frappe.generate_hash("woi-mri", 12)
wo_item.db_insert()
return wo
def _make_stock_entry_detail(self, transfer_qty, docstatus=1):
"""Insert a Stock Entry Detail row directly (parentless) for this MR + item.
The converted ``delivered_qty`` aggregate reads the child table alone
(``SUM(transfer_qty)`` filtered by material_request/item_code/docstatus), so a
parentless row with the docstatus set is enough to exercise it.
"""
sed = frappe.new_doc("Stock Entry Detail")
sed.parent = frappe.generate_hash("se-mri", 12)
sed.parenttype = "Stock Entry"
sed.parentfield = "items"
sed.idx = 1
sed.item_code = self.item_code
sed.item_name = self.item_code
sed.uom = "_Test UOM"
sed.stock_uom = "_Test UOM"
sed.conversion_factor = 1
sed.qty = transfer_qty
sed.transfer_qty = transfer_qty
sed.material_request = self.material_request.name
sed.docstatus = docstatus
sed.flags.name_set = True
sed.name = frappe.generate_hash("sed-mri", 12)
sed.db_insert()
return sed
def test_delivered_qty_sums_submitted_stock_entry_details(self):
# Two submitted rows for this MR + item must sum; a draft (docstatus 0) row must
# be excluded by the converted SUM(transfer_qty) aggregate.
self._make_stock_entry_detail(transfer_qty=3, docstatus=1)
self._make_stock_entry_detail(transfer_qty=4, docstatus=1)
self._make_stock_entry_detail(transfer_qty=99, docstatus=0) # draft -> ignored
items = [frappe._dict({"item_code": self.item_code})]
result = get_more_items_info(items, self.material_request.name)
self.assertEqual(result[0].delivered_qty, 7.0)
def test_delivered_qty_is_zero_when_no_stock_entry(self):
# No matching Stock Entry Detail -> SUM is NULL -> flt(None) must coerce to 0.0.
items = [frappe._dict({"item_code": self.item_code})]
result = get_more_items_info(items, self.material_request.name)
self.assertEqual(result[0].delivered_qty, 0.0)
def test_converted_query_returns_linked_work_order(self):
items = [frappe._dict({"item_code": self.item_code})]
result = get_more_items_info(items, self.material_request.name)
# Helper mutates and returns the same list of items.
self.assertEqual(len(result), 1)
item = result[0]
work_orders = item.work_orders
self.assertIsInstance(work_orders, list)
# Our seeded Work Order must be present with well-formed columns.
names = {wo.name for wo in work_orders}
self.assertIn(self.work_order.name, names)
seeded = next(wo for wo in work_orders if wo.name == self.work_order.name)
self.assertEqual(seeded.status, "Not Started")
self.assertEqual(seeded.consumed_qty, 0)
# Selected columns are exactly those projected by the query.
self.assertEqual(set(seeded.keys()), {"name", "status", "consumed_qty"})
def test_excluded_status_work_order_is_filtered_out(self):
# Flip the seeded Work Order to an excluded status; the query must drop it.
frappe.db.set_value("Work Order", self.work_order.name, "status", "Completed")
items = [frappe._dict({"item_code": self.item_code})]
result = get_more_items_info(items, self.material_request.name)
names = {wo.name for wo in result[0].work_orders}
self.assertNotIn(self.work_order.name, names)

56
erpnext/templates/pages/test_partners.py Normal file
View File

@@ -0,0 +1,56 @@
# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from erpnext.templates.pages.partners import get_context, page_title
from erpnext.tests.utils import ERPNextTestSuite
class TestPartnersPage(ERPNextTestSuite):
def _make_partner(self, name, show_in_website):
if not frappe.db.exists("Sales Partner", name):
frappe.get_doc(
{
"doctype": "Sales Partner",
"partner_name": name,
"territory": "_Test Territory",
"commission_rate": 5,
"show_in_website": show_in_website,
}
).insert(ignore_permissions=True)
return name
def test_get_context_lists_only_website_partners(self):
"""partners.py builds the /partners list via
frappe.get_all("Sales Partner", filters={"show_in_website": 1}, ...).
Seed one website-visible partner and one hidden control partner, then assert the
returned context contains the visible one and excludes the hidden one -- real
membership of the converted query's result, not a tautology."""
visible = self._make_partner("_Test Website Sales Partner", 1)
hidden = self._make_partner("_Test Hidden Sales Partner", 0)
result = get_context(frappe._dict())
# context shape: {"partners": [...], "title": page_title}
self.assertEqual(result["title"], page_title)
partner_names = [p.name for p in result["partners"]]
self.assertIn(
visible,
partner_names,
"website-flagged Sales Partner missing from /partners context",
)
self.assertNotIn(
hidden,
partner_names,
"Sales Partner with show_in_website=0 leaked into /partners context",
)
# every returned row really has show_in_website=1 (filter applied, not just appended)
for partner in result["partners"]:
self.assertEqual(
partner.show_in_website,
1,
f"Sales Partner {partner.name} returned despite show_in_website != 1",
)

81
erpnext/templates/pages/test_rfq.py Normal file
View File

@@ -0,0 +1,81 @@
# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from frappe.utils import formatdate
from erpnext.buying.doctype.request_for_quotation.mapper import make_supplier_quotation_from_rfq
from erpnext.buying.doctype.request_for_quotation.test_request_for_quotation import (
make_request_for_quotation,
)
from erpnext.templates.pages.rfq import get_link_quotation
from erpnext.tests.utils import ERPNextTestSuite
class TestRFQPage(ERPNextTestSuite):
"""Exercise the query-builder helper backing the RFQ supplier-portal page.
``get_link_quotation`` joins Supplier Quotation Item -> Supplier Quotation and
returns the linked quotations for a given (supplier, rfq) pair. The assertions
below seed a real RFQ + Supplier Quotation and verify the converted query
returns the expected row(s) on both engines.
"""
def test_get_link_quotation_returns_linked_quotation(self):
# Seed: RFQ for _Test Supplier / _Test Supplier 1, then a Supplier Quotation
# raised against it for _Test Supplier.
rfq = make_request_for_quotation()
supplier = rfq.suppliers[0].supplier # "_Test Supplier"
sq = make_supplier_quotation_from_rfq(rfq.name, for_supplier=supplier)
sq.submit()
# Sanity: the mapper stamps the child rows with the source RFQ, which is
# the column the converted query filters on.
self.assertEqual(sq.items[0].request_for_quotation, rfq.name)
# Seed a second Supplier Quotation Item under the SAME parent + RFQ so the
# Supplier Quotation Item -> Supplier Quotation join yields two identical rows.
# distinct() must collapse them back to one; without it len(result) would be 2.
dup = frappe.new_doc("Supplier Quotation Item")
dup.update(sq.items[0].as_dict())
dup.idx = sq.items[0].idx + 1
dup.flags.name_set = True
dup.name = frappe.generate_hash("sqi-rfq", 12)
dup.db_insert()
result = get_link_quotation(supplier, rfq.name)
# Real-state assertion: exactly the seeded quotation comes back, with the
# selected/derived columns the page template consumes.
self.assertIsNotNone(result)
# genuinely exercises distinct(): two SQ-item join rows collapse to one
self.assertEqual(len(result), 1)
row = result[0]
self.assertEqual(row.name, sq.name)
self.assertEqual(row.status, "Submitted")
# transaction_date is post-processed through formatdate() by the helper.
self.assertEqual(row.transaction_date, formatdate(sq.transaction_date))
self.assertEqual({r.name for r in result}, {sq.name})
def test_get_link_quotation_filters_by_supplier(self):
# The quotation belongs to supplier[0]; supplier[1] must see nothing for
# this RFQ. Guards the ``sq.supplier == supplier`` predicate.
rfq = make_request_for_quotation()
seeded_supplier = rfq.suppliers[0].supplier
other_supplier = rfq.suppliers[1].supplier
sq = make_supplier_quotation_from_rfq(rfq.name, for_supplier=seeded_supplier)
sq.submit()
self.assertIsNone(get_link_quotation(other_supplier, rfq.name))
def test_get_link_quotation_no_quotation(self):
# An RFQ with no Supplier Quotation raised yet returns None (helper coerces
# an empty list to None). Guards the ``request_for_quotation == rfq`` filter.
rfq = make_request_for_quotation()
supplier = rfq.suppliers[0].supplier
self.assertIsNone(get_link_quotation(supplier, rfq.name))

41
erpnext/templates/test_utils.py Normal file
View File

@@ -0,0 +1,41 @@
# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from erpnext.templates.utils import get_customer_from_contact_email
from erpnext.tests.utils import ERPNextTestSuite
class TestTemplateUtils(ERPNextTestSuite):
def test_contact_email_lookup_is_case_insensitive(self):
"""send_message resolves the Opportunity party by matching Contact.email_id with `==`.
Equality is case-SENSITIVE on Postgres (the query-builder ILIKE patch only rewrites LIKE),
while MariaDB's default collation is case-insensitive. A Contact email stored as
'Case.Test@Example.com' with a lowercase sender therefore matches on MariaDB but not on
Postgres -- so the Contact-Us form links a Lead instead of the Customer. The original raw SQL
used the same `c.email_id = %s`, so MariaDB output is unchanged: this is a Postgres-only break."""
customer_name = "_Test Contact Case Customer"
if not frappe.db.exists("Customer", customer_name):
frappe.get_doc(
{
"doctype": "Customer",
"customer_name": customer_name,
"customer_group": "_Test Customer Group",
"territory": "_Test Territory",
}
).insert(ignore_permissions=True)
frappe.get_doc(
{
"doctype": "Contact",
"first_name": "Case Test Contact",
"email_ids": [{"email_id": "Case.Test@Example.com", "is_primary": 1}],
"links": [{"link_doctype": "Customer", "link_name": customer_name}],
}
).insert(ignore_permissions=True)
# lowercase sender vs the stored mixed-case Contact email
matched = get_customer_from_contact_email("case.test@example.com")
self.assertTrue(matched, "Contact email lookup found no Customer for a case-differing sender")
self.assertEqual(matched[0][0], customer_name)

24
erpnext/templates/utils.py
View File

@@ -3,6 +3,7 @@
import frappe
from frappe.query_builder.functions import Lower
from frappe.rate_limiter import rate_limit
from frappe.utils import escape_html
@@ -24,13 +25,8 @@ def send_message(sender: str, message: str, subject: str = "Website Query"):
# Meant to silently fail instead of throwing error.
return
lead = customer = None
customer = frappe.db.sql(
"""select distinct dl.link_name from `tabDynamic Link` dl
left join `tabContact` c on dl.parent=c.name where dl.link_doctype='Customer'
and c.email_id = %s""",
sender,
)
lead = None
customer = get_customer_from_contact_email(sender)
if not customer:
lead = frappe.db.get_value("Lead", dict(email_id=sender))
@@ -68,3 +64,17 @@ def send_message(sender: str, message: str, subject: str = "Website Query"):
}
)
comm.insert(ignore_permissions=True)
def get_customer_from_contact_email(sender: str):
dl = frappe.qb.DocType("Dynamic Link")
contact = frappe.qb.DocType("Contact")
return (
frappe.qb.from_(dl)
.left_join(contact)
.on(dl.parent == contact.name)
.select(dl.link_name)
.distinct()
.where((dl.link_doctype == "Customer") & (Lower(contact.email_id) == sender.lower()))
.run()
)

18
erpnext/utilities/__init__.py
View File

@@ -10,11 +10,15 @@ from erpnext.utilities.activation import get_level
def update_doctypes():
for d in frappe.db.sql(
"""select df.parent, df.fieldname
from tabDocField df, tabDocType dt where df.fieldname
like "%description%" and df.parent = dt.name and dt.istable = 1""",
as_dict=1,
df = frappe.qb.DocType("DocField")
dt_table = frappe.qb.DocType("DocType")
for d in (
frappe.qb.from_(df)
.inner_join(dt_table)
.on(df.parent == dt_table.name)
.select(df.parent, df.fieldname)
.where(df.fieldname.like("%description%") & (dt_table.istable == 1))
.run(as_dict=1)
):
dt = frappe.get_doc("DocType", d.parent)
@@ -31,8 +35,8 @@ def get_site_info(site_info):
domain = None
if not company:
company = frappe.db.sql("select name from `tabCompany` order by creation asc")
company = company[0][0] if company else None
company = frappe.get_all("Company", order_by="creation asc", pluck="name")
company = company[0] if company else None
if company:
domain = frappe.get_cached_value("Company", cstr(company), "domain")

4
erpnext/utilities/activation.py
View File

@@ -55,7 +55,9 @@ def get_level(site_info):
sales_data.append({"Communication": communication_number})
# recent login
if frappe.db.sql("select name from tabUser where last_login > date_sub(now(), interval 2 day) limit 1"):
if frappe.db.exists(
"User", {"last_login": [">", frappe.utils.add_to_date(frappe.utils.now_datetime(), days=-2)]}
):
activation_level += 1
level = {"activation_level": activation_level, "sales_data": sales_data}

19
erpnext/utilities/naming.py
View File

@@ -20,10 +20,12 @@ def set_by_naming_series(doctype, fieldname, naming_series, hide_name_field=True
# set values for mandatory
try:
frappe.db.sql(
"""update `tab{doctype}` set naming_series={s} where
ifnull(naming_series, '')=''""".format(doctype=doctype, s="%s"),
get_default_naming_series(doctype),
dt = frappe.qb.DocType(doctype)
(
frappe.qb.update(dt)
.set(dt.naming_series, get_default_naming_series(doctype))
.where(dt.naming_series.isnull() | (dt.naming_series == ""))
.run()
)
except NamingSeriesNotSetError:
pass
@@ -42,7 +44,10 @@ def set_by_naming_series(doctype, fieldname, naming_series, hide_name_field=True
make_property_setter(doctype, fieldname, "reqd", 1, "Check", validate_fields_for_doctype=False)
# set values for mandatory
frappe.db.sql(
f"""update `tab{doctype}` set `{fieldname}`=`name` where
ifnull({fieldname}, '')=''"""
dt = frappe.qb.DocType(doctype)
(
frappe.qb.update(dt)
.set(dt[fieldname], dt.name)
.where(dt[fieldname].isnull() | (dt[fieldname] == ""))
.run()
)

70
erpnext/utilities/product.py
View File

@@ -2,7 +2,7 @@
# License: GNU General Public License v3. See license.txt
import frappe
from frappe.utils import cint, flt, fmt_money
from frappe.utils import cint, cstr, flt, fmt_money
from erpnext.accounts.doctype.pricing_rule.pricing_rule import get_pricing_rule_for_item
@@ -82,12 +82,15 @@ def get_price(item_code, price_list, customer_group, company, qty=1, party=None)
or ""
)
uom_conversion_factor = frappe.db.sql(
"""select C.conversion_factor
from `tabUOM Conversion Detail` C
inner join `tabItem` I on C.parent = I.name and C.uom = I.sales_uom
where I.name = %s""",
item_code,
uom_cd = frappe.qb.DocType("UOM Conversion Detail")
item_dt = frappe.qb.DocType("Item")
uom_conversion_factor = (
frappe.qb.from_(uom_cd)
.inner_join(item_dt)
.on((uom_cd.parent == item_dt.name) & (uom_cd.uom == item_dt.sales_uom))
.select(uom_cd.conversion_factor)
.where(item_dt.name == item_code)
.run()
)
uom_conversion_factor = uom_conversion_factor[0][0] if uom_conversion_factor else 1
@@ -119,46 +122,25 @@ def get_item_codes_by_attributes(attribute_filters, template_item_code=None):
if not attribute_values:
continue
wheres = []
query_values = []
for attribute_value in attribute_values:
wheres.append("( attribute = %s and attribute_value = %s )")
query_values += [attribute, attribute_value]
attribute_query = " or ".join(wheres)
iva = frappe.qb.DocType("Item Variant Attribute")
item_dt = frappe.qb.DocType("Item")
item_subquery = frappe.qb.from_(item_dt).select(item_dt.name)
if template_item_code:
variant_of_query = "AND t2.variant_of = %s"
query_values.append(template_item_code)
else:
variant_of_query = ""
item_subquery = item_subquery.where(item_dt.variant_of == template_item_code)
query = f"""
SELECT
t1.parent
FROM
`tabItem Variant Attribute` t1
WHERE
1 = 1
AND (
{attribute_query}
)
AND EXISTS (
SELECT
1
FROM
`tabItem` t2
WHERE
t2.name = t1.parent
{variant_of_query}
)
GROUP BY
t1.parent
ORDER BY
NULL
"""
item_codes = set([r[0] for r in frappe.db.sql(query, query_values)])
item_codes = set(
frappe.qb.from_(iva)
.select(iva.parent)
# attribute_value is a varchar column; cast values to str so postgres doesn't choke on
# `varchar = numeric` for numeric attributes (stored values are strings on both backends)
.where(
(iva.attribute == attribute) & (iva.attribute_value.isin([cstr(v) for v in attribute_values]))
)
.where(iva.parent.isin(item_subquery))
.groupby(iva.parent)
.run(pluck=True)
)
items.append(item_codes)
res = list(set.intersection(*items))

103
erpnext/utilities/report/test_product_util.py Normal file
View File

@@ -0,0 +1,103 @@
# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from frappe.utils import flt
from erpnext.tests.utils import ERPNextTestSuite
from erpnext.utilities.product import get_price
class TestProductUtil(ERPNextTestSuite):
"""Cover the ``get_price`` sales-UOM conversion query (UOM Conversion Detail join Item).
The converted query in ``erpnext.utilities.product.get_price`` resolves the
item's sales UOM conversion factor and scales ``formatted_price_sales_uom``
by it. We seed a sales UOM whose conversion factor differs from 1 so the join
produces an observable, non-trivial result.
"""
ITEM_CODE = "_Test Item"
PRICE_LIST = "_Test Selling Price List"
SALES_UOM = "_Test UOM 1"
SALES_UOM_FACTOR = 10.0
PRICE_LIST_RATE = 200.0
def setUp(self):
# _Test Item bootstrap ships uoms [_Test UOM (1.0), _Test UOM 1 (10.0)].
# Point its sales_uom at the 10x conversion so the joined query returns a
# factor != 1; assert against the bootstrapped conversion_factor to keep
# the check tied to real seeded state rather than a literal.
uom_cf = frappe.db.get_value(
"UOM Conversion Detail",
{"parent": self.ITEM_CODE, "parenttype": "Item", "uom": self.SALES_UOM},
"conversion_factor",
)
self.assertEqual(
flt(uom_cf),
self.SALES_UOM_FACTOR,
msg=f"Expected bootstrap UOM Conversion Detail {self.SALES_UOM} = {self.SALES_UOM_FACTOR}",
)
frappe.db.set_value("Item", self.ITEM_CODE, "sales_uom", self.SALES_UOM)
if not frappe.db.exists("Item Price", {"item_code": self.ITEM_CODE, "price_list": self.PRICE_LIST}):
frappe.get_doc(
{
"doctype": "Item Price",
"item_code": self.ITEM_CODE,
"price_list": self.PRICE_LIST,
"price_list_rate": self.PRICE_LIST_RATE,
}
).insert()
def test_sales_uom_conversion_factor_applied(self):
price = get_price(
item_code=self.ITEM_CODE,
price_list=self.PRICE_LIST,
customer_group="_Test Customer Group",
company="_Test Company",
)
self.assertIsNotNone(price, msg="get_price returned no price for seeded Item Price")
rate = flt(price["price_list_rate"])
self.assertTrue(rate, msg="seeded Item Price did not resolve a price_list_rate")
# The converted query (UOM Conversion Detail join Item on uom == sales_uom)
# multiplies the rate by the sales-UOM conversion factor for this field.
expected_sales_uom_price = frappe.utils.fmt_money(
rate * self.SALES_UOM_FACTOR, currency=price["currency"]
)
self.assertEqual(
price["formatted_price_sales_uom"],
expected_sales_uom_price,
msg="sales-UOM conversion factor (10x) was not applied by the converted join query",
)
# Guard against a degenerate factor of 1 silently passing: the sales-UOM
# price must differ from the plain formatted price.
self.assertNotEqual(
price["formatted_price_sales_uom"],
price["formatted_price"],
msg="formatted_price_sales_uom equals formatted_price; conversion factor was not picked up",
)
def test_factor_defaults_to_one_without_matching_sales_uom(self):
# When sales_uom has no matching UOM Conversion Detail row, the join
# returns nothing and the factor falls back to 1 (price unchanged).
frappe.db.set_value("Item", self.ITEM_CODE, "sales_uom", None)
price = get_price(
item_code=self.ITEM_CODE,
price_list=self.PRICE_LIST,
customer_group="_Test Customer Group",
company="_Test Company",
)
self.assertIsNotNone(price)
self.assertEqual(
price["formatted_price_sales_uom"],
price["formatted_price"],
msg="factor should default to 1 when no UOM Conversion Detail matches sales_uom",
)

38
erpnext/utilities/report/youtube_interactions/test_youtube_interactions.py Normal file
View File

@@ -0,0 +1,38 @@
# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from erpnext.tests.utils import ERPNextTestSuite
from erpnext.utilities.report.youtube_interactions.youtube_interactions import execute
class TestYoutubeInteractions(ERPNextTestSuite):
def test_zero_view_video_is_listed(self):
"""The original report filtered `WHERE view_count is not null`. The conversion keeps that exact
semantics with `.where(video.view_count.isnotnull())` (IS NOT NULL), NOT a `<> 0` test, so a
video with exactly 0 views is still reported. This guards against a regression to `!= 0`
(which would silently drop 0-view videos) and confirms the filter renders identically on both
engines."""
frappe.db.set_single_value("Video Settings", "enable_youtube_tracking", 1)
for title, views in (("_Test Zero Views Video", 0.0), ("_Test Ten Views Video", 10.0)):
if frappe.db.exists("Video", title):
frappe.delete_doc("Video", title, force=True)
frappe.get_doc(
{
"doctype": "Video",
"title": title,
"provider": "Vimeo", # skips the YouTube API call in validate()
"url": f"https://vimeo.com/{int(views)}",
"description": title,
"publish_date": "2024-01-15",
"view_count": views,
}
).insert()
_columns, data, *_rest = execute(frappe._dict({"from_date": "2024-01-01", "to_date": "2024-12-31"}))
titles = {row.get("title") for row in data}
self.assertIn("_Test Ten Views Video", titles)
# a real, freshly-synced video with 0 views must still be reported
self.assertIn("_Test Zero Views Video", titles)

29
erpnext/utilities/report/youtube_interactions/youtube_interactions.py
View File

@@ -30,18 +30,23 @@ def get_columns():
def get_data(filters):
return frappe.db.sql(
"""
SELECT
publish_date, title, provider, duration,
view_count, like_count, dislike_count, comment_count
FROM `tabVideo`
WHERE view_count is not null
and publish_date between %(from_date)s and %(to_date)s
ORDER BY view_count desc""",
filters,
as_dict=1,
)
video = frappe.qb.DocType("Video")
return (
frappe.qb.from_(video)
.select(
video.publish_date,
video.title,
video.provider,
video.duration,
video.view_count,
video.like_count,
video.dislike_count,
video.comment_count,
)
.where(video.view_count.isnotnull())
.where(video.publish_date[filters.get("from_date") : filters.get("to_date")])
.orderby(video.view_count, order=frappe.qb.desc)
).run(as_dict=True)
def get_chart_summary_data(data):

63
erpnext/utilities/test_utilities_init.py Normal file
View File

@@ -0,0 +1,63 @@
import frappe
from erpnext.tests.utils import ERPNextTestSuite
from erpnext.utilities import update_doctypes
class TestUtilitiesInit(ERPNextTestSuite):
def test_description_child_field_query_finds_core_child_fields(self):
"""The converted query in update_doctypes() joins DocField + DocType to find
description-bearing fields on child tables (istable=1). Reproduce the exact
query and assert known core child-doctype description fields are returned."""
df = frappe.qb.DocType("DocField")
dt_table = frappe.qb.DocType("DocType")
rows = (
frappe.qb.from_(df)
.inner_join(dt_table)
.on(df.parent == dt_table.name)
.select(df.parent, df.fieldname)
.where(df.fieldname.like("%description%") & (dt_table.istable == 1))
.run(as_dict=1)
)
# Map parent -> set of matched fieldnames for concrete assertions.
matched = {}
for d in rows:
matched.setdefault(d.parent, set()).add(d.fieldname)
# Known core child tables (istable=1) carrying a "description" field.
self.assertIn("Sales Invoice Item", matched)
self.assertIn("description", matched["Sales Invoice Item"])
self.assertIn("Purchase Invoice Item", matched)
self.assertIn("description", matched["Purchase Invoice Item"])
# Every returned fieldname must satisfy the LIKE predicate, and every
# returned parent must genuinely be a child table (istable=1) -- guards
# against the join/where being dropped during the qb conversion.
for d in rows:
self.assertIn("description", d.fieldname)
parents = {d.parent for d in rows}
istable_map = dict(
frappe.get_all(
"DocType",
filters={"name": ("in", list(parents))},
fields=["name", "istable"],
as_list=1,
)
)
for parent in parents:
self.assertEqual(
istable_map.get(parent),
1,
msg=f"{parent} returned by description-child query but is not a child table",
)
def test_update_doctypes_is_importable_and_callable(self):
"""update_doctypes() is the public entry point exercising the converted
query; ensure it imports and runs without error against real schema."""
self.assertTrue(callable(update_doctypes))
# Run it: it should only ever upgrade Text/Small Text description fields to
# Text Editor; core fixtures used above are already Text Editor, so this is
# effectively a no-op but must not raise.
update_doctypes()