From c67a57d9bd70c5bf11cfd35aedf47335ed2e64b8 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Tue, 8 Sep 2026 12:20:58 +0530 Subject: [PATCH] fix(stock): enforce serial permissions and case-insensitive numbers --- .../current/serial_batch_identity.md | 2 + erpnext/stock/doctype/batch/batch.py | 4 +- erpnext/stock/doctype/serial_no/serial_no.py | 2 +- erpnext/stock/serial_batch_identity.py | 57 ++++++++++---- .../test_serial_batch_identity_access.py | 54 +++++++++++++ .../test_serial_batch_identity_matching.py | 78 +++++++++++++++++++ erpnext/stock/utils.py | 25 +++--- 7 files changed, 195 insertions(+), 27 deletions(-) create mode 100644 erpnext/stock/tests/test_serial_batch_identity_access.py create mode 100644 erpnext/stock/tests/test_serial_batch_identity_matching.py diff --git a/erpnext/change_log/current/serial_batch_identity.md b/erpnext/change_log/current/serial_batch_identity.md index 688586fb16a..693d97a6b91 100644 --- a/erpnext/change_log/current/serial_batch_identity.md +++ b/erpnext/change_log/current/serial_batch_identity.md @@ -2,6 +2,7 @@ Serial No and Batch now have generated document IDs. Their physical numbers remain in `Serial No.serial_no` and `Batch.batch_id`. Different items can use the same physical number. Two records for the same item cannot share a number. +Physical-number matching and uniqueness ignore letter case on both databases. Stored labels keep their original spelling. Existing document IDs remain unchanged. Migration fills missing physical numbers from those IDs and replaces the global unique indexes. Historical transactions, bundles, and stock valuations retain their references. Item merges fail when they would introduce duplicate numbers. @@ -16,6 +17,7 @@ When a scanned number matches several items, select the item from the filtered I - Use returned document IDs in transaction fields and Serial and Batch Entry links, including legacy serial-number text lists. - Resolve physical numbers with `erpnext.stock.serial_batch_identity.resolve_serial_batch_numbers`. Supply `item_code` and lists named `serial_numbers` or `batch_numbers`. - The resolver returns `serial_nos` and `batch_nos`, containing document IDs in input order. Set `create` only for authorized creation. +- Resolving inward transaction input requires Serial No create permission. Transaction write permission alone does not permit serial creation. - For the bundle editor, use `serial_number` and `batch_number` for physical input. Use `serial_no` and `batch_no` for existing links. - CSV uploads continue accepting physical numbers. CSV downloads and standard prints show physical numbers. - A barcode scan can match several records. Interactive callers pass `allow_multiple=true` and select a returned candidate before updating a transaction. diff --git a/erpnext/stock/doctype/batch/batch.py b/erpnext/stock/doctype/batch/batch.py index ef9b1707d87..52d784df533 100644 --- a/erpnext/stock/doctype/batch/batch.py +++ b/erpnext/stock/doctype/batch/batch.py @@ -27,7 +27,7 @@ def get_name_from_hash(item_code=None): temp = None while not temp: temp = frappe.generate_hash()[:7].upper() - if frappe.db.exists("Batch", {"batch_id": temp, **({"item": item_code} if item_code else {})}): + if SerialBatchIdentity("Batch").exists(temp, item_code): temp = None return temp @@ -138,7 +138,7 @@ class Batch(Document): self.batch_id = get_name_from_hash(self.item) # User might have manually created a batch with next number - if frappe.db.exists("Batch", {"item": self.item, "batch_id": self.batch_id}): + if SerialBatchIdentity("Batch").exists(self.batch_id, self.item): self.batch_id = None def onload(self): diff --git a/erpnext/stock/doctype/serial_no/serial_no.py b/erpnext/stock/doctype/serial_no/serial_no.py index c9c363df983..7b9d7b0e7e4 100644 --- a/erpnext/stock/doctype/serial_no/serial_no.py +++ b/erpnext/stock/doctype/serial_no/serial_no.py @@ -132,7 +132,7 @@ def get_available_serial_nos(serial_no_series, qty, item_code=None) -> list[str] def get_new_serial_number(series, item_code=None): sr_no = make_autoname(series, "Serial No") - if frappe.db.exists("Serial No", {"serial_no": sr_no, **({"item_code": item_code} if item_code else {})}): + if SerialBatchIdentity("Serial No").exists(sr_no, item_code): sr_no = get_new_serial_number(series, item_code) return sr_no diff --git a/erpnext/stock/serial_batch_identity.py b/erpnext/stock/serial_batch_identity.py index a694bec94c7..4692f85c3c7 100644 --- a/erpnext/stock/serial_batch_identity.py +++ b/erpnext/stock/serial_batch_identity.py @@ -3,6 +3,7 @@ import frappe from frappe import _ from frappe.model.naming import make_autoname +from frappe.query_builder.functions import Lower from frappe.utils import cstr, now @@ -24,19 +25,14 @@ class SerialBatchIdentity: if not isinstance(item_code, str) or not item_code or any(not number for number in numbers): frappe.throw(_("Item and physical number are required")) - filters = {self.item_field: item_code, self.number_field: ("in", numbers)} - records = frappe.get_all(self.doctype, filters=filters, fields=["name", self.number_field]) + records = self.get_query(numbers, item_code).run(as_dict=True) ids = {row[self.number_field]: row.name for row in records} missing = [] for number in dict.fromkeys(numbers): if number in ids: continue # Use the database comparison rules, including its collation, for exact lookups. - name = ( - frappe.db.get_value(self.doctype, {self.item_field: item_code, self.number_field: number}) - if records - else None - ) + name = self.exists(number, item_code) if records else None if not name and create: missing.append(number) continue @@ -47,6 +43,27 @@ class SerialBatchIdentity: ids.update(self.create_many(item_code, missing, defaults)) return [ids[number] for number in numbers] + def get_query(self, numbers, item_code=None, *, fields=None, filters=None, ignore_permissions=True): + table = frappe.qb.DocType(self.doctype) + filters = {**(filters or {}), **({self.item_field: item_code} if item_code else {})} + return frappe.qb.get_query( + self.doctype, + fields=fields or ["name", self.number_field], + filters=filters, + ignore_permissions=ignore_permissions, + ).where( + self.number_key(table[self.number_field]).isin([self.number_key(number) for number in numbers]) + ) + + def number_key(self, value): + # Preserve MariaDB's collation. PostgreSQL needs an explicit case-insensitive comparison. + return Lower(value) if frappe.db.db_type == "postgres" else value + + def exists(self, number, item_code=None, *, exclude=None): + filters = {"name": ("!=", exclude)} if exclude else None + rows = self.get_query([number], item_code, fields=["name"], filters=filters).limit(1).run() + return rows[0][0] if rows else None + def create_many(self, item_code, numbers, defaults=None): if self.doctype == "Batch": return {number: self.create_batch(item_code, number, defaults) for number in numbers} @@ -110,10 +127,7 @@ class SerialBatchIdentity: def validate(self, doc): number = cstr(doc.get(self.number_field)).strip() doc.set(self.number_field, number) - filters = {self.item_field: doc.get(self.item_field), self.number_field: number} - if doc.name: - filters["name"] = ("!=", doc.name) - if number and frappe.db.exists(self.doctype, filters): + if number and self.exists(number, doc.get(self.item_field), exclude=doc.name): frappe.throw( _("{0} {1} already exists for Item {2}").format( self.doctype, number, doc.get(self.item_field) @@ -129,7 +143,19 @@ class SerialBatchIdentity: def sync_constraint(self): self.backfill_numbers() - frappe.db.add_unique(self.doctype, [self.item_field, self.number_field]) + if frappe.db.db_type == "postgres": + # The leading number expression also indexes scans without an item filter. + # add_unique only accepts column names, so expression indexes need explicit DDL. + frappe.db.sql_ddl( + { + "Serial No": 'CREATE UNIQUE INDEX IF NOT EXISTS "serial_no_number_item_ci" ' + 'ON "tabSerial No" (lower("serial_no"), "item_code")', + "Batch": 'CREATE UNIQUE INDEX IF NOT EXISTS "batch_number_item_ci" ' + 'ON "tabBatch" (lower("batch_id"), "item")', + }[self.doctype] + ) + else: + frappe.db.add_unique(self.doctype, [self.item_field, self.number_field]) @frappe.whitelist(methods=["POST"]) @@ -191,8 +217,8 @@ def resolve_transaction_serial_numbers(parent: dict | str, row: dict | str, numb parent.doctype, "write", doc=parent.name if not parent.__islocal else None, throw=True ) frappe.has_permission("Item", "read", doc=row.item_code or row.rm_item_code, throw=True) - frappe.has_permission("Serial No", "read", throw=True) create = parent.doctype in SUPPORTED_VOUCHER_TYPES and get_type_of_transaction(parent, row) == "Inward" + frappe.has_permission("Serial No", "create" if create else "read", throw=True) return SerialBatchIdentity("Serial No").resolve( row.item_code or row.rm_item_code, frappe.parse_json(numbers), @@ -236,7 +262,10 @@ def validate_item_merge(old, new): conflict = ( frappe.qb.from_(source) .join(target) - .on(source[identity.number_field] == target[identity.number_field]) + .on( + identity.number_key(source[identity.number_field]) + == identity.number_key(target[identity.number_field]) + ) .select(source[identity.number_field]) .where((source[identity.item_field] == old) & (target[identity.item_field] == new)) .limit(1) diff --git a/erpnext/stock/tests/test_serial_batch_identity_access.py b/erpnext/stock/tests/test_serial_batch_identity_access.py new file mode 100644 index 00000000000..2f5794ad627 --- /dev/null +++ b/erpnext/stock/tests/test_serial_batch_identity_access.py @@ -0,0 +1,54 @@ +import frappe + +from erpnext.stock.doctype.item.test_item import make_item +from erpnext.stock.serial_batch_identity import resolve_transaction_serial_numbers +from erpnext.tests.utils import ERPNextTestSuite + + +class TestSerialBatchIdentityAccess(ERPNextTestSuite): + def make_stock_user(self): + return frappe.get_doc( + { + "doctype": "User", + "email": f"serial-access-{frappe.generate_hash()}@example.test", + "first_name": "Serial Access Test", + "send_welcome_email": 0, + "roles": [{"role": "Stock User"}], + } + ).insert() + + def test_inward_resolution_requires_serial_create_permission(self): + item = make_item(properties={"has_serial_no": 1}) + user = self.make_stock_user() + parent = {"doctype": "Purchase Receipt", "__islocal": 1, "company": "_Test Company"} + row = {"item_code": item.name, "qty": 1} + with self.set_user(user.name): + self.assertTrue(frappe.has_permission("Purchase Receipt", "write")) + self.assertTrue(frappe.has_permission("Serial No", "read")) + self.assertFalse(frappe.has_permission("Serial No", "create")) + with self.assertRaises(frappe.PermissionError): + resolve_transaction_serial_numbers(parent, row, ["UNAUTHORIZED-SERIAL"]) + self.assertFalse(frappe.db.exists("Serial No", {"item_code": item.name})) + + def test_readers_can_resolve_existing_outward_serials(self): + item = make_item(properties={"has_serial_no": 1}) + parent = {"doctype": "Purchase Receipt", "__islocal": 1, "company": "_Test Company"} + row = {"item_code": item.name, "qty": 1} + names = resolve_transaction_serial_numbers(parent, row, ["EXISTING-SERIAL"]) + parent["is_return"] = 1 + row["qty"] = -1 + user = self.make_stock_user() + with self.set_user(user.name): + self.assertFalse(frappe.has_permission("Serial No", "create")) + self.assertEqual(resolve_transaction_serial_numbers(parent, row, ["EXISTING-SERIAL"]), names) + + def test_authorized_inward_resolution_creates_serials(self): + item = make_item(properties={"has_serial_no": 1}) + parent = {"doctype": "Purchase Receipt", "__islocal": 1, "company": "_Test Company"} + row = {"item_code": item.name, "qty": 1} + names = resolve_transaction_serial_numbers(parent, row, ["AUTHORIZED-SERIAL"]) + serial = frappe.get_doc("Serial No", names[0]) + self.assertEqual(serial.item_code, item.name) + self.assertEqual(serial.serial_no, "AUTHORIZED-SERIAL") + self.assertEqual(serial.company, parent["company"]) + self.assertEqual(resolve_transaction_serial_numbers(parent, row, ["AUTHORIZED-SERIAL"]), names) diff --git a/erpnext/stock/tests/test_serial_batch_identity_matching.py b/erpnext/stock/tests/test_serial_batch_identity_matching.py new file mode 100644 index 00000000000..7b0830b570a --- /dev/null +++ b/erpnext/stock/tests/test_serial_batch_identity_matching.py @@ -0,0 +1,78 @@ +import frappe + +from erpnext.stock.doctype.item.test_item import make_item +from erpnext.stock.serial_batch_identity import SerialBatchIdentity, validate_item_merge +from erpnext.stock.utils import scan_barcode +from erpnext.tests.utils import ERPNextTestSuite + + +class TestSerialBatchIdentityMatching(ERPNextTestSuite): + def make_number(self, doctype, number): + identity = SerialBatchIdentity(doctype) + item = make_item( + properties={"has_serial_no": int(doctype == "Serial No"), "has_batch_no": int(doctype == "Batch")} + ) + name = identity.resolve(item.name, [number], create=True, defaults={"company": "_Test Company"})[0] + return identity, item, name + + def test_case_insensitive_resolution_preserves_physical_label(self): + for doctype in ("Serial No", "Batch"): + identity, item, name = self.make_number(doctype, "Mixed-Lot-001") + for number in ("mixed-lot-001", "MIXED-LOT-001"): + self.assertEqual(identity.resolve(item.name, [number]), [name]) + self.assertEqual(identity.resolve(item.name, [number], create=True), [name]) + self.assertEqual(identity.labels([name]), {name: "Mixed-Lot-001"}) + + def test_case_insensitive_scan_keeps_items_separate(self): + for doctype, field in (("Serial No", "serial_no"), ("Batch", "batch_no")): + number = "SCAN-" + frappe.generate_hash().upper() + _, item_a, name_a = self.make_number(doctype, number) + _, item_b, name_b = self.make_number(doctype, number.lower()) + matches = scan_barcode(number.swapcase(), allow_multiple=True)["candidates"] + self.assertEqual({match[field] for match in matches}, {name_a, name_b}) + for item, name in ((item_a, name_a), (item_b, name_b)): + self.assertEqual(scan_barcode(number.swapcase(), {"item_code": item.name})[field], name) + + def test_controller_rejects_case_variant_for_same_item(self): + for doctype in ("Serial No", "Batch"): + identity, _, name = self.make_number(doctype, "Mixed-Lot-001") + duplicate = frappe.copy_doc(frappe.get_doc(doctype, name)) + duplicate.set(identity.number_field, "MIXED-LOT-001") + with self.assertRaises(frappe.DuplicateEntryError): + duplicate.insert() + + def test_database_rejects_case_variant_without_controller_validation(self): + for doctype in ("Serial No", "Batch"): + identity, _, name = self.make_number(doctype, "Mixed-Lot-001") + duplicate = frappe.get_doc(doctype, name) + duplicate.name = frappe.generate_hash() + duplicate.set(identity.number_field, "MIXED-LOT-001") + frappe.db.savepoint("case_variant") + try: + with self.assertRaises((frappe.DuplicateEntryError, frappe.UniqueValidationError)): + duplicate.db_insert() + finally: + frappe.db.rollback(save_point="case_variant") + + def test_item_merge_rejects_case_variants(self): + for doctype in ("Serial No", "Batch"): + _, item_a, _ = self.make_number(doctype, "Mixed-Lot-001") + _, item_b, _ = self.make_number(doctype, "MIXED-LOT-001") + with self.assertRaises(frappe.ValidationError): + validate_item_merge(item_a.name, item_b.name) + + def test_generated_numbers_skip_case_variant_collisions(self): + from erpnext.stock.doctype.serial_no.serial_no import get_new_serial_number + + for doctype in ("Serial No", "Batch"): + prefix = "CASE-" + frappe.generate_hash().upper() + "-" + _, item, _ = self.make_number(doctype, prefix.lower() + "00001") + series = prefix + ".#####" + if doctype == "Serial No": + number = get_new_serial_number(series, item.name) + else: + item.create_new_batch = 1 + item.batch_number_series = series + item.save() + number = frappe.get_doc({"doctype": "Batch", "item": item.name}).insert().batch_id + self.assertEqual(number, prefix + "00002") diff --git a/erpnext/stock/utils.py b/erpnext/stock/utils.py index 16e09ad8ce7..ba107c79090 100644 --- a/erpnext/stock/utils.py +++ b/erpnext/stock/utils.py @@ -598,6 +598,8 @@ def check_pending_reposting(posting_date: str, company: str | None = None, throw @frappe.whitelist(methods=["GET", "POST"]) def scan_barcode(search_value: str, ctx: dict | str | None = None, allow_multiple: bool = False) -> dict: + from erpnext.stock.serial_batch_identity import SerialBatchIdentity + ctx = frappe._dict(frappe.parse_json(ctx) or {}) if ctx.item_code and not isinstance(ctx.item_code, str): frappe.throw(_("Item Code must be a string")) @@ -612,23 +614,26 @@ def scan_barcode(search_value: str, ctx: dict | str | None = None, allow_multipl if barcode: candidates.append(barcode) - for doctype, number_field, item_field, fields in ( + for doctype, fields in ( ( "Serial No", - "serial_no", - "item_code", ["name as serial_no", "serial_no as serial_number", "item_code", "batch_no"], ), - ("Batch", "batch_id", "item", ["name as batch_no", "batch_id as batch_number", "item as item_code"]), + ("Batch", ["name as batch_no", "batch_id as batch_number", "item as item_code"]), ): if not frappe.has_permission(doctype, "read"): continue - filters = {number_field: search_value} - if ctx.item_code: - filters[item_field] = ctx.item_code - if doctype == "Batch": - filters["disabled"] = 0 - candidates.extend(frappe.get_list(doctype, filters=filters, fields=fields, limit_page_length=0)) + candidates.extend( + SerialBatchIdentity(doctype) + .get_query( + [search_value], + ctx.item_code, + fields=fields, + filters={"disabled": 0} if doctype == "Batch" else None, + ignore_permissions=False, + ) + .run(as_dict=True) + ) for candidate in candidates: _update_item_info(candidate, ctx)