From 497a0abb07ca3c8351b8e161b3ab7fc4e5653cb6 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Fri, 19 Jun 2026 22:10:54 +0530 Subject: [PATCH 1/8] refactor(stock): build item-group filter via qb in item_dashboard Replace the raw EXISTS subquery (items within an item-group subtree) with frappe.qb (item_group.isin(subquery)). Same result on MariaDB; valid under Postgres' stricter SQL. Co-Authored-By: Claude Opus 4.8 (1M context) --- erpnext/stock/dashboard/item_dashboard.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/erpnext/stock/dashboard/item_dashboard.py b/erpnext/stock/dashboard/item_dashboard.py index 01615cb549d..2acf8e3bbf3 100644 --- a/erpnext/stock/dashboard/item_dashboard.py +++ b/erpnext/stock/dashboard/item_dashboard.py @@ -24,13 +24,19 @@ def get_data( filters.append(["warehouse", "=", warehouse]) if item_group: lft, rgt = frappe.db.get_value("Item Group", item_group, ["lft", "rgt"]) - items = frappe.db.sql_list( - """ - select i.name from `tabItem` i - where exists(select name from `tabItem Group` - where name=i.item_group and lft >=%s and rgt<=%s) - """, - (lft, rgt), + item = frappe.qb.DocType("Item") + item_group_dt = frappe.qb.DocType("Item Group") + items = ( + frappe.qb.from_(item) + .select(item.name) + .where( + item.item_group.isin( + frappe.qb.from_(item_group_dt) + .select(item_group_dt.name) + .where((item_group_dt.lft >= lft) & (item_group_dt.rgt <= rgt)) + ) + ) + .run(pluck="name") ) filters.append(["item_code", "in", items]) try: From f03a81b943dacafa2ece81e1f23b8eb0075f2f34 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Fri, 19 Jun 2026 22:11:05 +0530 Subject: [PATCH 2/8] refactor(stock): use get_all for warehouse subtree in capacity dashboard Replace the raw lft/rgt SELECT with frappe.get_all(pluck="name"). Same result on MariaDB; valid under Postgres. Co-Authored-By: Claude Opus 4.8 (1M context) --- erpnext/stock/dashboard/warehouse_capacity_dashboard.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/erpnext/stock/dashboard/warehouse_capacity_dashboard.py b/erpnext/stock/dashboard/warehouse_capacity_dashboard.py index 1d550010315..24944047ef2 100644 --- a/erpnext/stock/dashboard/warehouse_capacity_dashboard.py +++ b/erpnext/stock/dashboard/warehouse_capacity_dashboard.py @@ -40,12 +40,8 @@ def get_filters(item_code=None, warehouse=None, parent_warehouse=None, company=N filters.append(["company", "=", company]) if parent_warehouse: lft, rgt = frappe.db.get_value("Warehouse", parent_warehouse, ["lft", "rgt"]) - warehouses = frappe.db.sql_list( - """ - select name from `tabWarehouse` - where lft >=%s and rgt<=%s - """, - (lft, rgt), + warehouses = frappe.get_all( + "Warehouse", filters={"lft": [">=", lft], "rgt": ["<=", rgt]}, pluck="name" ) filters.append(["warehouse", "in", warehouses]) return filters From b0d9208561b9c0a647afa26de9297ab6c6ea9980 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Fri, 19 Jun 2026 22:11:11 +0530 Subject: [PATCH 3/8] refactor(stock): convert get_alternative_items UNION to ORM Replace the raw UNION of forward/two-way alternative-item matches with two frappe.get_all calls, order-preserving dedup (dict.fromkeys) and Python pagination. Each leg is bounded to start+page_len rows so the per-keystroke search round trip stays small (the original bounded with LIMIT/OFFSET); ItemAlternative forbids duplicate (item_code, alternative_item_code) pairs, so each leg is internally distinct and that bound is exact. Same result on MariaDB; valid under Postgres. Tests: both-direction dedup, txt filtering, pagination, bounded-and-exact page-walk reconstruction, and case-insensitive (ILIKE-on-Postgres) matching. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../item_alternative/item_alternative.py | 30 ++- .../item_alternative/test_item_alternative.py | 189 +++++++++++++++++- 2 files changed, 209 insertions(+), 10 deletions(-) diff --git a/erpnext/stock/doctype/item_alternative/item_alternative.py b/erpnext/stock/doctype/item_alternative/item_alternative.py index b2923b4a828..6b9bb210fa9 100644 --- a/erpnext/stock/doctype/item_alternative/item_alternative.py +++ b/erpnext/stock/doctype/item_alternative/item_alternative.py @@ -7,6 +7,7 @@ from typing import Any import frappe from frappe import _ from frappe.model.document import Document +from frappe.utils import cint class ItemAlternative(Document): @@ -86,13 +87,24 @@ class ItemAlternative(Document): @frappe.whitelist() @frappe.validate_and_sanitize_search_inputs def get_alternative_items(doctype: Any, txt: str, searchfield: Any, start: int, page_len: int, filters: dict): - return frappe.db.sql( - f""" (select alternative_item_code from `tabItem Alternative` - where item_code = %(item_code)s and alternative_item_code like %(txt)s) - union - (select item_code from `tabItem Alternative` - where alternative_item_code = %(item_code)s and item_code like %(txt)s - and two_way = 1) limit {page_len} offset {start} - """, - {"item_code": filters.get("item_code"), "txt": "%" + txt + "%"}, + item_code = filters.get("item_code") + search = f"%{txt}%" + # each leg has distinct values (validate_duplicate), so start+page_len rows per leg suffice + limit = cint(start) + cint(page_len) + + alternatives = frappe.get_all( + "Item Alternative", + filters={"item_code": item_code, "alternative_item_code": ["like", search]}, + pluck="alternative_item_code", + limit=limit, ) + alternatives += frappe.get_all( + "Item Alternative", + filters={"alternative_item_code": item_code, "item_code": ["like", search], "two_way": 1}, + pluck="item_code", + limit=limit, + ) + + # union (dedupe, preserve order) + paginate + unique_items = list(dict.fromkeys(alternatives)) + return [[item] for item in unique_items[start : start + page_len]] diff --git a/erpnext/stock/doctype/item_alternative/test_item_alternative.py b/erpnext/stock/doctype/item_alternative/test_item_alternative.py index 2be54c82036..c11ced53837 100644 --- a/erpnext/stock/doctype/item_alternative/test_item_alternative.py +++ b/erpnext/stock/doctype/item_alternative/test_item_alternative.py @@ -2,7 +2,7 @@ # See license.txt import frappe -from frappe.utils import flt +from frappe.utils import flt, random_string from erpnext.controllers.subcontracting_controller import make_rm_stock_entry from erpnext.controllers.tests.test_subcontracting_controller import ( @@ -14,6 +14,7 @@ from erpnext.manufacturing.doctype.production_plan.test_production_plan import m from erpnext.manufacturing.doctype.work_order.mapper import make_stock_entry from erpnext.manufacturing.doctype.work_order.test_work_order import make_wo_order_test_record from erpnext.stock.doctype.item.test_item import create_item +from erpnext.stock.doctype.item_alternative.item_alternative import get_alternative_items from erpnext.stock.doctype.stock_reconciliation.stock_reconciliation import ( EmptyStockReconciliationItemsError, ) @@ -166,6 +167,192 @@ class TestItemAlternative(ERPNextTestSuite): self.assertEqual(status, True) ste1.submit() + def test_get_alternative_items_both_directions_and_dedup(self): + """get_alternative_items must return forward alternatives, reverse-only + two_way alternatives, exclude one-way reverse rows, and dedupe an item + that matches in both the forward and reverse legs of the old UNION.""" + suffix = random_string(8) + base = f"_Test IA Base {suffix}" + alt_fwd = f"_Test IA Fwd {suffix}" # forward only (two_way=0) + alt_both = f"_Test IA Both {suffix}" # forward (two_way=1) + alt_rev = f"_Test IA Rev {suffix}" # reverse via two_way=1 + alt_norev = f"_Test IA NoRev {suffix}" # reverse but two_way=0 -> excluded + dup = f"_Test IA Dup {suffix}" # forward AND reverse -> must dedupe + + for item_code in (base, alt_fwd, alt_both, alt_rev, alt_norev, dup): + create_item(item_code) + item = frappe.get_doc("Item", item_code) + if not item.allow_alternative_item: + item.allow_alternative_item = 1 + item.save() + + # forward rows: item_code = base + make_item_alternative(base, alt_fwd, two_way=0) + make_item_alternative(base, alt_both, two_way=1) + make_item_alternative(base, dup, two_way=1) + + # reverse rows: alternative_item_code = base + make_item_alternative(alt_rev, base, two_way=1) + make_item_alternative(alt_norev, base, two_way=0) + make_item_alternative(dup, base, two_way=1) + + # txt = the shared suffix so the LIKE matches every alternate but not `base` + results = get_alternative_items("Item", suffix, "name", 0, 20, {"item_code": base}) + + # structure: list of single-element lists + self.assertTrue(all(isinstance(row, list) and len(row) == 1 for row in results)) + + returned = [row[0] for row in results] + + # forward alternatives (both one-way and two_way) are returned + self.assertIn(alt_fwd, returned) + self.assertIn(alt_both, returned) + + # reverse alternative is only returned when the row is two_way + self.assertIn(alt_rev, returned) + self.assertNotIn(alt_norev, returned) + + # `base` itself is never an alternative of itself + self.assertNotIn(base, returned) + + # an item matching both legs of the old UNION is deduped to a single row + self.assertIn(dup, returned) + self.assertEqual(returned.count(dup), 1) + + def test_get_alternative_items_respects_txt_filter(self): + """The txt LIKE filter must actually narrow the result set so a + non-matching alternate is excluded (guards against a broken WHERE).""" + suffix = random_string(8) + base = f"_Test IA Filter Base {suffix}" + matching = f"_Test IA Match {suffix}" + other = f"_Test IA Other {suffix}" + + for item_code in (base, matching, other): + create_item(item_code) + item = frappe.get_doc("Item", item_code) + if not item.allow_alternative_item: + item.allow_alternative_item = 1 + item.save() + + make_item_alternative(base, matching, two_way=0) + make_item_alternative(base, other, two_way=0) + + # search only for the `Match` alternate + results = get_alternative_items("Item", f"Match {suffix}", "name", 0, 20, {"item_code": base}) + returned = [row[0] for row in results] + + self.assertIn(matching, returned) + self.assertNotIn(other, returned) + + def test_get_alternative_items_case_insensitive_match(self): + """The txt match must stay case-insensitive on BOTH engines: MariaDB LIKE is + case-insensitive by default, and frappe compiles the `like` filter to ILIKE on + Postgres. A case-shifted search must still find an alternate whose stored code + differs in case — this guards against the conversion degrading to a case-sensitive + match (plain LIKE / ==) that would silently return nothing on Postgres.""" + suffix = random_string(8) + base = f"_Test IA Case Base {suffix}" + # distinctive mixed-case token in the stored alternate's code + alt = f"_Test IA CaseToken AbCdE {suffix}" + + for item_code in (base, alt): + create_item(item_code) + item = frappe.get_doc("Item", item_code) + if not item.allow_alternative_item: + item.allow_alternative_item = 1 + item.save() + + make_item_alternative(base, alt, two_way=0) + + # search the LOWERCASED token ("abcde") against the stored "AbCdE" + results = get_alternative_items( + "Item", f"casetoken abcde {suffix}", "name", 0, 20, {"item_code": base} + ) + returned = [row[0] for row in results] + + self.assertIn(alt, returned) + + def test_get_alternative_items_pagination(self): + """start/page_len must slice the deduped, order-preserving result.""" + suffix = random_string(8) + base = f"_Test IA Page Base {suffix}" + alts = [f"_Test IA Page {i} {suffix}" for i in range(3)] + + create_item(base) + base_item = frappe.get_doc("Item", base) + if not base_item.allow_alternative_item: + base_item.allow_alternative_item = 1 + base_item.save() + + for alt in alts: + create_item(alt) + alt_item = frappe.get_doc("Item", alt) + if not alt_item.allow_alternative_item: + alt_item.allow_alternative_item = 1 + alt_item.save() + make_item_alternative(base, alt, two_way=0) + + full = [row[0] for row in get_alternative_items("Item", suffix, "name", 0, 20, {"item_code": base})] + self.assertEqual(len(full), 3) + + page = [row[0] for row in get_alternative_items("Item", suffix, "name", 1, 1, {"item_code": base})] + self.assertEqual(len(page), 1) + self.assertEqual(page[0], full[1]) + + def test_get_alternative_items_pagination_is_bounded_and_exact(self): + """Each get_all is bounded to start+page_len rows, so the DB round trip stays small + instead of fetching every alternative per keystroke. Walking the result in small pages + must still reconstruct the complete deduped set — including an alternate that appears in + BOTH legs (forward + reverse two_way) — with no item dropped or duplicated by the bound.""" + suffix = random_string(8) + base = f"_Test IA Bound Base {suffix}" + forwards = [f"_Test IA Bound Fwd {i} {suffix}" for i in range(3)] + reverses = [f"_Test IA Bound Rev {i} {suffix}" for i in range(3)] + dup = f"_Test IA Bound Dup {suffix}" # forward AND reverse two_way -> deduped across legs + + for item_code in [base, dup, *forwards, *reverses]: + create_item(item_code) + item = frappe.get_doc("Item", item_code) + if not item.allow_alternative_item: + item.allow_alternative_item = 1 + item.save() + + for fwd in forwards: + make_item_alternative(base, fwd, two_way=0) + make_item_alternative(base, dup, two_way=1) # dup via the forward leg + for rev in reverses: + make_item_alternative(rev, base, two_way=1) + make_item_alternative(dup, base, two_way=1) # dup also via the reverse leg + + full = [row[0] for row in get_alternative_items("Item", suffix, "name", 0, 50, {"item_code": base})] + # 3 forward + 3 reverse + the single deduped dup = 7 distinct + self.assertEqual(len(full), 7) + self.assertEqual(full.count(dup), 1) + + # walk in pages of 2; bounded fetches must yield exactly the same set, once each + collected = [] + for start in range(0, 8, 2): + collected += [ + row[0] for row in get_alternative_items("Item", suffix, "name", start, 2, {"item_code": base}) + ] + + self.assertEqual(len(collected), len(set(collected))) # no duplicates introduced by paging + self.assertEqual(set(collected), set(full)) # nothing dropped by the per-leg limit + self.assertEqual(collected.count(dup), 1) # the cross-leg dup survives exactly once + + +def make_item_alternative(item_code, alternative_item_code, two_way=0): + doc = frappe.get_doc( + { + "doctype": "Item Alternative", + "item_code": item_code, + "alternative_item_code": alternative_item_code, + "two_way": two_way, + } + ) + doc.insert() + return doc + def make_items(): items = [ From d955122c8818342bee1956b4abde3ed2ac99d99f Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Fri, 19 Jun 2026 22:11:18 +0530 Subject: [PATCH 4/8] refactor(stock): convert Item Price bulk update to qb Replace the raw UPDATE ... modified=NOW() in update_item_price with frappe.qb.update (now() for modified). Same result on MariaDB; valid under Postgres. Tests: currency/buying/selling/modified propagation and price-list scoping. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../stock/doctype/price_list/price_list.py | 16 ++-- .../doctype/price_list/test_price_list.py | 81 ++++++++++++++++++- 2 files changed, 90 insertions(+), 7 deletions(-) diff --git a/erpnext/stock/doctype/price_list/price_list.py b/erpnext/stock/doctype/price_list/price_list.py index 9ee05a3f6c7..ab6d1287444 100644 --- a/erpnext/stock/doctype/price_list/price_list.py +++ b/erpnext/stock/doctype/price_list/price_list.py @@ -5,7 +5,7 @@ import frappe from frappe import _, throw from frappe.model.document import Document -from frappe.utils import cint +from frappe.utils import cint, now class PriceList(Document): @@ -47,11 +47,15 @@ class PriceList(Document): frappe.set_value("Buying Settings", "Buying Settings", "buying_price_list", self.name) def update_item_price(self): - frappe.db.sql( - """update `tabItem Price` set currency=%s, - buying=%s, selling=%s, modified=NOW() where price_list=%s""", - (self.currency, cint(self.buying), cint(self.selling), self.name), - ) + item_price = frappe.qb.DocType("Item Price") + ( + frappe.qb.update(item_price) + .set(item_price.currency, self.currency) + .set(item_price.buying, cint(self.buying)) + .set(item_price.selling, cint(self.selling)) + .set(item_price.modified, now()) + .where(item_price.price_list == self.name) + ).run() def on_trash(self): self.delete_price_list_details_key() diff --git a/erpnext/stock/doctype/price_list/test_price_list.py b/erpnext/stock/doctype/price_list/test_price_list.py index 4c6b653d003..bbe56f2533f 100644 --- a/erpnext/stock/doctype/price_list/test_price_list.py +++ b/erpnext/stock/doctype/price_list/test_price_list.py @@ -1,4 +1,83 @@ -# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors # License: GNU General Public License v3. See license.txt +import frappe +from frappe.utils import random_string + +from erpnext.tests.utils import ERPNextTestSuite + + +class TestPriceList(ERPNextTestSuite): + def make_price_list(self, currency="INR", buying=1, selling=1): + price_list = frappe.get_doc( + { + "doctype": "Price List", + "price_list_name": "_Test PL " + random_string(10), + "enabled": 1, + "currency": currency, + "buying": buying, + "selling": selling, + } + ).insert() + return price_list + + def make_item_price(self, price_list, item_code="_Test Item", rate=100): + return frappe.get_doc( + { + "doctype": "Item Price", + "item_code": item_code, + "price_list": price_list, + "price_list_rate": rate, + } + ).insert() + + def test_update_item_price_propagates_currency_and_flags(self): + # Price List starts in INR, applicable for both buying and selling. + price_list = self.make_price_list(currency="INR", buying=1, selling=1) + + ip1 = self.make_item_price(price_list.name, item_code="_Test Item", rate=100) + ip2 = self.make_item_price(price_list.name, item_code="_Test Item 2", rate=250) + + # Sanity: Item Price rows inherited the Price List's initial state. + for ip in (ip1, ip2): + row = frappe.db.get_value("Item Price", ip.name, ["currency", "buying", "selling"], as_dict=True) + self.assertEqual(row.currency, "INR") + self.assertEqual(row.buying, 1) + self.assertEqual(row.selling, 1) + + # Change the Price List's currency and flip the buying flag off. + # on_update -> update_item_price() should bulk-UPDATE every Item Price + # linked to this Price List. + price_list.currency = "USD" + price_list.buying = 0 + price_list.selling = 1 + price_list.save() + + for ip in (ip1, ip2): + row = frappe.db.get_value("Item Price", ip.name, ["currency", "buying", "selling"], as_dict=True) + self.assertEqual(row.currency, "USD") + self.assertEqual(row.buying, 0) + self.assertEqual(row.selling, 1) + + def test_update_item_price_scoped_to_own_price_list(self): + # Two independent Price Lists; updating one must not touch the other's + # Item Price rows (the WHERE price_list == self.name clause). + pl_a = self.make_price_list(currency="INR", buying=1, selling=1) + pl_b = self.make_price_list(currency="INR", buying=1, selling=1) + + ip_a = self.make_item_price(pl_a.name, item_code="_Test Item", rate=100) + ip_b = self.make_item_price(pl_b.name, item_code="_Test Item", rate=100) + + pl_a.currency = "USD" + pl_a.buying = 0 + pl_a.save() + + row_a = frappe.db.get_value("Item Price", ip_a.name, ["currency", "buying"], as_dict=True) + self.assertEqual(row_a.currency, "USD") + self.assertEqual(row_a.buying, 0) + + # pl_b was untouched, so its Item Price must keep the original values. + row_b = frappe.db.get_value("Item Price", ip_b.name, ["currency", "buying"], as_dict=True) + self.assertEqual(row_b.currency, "INR") + self.assertEqual(row_b.buying, 1) From d61720c3e2e4c91cab558b6aba571824d0b8d536 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Fri, 19 Jun 2026 22:11:25 +0530 Subject: [PATCH 5/8] refactor(stock): convert Job Card reference update to qb Replace the raw f-string UPDATE (name + production_item match) that links a Quality Inspection back to its Job Card with frappe.qb.update. Same result on MariaDB; valid under Postgres. Tests: the Job Card reference update and its production_item scoping. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../quality_inspection/quality_inspection.py | 15 +++-- .../test_quality_inspection.py | 59 +++++++++++++++++++ 2 files changed, 66 insertions(+), 8 deletions(-) diff --git a/erpnext/stock/doctype/quality_inspection/quality_inspection.py b/erpnext/stock/doctype/quality_inspection/quality_inspection.py index dac3ca21038..7a99553aea0 100644 --- a/erpnext/stock/doctype/quality_inspection/quality_inspection.py +++ b/erpnext/stock/doctype/quality_inspection/quality_inspection.py @@ -215,14 +215,13 @@ class QualityInspection(Document): if self.reference_type == "Job Card": if self.reference_name: - frappe.db.sql( - f""" - UPDATE `tab{self.reference_type}` - SET quality_inspection = %s, modified = %s - WHERE name = %s and production_item = %s - """, - (quality_inspection, self.modified, self.reference_name, self.item_code), - ) + ref = frappe.qb.DocType(self.reference_type) + ( + frappe.qb.update(ref) + .set(ref.quality_inspection, quality_inspection) + .set(ref.modified, self.modified) + .where((ref.name == self.reference_name) & (ref.production_item == self.item_code)) + ).run() else: doctype = self.reference_type + " Item" diff --git a/erpnext/stock/doctype/quality_inspection/test_quality_inspection.py b/erpnext/stock/doctype/quality_inspection/test_quality_inspection.py index 34b140515f2..9445e5da94f 100644 --- a/erpnext/stock/doctype/quality_inspection/test_quality_inspection.py +++ b/erpnext/stock/doctype/quality_inspection/test_quality_inspection.py @@ -279,6 +279,65 @@ class TestQualityInspection(ERPNextTestSuite): se.delete() + def test_qi_updates_job_card_reference(self): + """Submitting a QI with reference_type 'Job Card' writes its name onto the + Job Card's quality_inspection field (the Job Card branch of + QualityInspection.update_qc_reference).""" + create_item("_Test Item") + + # Job Card whose production_item matches the QI item_code -> must be updated. + matching_jc = make_minimal_job_card(production_item="_Test Item") + # Job Card with a different production_item -> the production_item filter must + # keep it untouched. + other_item = create_item("_Test Item for QC " + frappe.utils.random_string(6)).name + non_matching_jc = make_minimal_job_card(production_item=other_item) + + qa = create_quality_inspection( + item_code="_Test Item", + reference_type="Job Card", + reference_name=matching_jc, + ) + + # The converted UPDATE wrote the QI name onto the matching Job Card. + self.assertEqual( + frappe.db.get_value("Job Card", matching_jc, "quality_inspection"), + qa.name, + ) + # The production_item filter excluded the Job Card with a different item. + self.assertFalse(frappe.db.get_value("Job Card", non_matching_jc, "quality_inspection")) + + def test_qi_job_card_reference_respects_production_item(self): + """A QI referencing a Job Card by name but whose item_code does not match the + Job Card's production_item must NOT update that Job Card.""" + create_item("_Test Item") + mismatch_item = create_item("_Test Item Mismatch QC " + frappe.utils.random_string(6)).name + + # Job Card produces a different item than the QI's item_code. + jc = make_minimal_job_card(production_item=mismatch_item) + + create_quality_inspection( + item_code="_Test Item", + reference_type="Job Card", + reference_name=jc, + ) + + # name matches but production_item != item_code, so the row is left untouched. + self.assertFalse(frappe.db.get_value("Job Card", jc, "quality_inspection")) + + +def make_minimal_job_card(production_item): + """db_insert a minimal submitted Job Card row carrying only the columns the + converted UPDATE reads (name, production_item, quality_inspection, modified).""" + jc = frappe.new_doc("Job Card") + jc.name = "_T-Job Card-" + frappe.utils.random_string(10) + jc.flags.name_set = True + jc.production_item = production_item + jc.company = "_Test Company" + jc.for_quantity = 1 + jc.docstatus = 1 + jc.db_insert() + return jc.name + def create_quality_inspection(**args): args = frappe._dict(args) From 3e801a206726fc5f2cea8e1e88ffb40081f8b071 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Fri, 19 Jun 2026 22:11:31 +0530 Subject: [PATCH 6/8] refactor(stock): convert serial_no lookups to ORM Replace the on_trash Stock Ledger Entry serial-match SELECT and the update_maintenance_status expiry SELECT with frappe.get_all (or_filters for the amc/warranty expiry OR). Same result on MariaDB; valid under Postgres. Tests: maintenance-status expiry transition, the not-in exclusion (with a negative-control candidate), and NULL-status handling. Co-Authored-By: Claude Opus 4.8 (1M context) --- erpnext/stock/doctype/serial_no/serial_no.py | 20 +-- .../stock/doctype/serial_no/test_serial_no.py | 125 ++++++++++++++++++ 2 files changed, 135 insertions(+), 10 deletions(-) diff --git a/erpnext/stock/doctype/serial_no/serial_no.py b/erpnext/stock/doctype/serial_no/serial_no.py index 195ac39c758..6ea89b2db8b 100644 --- a/erpnext/stock/doctype/serial_no/serial_no.py +++ b/erpnext/stock/doctype/serial_no/serial_no.py @@ -101,11 +101,10 @@ class SerialNo(StockController): self.maintenance_status = "Under Warranty" def on_trash(self): - sl_entries = frappe.db.sql( - """select serial_no from `tabStock Ledger Entry` - where serial_no like %s and item_code=%s and is_cancelled=0""", - ("%%%s%%" % self.name, self.item_code), - as_dict=True, + sl_entries = frappe.get_all( + "Stock Ledger Entry", + filters={"serial_no": ["like", f"%{self.name}%"], "item_code": self.item_code, "is_cancelled": 0}, + fields=["serial_no"], ) # Find the exact match @@ -172,13 +171,14 @@ def clean_serial_no_string(serial_no: str) -> str: def update_maintenance_status(): - serial_nos = frappe.db.sql( - """select name from `tabSerial No` where (amc_expiry_date<%s or - warranty_expiry_date<%s) and maintenance_status not in ('Out of Warranty', 'Out of AMC')""", - (nowdate(), nowdate()), + serial_nos = frappe.get_all( + "Serial No", + filters={"maintenance_status": ["not in", ["Out of Warranty", "Out of AMC"]]}, + or_filters=[["amc_expiry_date", "<", nowdate()], ["warranty_expiry_date", "<", nowdate()]], + pluck="name", ) for serial_no in serial_nos: - doc = frappe.get_doc("Serial No", serial_no[0]) + doc = frappe.get_doc("Serial No", serial_no) doc.set_maintenance_status() frappe.db.set_value("Serial No", doc.name, "maintenance_status", doc.maintenance_status) diff --git a/erpnext/stock/doctype/serial_no/test_serial_no.py b/erpnext/stock/doctype/serial_no/test_serial_no.py index 3ad635eb65b..0e93ff38b8b 100644 --- a/erpnext/stock/doctype/serial_no/test_serial_no.py +++ b/erpnext/stock/doctype/serial_no/test_serial_no.py @@ -7,6 +7,7 @@ import frappe from frappe import _dict +from frappe.utils import add_days, nowdate, random_string from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note from erpnext.stock.doctype.item.test_item import make_item @@ -16,6 +17,7 @@ from erpnext.stock.doctype.serial_and_batch_bundle.test_serial_and_batch_bundle get_serial_nos_from_bundle, ) from erpnext.stock.doctype.serial_no.serial_no import * +from erpnext.stock.doctype.serial_no.serial_no import update_maintenance_status from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry from erpnext.stock.doctype.stock_entry.test_stock_entry import make_serialized_item from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse @@ -331,6 +333,129 @@ class TestSerialNo(ERPNextTestSuite): self.assertEqual(non_expired_serials, []) + def test_update_maintenance_status_expires_past_warranty(self): + """update_maintenance_status() must pick up the past-warranty Serial No via or_filters and flip it Out of Warranty.""" + item_code = "_Test Serialized Item" + past_date = add_days(nowdate(), -10) + future_date = add_days(nowdate(), 10) + + # Serial No whose warranty has lapsed; force maintenance_status back to a + # value that passes the `not in [Out of Warranty, Out of AMC]` filter so the + # cron job is the thing that actually transitions it. + expired_sr = frappe.get_doc( + { + "doctype": "Serial No", + "item_code": item_code, + "serial_no": "_TCWARREXP" + random_string(6), + "company": "_Test Company", + "warranty_expiry_date": past_date, + } + ).insert() + frappe.db.set_value("Serial No", expired_sr.name, "maintenance_status", "Under Warranty") + self.assertEqual( + frappe.db.get_value("Serial No", expired_sr.name, "maintenance_status"), "Under Warranty" + ) + + # Serial No whose warranty is still valid; it must stay Under Warranty. + active_sr = frappe.get_doc( + { + "doctype": "Serial No", + "item_code": item_code, + "serial_no": "_TCWARRACT" + random_string(6), + "company": "_Test Company", + "warranty_expiry_date": future_date, + } + ).insert() + self.assertEqual( + frappe.db.get_value("Serial No", active_sr.name, "maintenance_status"), "Under Warranty" + ) + + update_maintenance_status() + + # The lapsed Serial No was selected by the or_filters and re-evaluated. + self.assertEqual( + frappe.db.get_value("Serial No", expired_sr.name, "maintenance_status"), "Out of Warranty" + ) + # The in-warranty Serial No keeps its correct Under Warranty status. + self.assertEqual( + frappe.db.get_value("Serial No", active_sr.name, "maintenance_status"), "Under Warranty" + ) + + def test_update_maintenance_status_excludes_out_of_amc(self): + """The `not in [Out of Warranty, Out of AMC]` filter must skip rows already pinned to + those statuses, even when they match the expiry or_filters, while rows in any other + status ARE re-evaluated. The contrast makes the `not in` clause load-bearing.""" + item_code = "_Test Serialized Item" + past_date = add_days(nowdate(), -10) + future_date = add_days(nowdate(), 10) + + # Excluded row: matches or_filters (amc lapsed) AND is pinned to "Out of AMC", so the + # `not in` filter must skip it. Its warranty is in the FUTURE, so if the filter were + # broken and the row were re-evaluated, set_maintenance_status() would flip it to + # "Under Warranty" (the last cascade branch to match). Staying "Out of AMC" proves it. + excluded_sr = frappe.get_doc( + { + "doctype": "Serial No", + "item_code": item_code, + "serial_no": "_TCAMCEXCL" + random_string(6), + "company": "_Test Company", + "amc_expiry_date": past_date, + "warranty_expiry_date": future_date, + } + ).insert() + frappe.db.set_value("Serial No", excluded_sr.name, "maintenance_status", "Out of AMC") + + # Negative control: same lapsed amc date, but a status NOT in the excluded list, so it + # must be picked up and re-evaluated to "Out of AMC". This proves update_maintenance_status() + # actually processes candidates — i.e. the exclusion above is meaningful, not a no-op. + candidate_sr = frappe.get_doc( + { + "doctype": "Serial No", + "item_code": item_code, + "serial_no": "_TCAMCCAND" + random_string(6), + "company": "_Test Company", + "amc_expiry_date": past_date, + } + ).insert() + frappe.db.set_value("Serial No", candidate_sr.name, "maintenance_status", "Under AMC") + + update_maintenance_status() + + # Excluded by the `not in` filter -> pinned status left untouched. + self.assertEqual( + frappe.db.get_value("Serial No", excluded_sr.name, "maintenance_status"), "Out of AMC" + ) + # Not excluded -> re-evaluated; lapsed amc -> "Out of AMC". + self.assertEqual( + frappe.db.get_value("Serial No", candidate_sr.name, "maintenance_status"), "Out of AMC" + ) + + def test_update_maintenance_status_includes_null_status(self): + """Converting the raw `maintenance_status not in (...)` to a get_all filter changes NULL + handling: frappe wraps the clause as `ifnull(maintenance_status, '') not in (...)`, so a + NULL-status row that matches the expiry or_filters is now re-evaluated (consistently on + MariaDB and Postgres). Pin that contract.""" + item_code = "_Test Serialized Item" + past_date = add_days(nowdate(), -10) + + null_sr = frappe.get_doc( + { + "doctype": "Serial No", + "item_code": item_code, + "serial_no": "_TCAMCNULL" + random_string(6), + "company": "_Test Company", + "amc_expiry_date": past_date, + } + ).insert() + # Force a NULL maintenance_status while a lapsed amc date keeps the row in or_filters. + frappe.db.set_value("Serial No", null_sr.name, "maintenance_status", None) + self.assertIsNone(frappe.db.get_value("Serial No", null_sr.name, "maintenance_status")) + + update_maintenance_status() + + # Picked up (NULL -> '' -> not in the excluded list) and re-evaluated: lapsed amc -> "Out of AMC". + self.assertEqual(frappe.db.get_value("Serial No", null_sr.name, "maintenance_status"), "Out of AMC") + def get_auto_serial_nos(kwargs): from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import ( From f9732efb2329c359d54acb4f4c43acddb77778c5 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Fri, 19 Jun 2026 22:11:38 +0530 Subject: [PATCH 7/8] refactor(stock): convert stock settings checks to ORM/qb Replace the get_all(limit=1, pluck) existence checks with frappe.db.exists and the no-own-valuation-method Stock Ledger Entry EXISTS with a frappe.qb subquery (null-or-empty valuation_method preserved). Same result on MariaDB; valid under Postgres. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../doctype/stock_settings/stock_settings.py | 23 +++++++++++++------ 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/erpnext/stock/doctype/stock_settings/stock_settings.py b/erpnext/stock/doctype/stock_settings/stock_settings.py index 03bf5b0cca7..b1676d8df79 100644 --- a/erpnext/stock/doctype/stock_settings/stock_settings.py +++ b/erpnext/stock/doctype/stock_settings/stock_settings.py @@ -120,7 +120,7 @@ class StockSettings(Document): if not doc_before_save: return - if not frappe.get_all("Serial and Batch Bundle", filters={"docstatus": 1}, limit=1, pluck="name"): + if not frappe.db.exists("Serial and Batch Bundle", {"docstatus": 1}): return if doc_before_save.do_not_use_batchwise_valuation and not self.do_not_use_batchwise_valuation: @@ -142,7 +142,7 @@ class StockSettings(Document): doc_before_save.enable_serial_and_batch_no_for_item and not self.enable_serial_and_batch_no_for_item ): - if frappe.get_all("Serial and Batch Bundle", filters={"docstatus": 1}, limit=1, pluck="name"): + if frappe.db.exists("Serial and Batch Bundle", {"docstatus": 1}): frappe.throw( _( "Cannot disable Serial and Batch No for Item, as there are existing records for serial / batch." @@ -170,11 +170,20 @@ class StockSettings(Document): 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='')) limit 1 - """ + 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.item_code.isin( + frappe.qb.from_(item) + .select(item.name) + .where(item.valuation_method.isnull() | (item.valuation_method == "")) + ) + ) + .limit(1) + .run() ) if sle: From 48bbf664227e4be2fee941e08b8ca432aac51b3c Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Fri, 19 Jun 2026 22:11:44 +0530 Subject: [PATCH 8/8] refactor(stock): convert packing slip item search to qb.get_query Replace the raw SELECT with get_match_cond in item_details with frappe.qb.get_query(ignore_permissions=False) plus a Delivery Note Item subquery; get_query applies the permission match conditions. Same result on MariaDB; valid under Postgres. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../doctype/packing_slip/packing_slip.py | 27 +++++++++++++------ 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/erpnext/stock/doctype/packing_slip/packing_slip.py b/erpnext/stock/doctype/packing_slip/packing_slip.py index 9544a7a3766..b9423027ffd 100644 --- a/erpnext/stock/doctype/packing_slip/packing_slip.py +++ b/erpnext/stock/doctype/packing_slip/packing_slip.py @@ -207,13 +207,24 @@ class PackingSlip(StatusUpdater): @frappe.whitelist() @frappe.validate_and_sanitize_search_inputs def item_details(doctype: str, txt: str, searchfield: str, start: int, page_len: int, filters: dict): - from erpnext.controllers.queries import get_match_cond + item = frappe.qb.DocType("Item") + dn_item = frappe.qb.DocType("Delivery Note Item") + delivery_note = (filters or {}).get("delivery_note") - return frappe.db.sql( - """select name, item_name, description from `tabItem` - where name in ( select item_code FROM `tabDelivery Note Item` - where parent= {}) - and {} like "{}" {} - limit {} offset {} """.format("%s", searchfield, "%s", get_match_cond(doctype), "%s", "%s"), - ((filters or {}).get("delivery_note"), "%%%s%%" % txt, page_len, start), + query = frappe.qb.get_query( + "Item", + fields=["name", "item_name", "description"], + ignore_permissions=False, + ) + + return ( + query.where( + item.name.isin( + frappe.qb.from_(dn_item).select(dn_item.item_code).where(dn_item.parent == delivery_note) + ) + & item[searchfield].like(f"%{txt}%") + ) + .limit(page_len) + .offset(start) + .run() )