fix: do not allow to make changes in SABB after submit

(cherry picked from commit e36426e235)

# Conflicts:
#	erpnext/stock/doctype/stock_reconciliation/stock_reconciliation.py
This commit is contained in:
Rohit Waghchaure
2026-06-08 13:07:26 +05:30
committed by Mergify
parent ecd3a19912
commit a03e3bfe9f
3 changed files with 69 additions and 160 deletions

View File

@@ -2229,7 +2229,16 @@ def get_type_of_transaction(parent_doc, child_row):
def update_serial_batch_no_ledgers(bundle, entries, child_row, parent_doc, warehouse=None) -> object:
frappe.has_permission("Serial and Batch Bundle", "write", throw=True)
doc = frappe.get_doc("Serial and Batch Bundle", bundle)
if doc.docstatus == 1:
doc.throw_error_message(
_("Serial and Batch Bundle {0} is submitted and its entries cannot be modified.").format(
frappe.bold(bundle)
)
)
doc.voucher_detail_no = child_row.name
doc.posting_datetime = combine_datetime(parent_doc.get("posting_date"), parent_doc.get("posting_time"))

View File

@@ -737,6 +737,60 @@ class TestSerialandBatchBundle(ERPNextTestSuite):
docstatus = frappe.db.get_value("Serial and Batch Bundle", bundle, "docstatus")
self.assertEqual(docstatus, 2)
def test_submitted_bundle_entries_cannot_be_mutated(self):
# A submitted Serial and Batch Bundle is the immutable source of truth for the stock
# ledger, live batch availability and repost/valuation replay. update_serial_batch_no_ledgers
# (which the whitelisted add_serial_batch_ledgers delegates to for an existing bundle) must
# refuse to rebuild -- and thereby inflate -- the quantities of an already submitted bundle.
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
update_serial_batch_no_ledgers,
)
item_code = make_item(
properties={
"has_batch_no": 1,
"create_new_batch": 1,
"batch_number_series": "TAMPER-SBB-.#####",
}
).name
se = make_stock_entry(
item_code=item_code,
target="_Test Warehouse - _TC",
qty=10,
rate=100,
)
bundle = se.items[0].serial_and_batch_bundle
self.assertEqual(frappe.db.get_value("Serial and Batch Bundle", bundle, "docstatus"), 1)
original = frappe.db.get_value(
"Serial and Batch Entry", {"parent": bundle}, ["name", "batch_no", "qty"], as_dict=True
)
self.assertEqual(original.qty, 10)
# Attempt to forge the submitted bundle: keep the same batch but inflate qty. The guard
# fires immediately after the bundle is loaded (docstatus check), so child_row / parent_doc
# only need the minimal fields the function reads.
tampered_entries = [{"batch_no": original.batch_no, "qty": 1000}]
child_row = frappe._dict({"name": se.items[0].name})
parent_doc = frappe._dict({"posting_date": today(), "posting_time": nowtime()})
self.assertRaises(
frappe.ValidationError,
update_serial_batch_no_ledgers,
bundle,
tampered_entries,
child_row,
parent_doc,
)
# The on-disk quantity must be untouched by the rejected mutation attempt.
self.assertEqual(
frappe.db.get_value("Serial and Batch Entry", original.name, "qty"),
10,
)
def test_batch_duplicate_entry(self):
item_code = make_item(properties={"has_batch_no": 1}).name

View File

@@ -9,7 +9,7 @@ from frappe.utils import add_to_date, cint, cstr, flt, get_datetime, now
import erpnext
from erpnext.accounts.utils import get_company_default
from erpnext.controllers.stock_controller import StockController, create_repost_item_valuation_entry
from erpnext.controllers.stock_controller import StockController
from erpnext.stock.doctype.batch.batch import get_available_batches, get_batch_qty
from erpnext.stock.doctype.inventory_dimension.inventory_dimension import get_inventory_dimensions
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
@@ -17,7 +17,12 @@ from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle impor
get_available_serial_nos,
)
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
<<<<<<< HEAD
from erpnext.stock.utils import get_combine_datetime, get_incoming_rate, get_stock_balance
=======
from erpnext.stock.doctype.stock_reconciliation_item.stock_reconciliation_item import StockReconciliationItem
from erpnext.stock.utils import get_incoming_rate, get_stock_balance
>>>>>>> e36426e235 (fix: do not allow to make changes in SABB after submit)
class OpeningEntryAccountError(frappe.ValidationError):
@@ -852,22 +857,6 @@ class StockReconciliation(StockController):
sl_entries.append(args)
def update_valuation_rate_for_serial_no(self):
for d in self.items:
if not d.serial_no:
continue
serial_nos = get_serial_nos(d.serial_no)
self.update_valuation_rate_for_serial_nos(d, serial_nos)
def update_valuation_rate_for_serial_nos(self, row, serial_nos):
valuation_rate = row.valuation_rate if self.docstatus == 1 else row.current_valuation_rate
if valuation_rate is None:
return
for d in serial_nos:
frappe.db.set_value("Serial No", d, "purchase_rate", valuation_rate)
def get_sle_for_items(self, row, serial_nos=None, current_bundle=True):
"""Insert Stock Ledger Entries"""
@@ -1021,11 +1010,6 @@ class StockReconciliation(StockController):
d.quantity_difference = flt(d.qty) - flt(d.current_qty)
d.amount_difference = flt(d.amount) - flt(d.current_amount)
def get_items_for(self, warehouse):
self.items = []
for item in get_items(warehouse, self.posting_date, self.posting_time, self.company):
self.append("items", item)
def submit(self):
if len(self.items) > 100:
msgprint(
@@ -1048,144 +1032,6 @@ class StockReconciliation(StockController):
else:
self._cancel()
def add_missing_stock_ledger_entry(self, row, voucher_detail_no, sle_creation):
if row.current_qty == 0:
return
new_sle = frappe.get_doc(self.get_sle_for_items(row))
new_sle.actual_qty = row.current_qty * -1
new_sle.valuation_rate = row.current_valuation_rate
new_sle.serial_and_batch_bundle = row.current_serial_and_batch_bundle
new_sle.flags.ignore_permissions = 1
new_sle.submit()
creation = add_to_date(sle_creation, seconds=-1)
new_sle.db_set("creation", creation)
if not frappe.db.exists(
"Repost Item Valuation",
{"item": row.item_code, "warehouse": row.warehouse, "docstatus": 1, "status": "Queued"},
):
create_repost_item_valuation_entry(
{
"based_on": "Item and Warehouse",
"item_code": row.item_code,
"warehouse": row.warehouse,
"company": self.company,
"allow_negative_stock": 1,
"posting_date": self.posting_date,
"posting_time": self.posting_time,
}
)
def has_negative_stock_allowed(self):
allow_negative_stock = cint(frappe.db.get_single_value("Stock Settings", "allow_negative_stock"))
if allow_negative_stock:
return True
if any(
((d.serial_and_batch_bundle or d.batch_no) and flt(d.qty) == flt(d.current_qty))
for d in self.items
):
allow_negative_stock = True
return allow_negative_stock
def get_current_qty_for_serial_or_batch(self, row, sle_creation):
doc = frappe.get_doc("Serial and Batch Bundle", row.current_serial_and_batch_bundle)
current_qty = 0.0
if doc.has_serial_no:
current_qty = self.get_current_qty_for_serial_nos(doc, sle_creation)
elif doc.has_batch_no:
current_qty = self.get_current_qty_for_batch_nos(doc, sle_creation)
return abs(current_qty)
def get_current_qty_for_serial_nos(self, doc, sle_creation):
serial_nos_details = get_available_serial_nos(
frappe._dict(
{
"item_code": doc.item_code,
"warehouse": doc.warehouse,
"posting_datetime": doc.posting_datetime,
"creation": sle_creation,
"voucher_no": self.name,
"ignore_warehouse": 1,
}
)
)
if not serial_nos_details:
return 0.0
doc.delete_serial_batch_entries()
current_qty = 0.0
for serial_no_row in serial_nos_details:
current_qty += 1
doc.append(
"entries",
{
"serial_no": serial_no_row.serial_no,
"qty": -1,
"warehouse": doc.warehouse,
"batch_no": serial_no_row.batch_no,
},
)
doc.set_incoming_rate(save=True)
doc.calculate_qty_and_amount(save=True)
doc.db_update_all()
return current_qty
def get_current_qty_for_batch_nos(self, doc, sle_creation):
current_qty = 0.0
precision = doc.entries[0].precision("qty")
for d in doc.entries:
qty = (
get_batch_qty(
d.batch_no,
doc.warehouse,
creation=sle_creation,
posting_datetime=doc.posting_datetime,
ignore_voucher_nos=[doc.voucher_no],
for_stock_levels=True,
consider_negative_batches=True,
do_not_check_future_batches=True,
)
or 0
) * -1
if flt(d.qty, precision) != flt(qty, precision):
d.db_set("qty", qty)
current_qty += qty
return current_qty
def get_batch_qty_for_stock_reco(
item_code, warehouse, batch_no, posting_date, posting_time, voucher_no, sle_creation
):
posting_datetime = get_combine_datetime(posting_date, posting_time)
qty = (
get_batch_qty(
batch_no,
warehouse,
item_code,
creation=sle_creation,
posting_datetime=posting_datetime,
ignore_voucher_nos=[voucher_no],
for_stock_levels=True,
consider_negative_batches=True,
do_not_check_future_batches=True,
)
or 0
)
return flt(qty)
@frappe.whitelist()
def get_items(warehouse, posting_date, posting_time, company, item_code=None, ignore_empty_stock=False):