refactor: stock reservation feature

This commit is contained in:
Rohit Waghchaure
2026-06-10 22:35:15 +05:30
parent aa9f225c41
commit 9c5f9218b5
15 changed files with 599 additions and 83 deletions

View File

@@ -984,9 +984,14 @@ class SellingController(StockController):
qty_can_be_deliver = 0
if sre_doc.reservation_based_on == "Serial and Batch":
sbb = frappe.get_doc("Serial and Batch Bundle", item.serial_and_batch_bundle)
# Delivered serial/batch may live in a Serial and Batch Bundle or directly in the
# row's serial_no/batch_no fields (use_serial_batch_fields). Read from whichever is
# present so this never crashes on a missing bundle.
(
delivered_serial_nos,
delivered_batch_qty,
) = get_delivered_serial_batch_for_reservation(item)
if sre_doc.has_serial_no:
delivered_serial_nos = [d.serial_no for d in sbb.entries]
for entry in sre_doc.sb_entries:
if entry.serial_no in delivered_serial_nos:
entry.delivered_qty = 1 # Qty will always be 0 or 1 for Serial No.
@@ -994,16 +999,16 @@ class SellingController(StockController):
qty_can_be_deliver += 1
delivered_serial_nos.remove(entry.serial_no)
else:
delivered_batch_qty = {d.batch_no: -1 * d.qty for d in sbb.entries}
for entry in sre_doc.sb_entries:
if entry.batch_no in delivered_batch_qty:
available_batch_qty = delivered_batch_qty.get(entry.batch_no, 0)
if available_batch_qty > 0:
delivered_qty = min(
(entry.qty - entry.delivered_qty), delivered_batch_qty[entry.batch_no]
(entry.qty - entry.delivered_qty), available_batch_qty
)
entry.delivered_qty += delivered_qty
entry.db_update()
qty_can_be_deliver += delivered_qty
delivered_batch_qty[entry.batch_no] -= delivered_qty
delivered_batch_qty[entry.batch_no] = available_batch_qty - delivered_qty
else:
# `Delivered Qty` should be less than or equal to `Reserved Qty`.
qty_can_be_deliver = min(
@@ -1179,3 +1184,31 @@ def get_serial_and_batch_bundle(child, parent, delivery_note_child=None):
child.db_set("serial_and_batch_bundle", doc.name)
return doc.name
def get_delivered_serial_batch_for_reservation(item):
"""Serial nos and per-batch qty delivered by a stock row.
The detail may be stored in a Serial and Batch Bundle or directly in the row's
``serial_no``/``batch_no`` fields (``use_serial_batch_fields``). Reading from whichever is
present keeps the Stock Reservation Entry delivered-qty update independent of a bundle being
created -- delivering reserved serial/batch stock used to crash when the row had no bundle.
"""
serial_nos, batch_qty = [], {}
if item.get("serial_and_batch_bundle"):
bundle = frappe.get_doc("Serial and Batch Bundle", item.serial_and_batch_bundle)
for row in bundle.entries:
if row.serial_no:
serial_nos.append(row.serial_no)
if row.batch_no:
batch_qty[row.batch_no] = batch_qty.get(row.batch_no, 0) + abs(flt(row.qty))
else:
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
if item.get("serial_no"):
serial_nos = get_serial_nos(item.serial_no)
if item.get("batch_no"):
batch_qty[item.batch_no] = abs(flt(item.get("stock_qty") or item.get("qty")))
return serial_nos, batch_qty

View File

@@ -116,17 +116,57 @@ def _sub_assembly_reserved_filter(table, child, item_code, warehouse):
)
class ProductionPlanStockReservation:
"""Reservation lifecycle for a Production Plan.
A Production Plan reserves stock for two of its child tables: the sub-assembly
items it will manufacture and the raw materials of its material-request rows
(see ``_RESERVATION_TABLES``). On submit the rows are reserved; on cancel the
reservations are released.
The reserved-qty *query* helpers in this module
(``get_reserved_qty_for_production_plan`` etc.) are read-only and answer "how
much is reserved?" for bins and reports, so they stay module-level functions
rather than methods, mirroring the engine's own query helpers.
"""
def __init__(self, doc):
self.doc = doc
def reserve(self, items: str | list | None = None, table_name: str | None = None, notify: bool = False):
"""Reserve (docstatus 1) or release (docstatus 2) stock for the plan's tables."""
if items and isinstance(items, str):
items = parse_json(items)
for child_table_name, kwargs in _RESERVATION_TABLES.items():
if table_name and table_name != child_table_name:
continue
self._reserve_or_cancel_plan_table(items, kwargs)
self.doc.reload()
def _reserve_or_cancel_plan_table(self, items, kwargs):
sre = StockReservation(self.doc, items=items, kwargs=kwargs)
if self.doc.docstatus == 1:
if sre.make_stock_reservation_entries():
frappe.msgprint(_("Stock Reservation Entries Created"), alert=True)
elif self.doc.docstatus == 2:
sre.cancel_stock_reservation_entries()
def cancel(self, sre_list: str | list | None = None):
"""Cancel specific (or all) Stock Reservation Entries held by the plan."""
StockReservation(self.doc).cancel_stock_reservation_entries(sre_list)
self.doc.reload()
@frappe.whitelist()
def make_stock_reservation_entries(
doc: str | Document, items: str | list | None = None, table_name: str | None = None, notify: bool = False
):
"""Whitelisted entry point: verify Production Plan write access, then reserve stock."""
if isinstance(doc, str):
doc = parse_json(doc)
doc = frappe.get_doc("Production Plan", doc.get("name"))
doc = _load_production_plan(doc)
frappe.has_permission("Production Plan", "write", doc=doc, throw=True)
reserve_stock_for_production_plan(doc, items=items, table_name=table_name, notify=notify)
ProductionPlanStockReservation(doc).reserve(items=items, table_name=table_name, notify=notify)
def reserve_stock_for_production_plan(
@@ -134,35 +174,19 @@ def reserve_stock_for_production_plan(
):
"""Reserve stock for a Production Plan. Internal: no permission check (also called
from the Production Plan submit/cancel lifecycle)."""
if items and isinstance(items, str):
items = parse_json(items)
for child_table_name, kwargs in _RESERVATION_TABLES.items():
if table_name and table_name != child_table_name:
continue
_reserve_or_cancel_plan_table(doc, items, kwargs)
doc.reload()
def _reserve_or_cancel_plan_table(doc, items, kwargs):
sre = StockReservation(doc, items=items, kwargs=kwargs)
if doc.docstatus == 1:
if sre.make_stock_reservation_entries():
frappe.msgprint(_("Stock Reservation Entries Created"), alert=True)
elif doc.docstatus == 2:
sre.cancel_stock_reservation_entries()
ProductionPlanStockReservation(doc).reserve(items=items, table_name=table_name, notify=notify)
@frappe.whitelist()
def cancel_stock_reservation_entries(doc: str | Document, sre_list: str | list):
"""Whitelisted entry point: verify Production Plan write access, then cancel reservations."""
doc = _load_production_plan(doc)
frappe.has_permission("Production Plan", "write", doc=doc, throw=True)
ProductionPlanStockReservation(doc).cancel(sre_list)
def _load_production_plan(doc: str | Document) -> Document:
if isinstance(doc, str):
doc = parse_json(doc)
doc = frappe.get_doc("Production Plan", doc.get("name"))
frappe.has_permission("Production Plan", "write", doc=doc, throw=True)
sre = StockReservation(doc)
sre.cancel_stock_reservation_entries(sre_list)
doc.reload()
return doc

View File

@@ -2322,6 +2322,74 @@ class TestProductionPlan(ERPNextTestSuite):
self.assertEqual(len(reserved_entries), 0)
frappe.db.set_single_value("Stock Settings", "enable_stock_reservation", 0)
def test_stock_reservation_restored_on_work_order_cancel(self):
# Spec #5 (cancellation path): when a Work Order created from a Production Plan is cancelled,
# the reservation that was transferred PP -> WO must flow back to the still-open Production
# Plan, not silently vanish.
from erpnext.manufacturing.doctype.bom.test_bom import create_nested_bom
frappe.db.set_single_value("Stock Settings", "enable_stock_reservation", 1)
try:
bom_tree = {
"FG For SR Cancel": {"Sub Assembly For SR Cancel 1": {"Raw Material For SR Cancel 1": {}}}
}
parent_bom = create_nested_bom(bom_tree, prefix="")
warehouse = "_Test Warehouse - _TC"
# Plenty of stock so the Production Plan reserves everything directly on submit.
for item_code in ["Sub Assembly For SR Cancel 1", "Raw Material For SR Cancel 1"]:
make_stock_entry(item_code=item_code, target=warehouse, qty=20, basic_rate=100)
plan = create_production_plan(
item_code=parent_bom.item,
planned_qty=10,
skip_available_sub_assembly_item=1,
ignore_existing_ordered_qty=1,
do_not_submit=1,
warehouse=warehouse,
sub_assembly_warehouse=warehouse,
for_warehouse=warehouse,
reserve_stock=1,
)
plan.get_sub_assembly_items()
plan.set("mr_items", [])
for d in get_items_for_material_requests(plan.as_dict()):
plan.append("mr_items", d)
plan.save()
plan.submit()
def pp_reserved():
return sum(
r.reserved_qty
for r in StockReservation(plan).get_reserved_entries("Production Plan", plan.name)
)
reserved_before = pp_reserved()
self.assertGreater(reserved_before, 0, "Production Plan should reserve stock on submit")
plan.make_work_order()
work_orders = frappe.get_all("Work Order", filters={"production_plan": plan.name}, pluck="name")
work_orders = list(set(work_orders))
for wo_name in work_orders:
wo_doc = frappe.get_doc("Work Order", wo_name)
wo_doc.source_warehouse = warehouse
wo_doc.wip_warehouse = warehouse
wo_doc.fg_warehouse = warehouse
wo_doc.submit()
# After all Work Orders are submitted the reservation has fully transferred off the plan.
self.assertEqual(pp_reserved(), 0, "Reservation should transfer PP -> WO on submit")
# Cancelling the Work Orders must return the reservation to the Production Plan.
for wo_name in work_orders:
frappe.get_doc("Work Order", wo_name).cancel()
self.assertEqual(
pp_reserved(), reserved_before, "Cancelling the Work Order must restore the PP reservation"
)
finally:
frappe.db.set_single_value("Stock Settings", "enable_stock_reservation", 0)
def test_stock_reservation_of_serial_nos_against_production_plan(self):
from erpnext.buying.doctype.purchase_order.mapper import make_purchase_receipt
from erpnext.manufacturing.doctype.bom.test_bom import create_nested_bom

View File

@@ -14,8 +14,8 @@ from pypika import functions as fn
from erpnext.manufacturing.doctype.bom.bom import get_bom_items_as_dict
from erpnext.manufacturing.doctype.work_order.mapper import check_if_scrap_warehouse_mandatory
from erpnext.manufacturing.doctype.work_order.services.stock_reservation import (
StockReservationService,
from erpnext.manufacturing.doctype.work_order.services.reservation import (
WorkOrderStockReservation,
get_consumed_qty,
get_row_wise_serial_batch,
)
@@ -44,7 +44,7 @@ class RequiredItemsService:
# update in bin
self.update_reserved_qty_for_production()
StockReservationService(self.doc).validate_reserved_qty()
WorkOrderStockReservation(self.doc).validate_reserved_qty()
def update_reserved_qty_for_production(self, items=None):
"""update reserved_qty_for_production in bins"""
@@ -142,7 +142,7 @@ class RequiredItemsService:
transferred_qty = transferred_items.get(row.item_code) or 0.0
row.db_set("transferred_qty", transferred_qty, update_modified=False)
if self.doc.reserve_stock:
StockReservationService(self.doc).update_qty_in_stock_reservation(
WorkOrderStockReservation(self.doc).update_qty_in_stock_reservation(
row, transferred_qty, row_wise_serial_batch
)
@@ -189,7 +189,7 @@ class RequiredItemsService:
continue
warehouse = wip_warehouse or item.source_warehouse
StockReservationService(self.doc).update_consumed_qty_in_stock_reservation(
WorkOrderStockReservation(self.doc).update_consumed_qty_in_stock_reservation(
item, consumed_qty, warehouse
)

View File

@@ -3,7 +3,7 @@
"""Stock reservation logic for Work Order.
Extracted from work_order.py. ``StockReservationService`` wraps a Work Order
Extracted from work_order.py. ``WorkOrderStockReservation`` wraps a Work Order
document (composition) and owns the reservation-related behaviour; the
module-level helpers are reused by the controller and by Production Plan.
work_order.py re-exports them to preserve whitelist dotted-paths and imports.
@@ -51,7 +51,7 @@ _SERIAL_BATCH_FIELDS = [
]
class StockReservationService:
class WorkOrderStockReservation:
def __init__(self, doc):
self.doc = doc
@@ -94,6 +94,11 @@ class StockReservationService:
self.doc.db_set("status", self.doc.get_status())
def update_qty_in_stock_reservation(self, row, transferred_qty, row_wise_serial_batch):
# `transferred_qty` is the absolute qty transferred to WIP recomputed from submitted stock
# entries, so this method must also be able to *lower* it (e.g. when a transfer is cancelled).
# A fully-transferred entry is "Closed"; it must stay eligible here, otherwise cancelling the
# transfer can never reset its transferred_qty and the Store reservation is lost. Only truly
# cancelled entries (docstatus 2) are excluded.
names = frappe.get_all(
"Stock Reservation Entry",
filters={
@@ -101,9 +106,10 @@ class StockReservationService:
"item_code": row.item_code,
"voucher_detail_no": row.name,
"warehouse": row.source_warehouse,
"status": ("not in", ["Closed", "Cancelled", "Completed"]),
"docstatus": 1,
},
pluck="name",
order_by="creation",
)
for name in names:
transferred_qty = self._apply_transferred_qty(name, transferred_qty, row_wise_serial_batch)
@@ -424,14 +430,26 @@ class StockReservationService:
)
def cancel_reserved_qty_for_wip_and_fg(self, ste_doc):
# Reservations created by this stock entry are identified by `from_voucher_no`. They can be
# held against the Work Order *or* against another voucher -- e.g. the finished good of an
# SO-linked Work Order is reserved against the Sales Order. They must be cancelled directly:
# routing through the Work-Order-scoped `cancel_stock_reservation_entries` would silently skip
# any entry whose `voucher_type`/`voucher_no` is not the Work Order, leaving the finished good
# reserved and making the manufacture impossible to cancel (NegativeStockError).
cancelled = False
for row in ste_doc.items:
sre_list = frappe.get_all(
"Stock Reservation Entry",
filters={"from_voucher_no": ste_doc.name, "from_voucher_detail_no": row.name, "docstatus": 1},
pluck="name",
)
if sre_list:
unreserve_stock_for_work_order(self.doc, sre_list)
for name in sre_list:
frappe.get_doc("Stock Reservation Entry", name).cancel()
cancelled = True
if cancelled:
self.doc.reload()
self.doc.db_set("status", self.doc.get_status())
def release_reserved_qty_for_subcontract_transfer(self):
"""Free this Work Order's own reservation for items sent to a subcontractor.

View File

@@ -3417,6 +3417,137 @@ class TestWorkOrder(ERPNextTestSuite):
self.assertRaises(frappe.ValidationError, transfer_entry.submit)
def test_stock_reservation_moves_from_store_to_wip_on_transfer(self):
# Spec #7: a Material Transfer for Manufacture (Store -> WIP) for a reserve_stock Work Order
# must move the reservation from the Store warehouse to the WIP warehouse; cancelling the
# transfer must move it back.
from erpnext.stock.doctype.stock_entry.stock_entry_utils import (
make_stock_entry as make_stock_entry_test_record,
)
frappe.db.set_single_value("Stock Settings", "enable_stock_reservation", 1)
try:
store = "Stores - _TC"
wip = "Work In Progress - _TC"
fg = make_item("Test SR Move FG", {"is_stock_item": 1}).name
rm = make_item("Test SR Move RM", {"is_stock_item": 1}).name
bom = make_bom(item=fg, raw_materials=[rm], source_warehouse=store, do_not_submit=True)
bom.save()
bom.submit()
make_stock_entry_test_record(item_code=rm, target=store, qty=10, basic_rate=100)
wo = make_wo_order_test_record(
production_item=fg,
qty=10,
bom_no=bom.name,
reserve_stock=1,
source_warehouse=store,
wip_warehouse=wip,
fg_warehouse=wip,
do_not_save=True,
)
wo.save()
wo.submit()
def reserved_in(warehouse):
return sum(
flt(r.reserved_qty) - flt(r.transferred_qty) - flt(r.delivered_qty) - flt(r.consumed_qty)
for r in frappe.get_all(
"Stock Reservation Entry",
filters={
"voucher_no": wo.name,
"item_code": rm,
"warehouse": warehouse,
"docstatus": 1,
},
fields=["reserved_qty", "transferred_qty", "delivered_qty", "consumed_qty"],
)
)
self.assertEqual(reserved_in(store), 10, "RM should be reserved in Store after WO submit")
self.assertEqual(reserved_in(wip), 0)
# Transfer Store -> WIP.
se = frappe.get_doc(make_stock_entry(wo.name, "Material Transfer for Manufacture", 10))
se.submit()
self.assertEqual(reserved_in(store), 0, "Store reservation should be freed after transfer")
self.assertEqual(reserved_in(wip), 10, "Reservation should move to WIP after transfer")
# Cancel the transfer -> reservation returns to Store.
se.cancel()
self.assertEqual(reserved_in(store), 10, "Cancelling transfer must restore Store reservation")
self.assertEqual(reserved_in(wip), 0)
finally:
frappe.db.set_single_value("Stock Settings", "enable_stock_reservation", 0)
def test_sales_order_linked_work_order_reserves_finished_good(self):
# Spec #8: when a Work Order is linked to a Sales Order, manufacturing the finished good must
# reserve it against that Sales Order; cancelling the manufacture must release the reservation.
from erpnext.stock.doctype.stock_entry.stock_entry_utils import (
make_stock_entry as make_stock_entry_test_record,
)
frappe.db.set_single_value("Stock Settings", "enable_stock_reservation", 1)
try:
warehouse = "_Test Warehouse - _TC"
wip = "_Test Warehouse 1 - _TC"
fg = make_item("Test SR SO-WO FG", {"is_stock_item": 1}).name
rm = make_item("Test SR SO-WO RM", {"is_stock_item": 1}).name
bom = make_bom(item=fg, raw_materials=[rm], source_warehouse=warehouse, do_not_submit=True)
bom.save()
bom.submit()
make_stock_entry_test_record(item_code=rm, target=warehouse, qty=10, basic_rate=100)
# The finished good is reserved in the Sales Order item's warehouse, so the WO must produce
# the FG into that same warehouse.
so = make_sales_order(item_code=fg, warehouse=warehouse, qty=10, rate=500)
wo = make_wo_order_test_record(
production_item=fg,
qty=10,
bom_no=bom.name,
sales_order=so.name,
reserve_stock=1,
source_warehouse=warehouse,
wip_warehouse=wip,
fg_warehouse=warehouse,
do_not_save=True,
)
wo.save()
wo.submit()
# Transfer the raw material to WIP, then manufacture the finished good.
transfer = frappe.get_doc(make_stock_entry(wo.name, "Material Transfer for Manufacture", 10))
transfer.submit()
manufacture = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 10))
manufacture.submit()
def so_fg_reserved():
return sum(
flt(r.reserved_qty)
for r in frappe.get_all(
"Stock Reservation Entry",
filters={
"voucher_type": "Sales Order",
"voucher_no": so.name,
"item_code": fg,
"docstatus": 1,
},
fields=["reserved_qty"],
)
)
self.assertEqual(so_fg_reserved(), 10, "Finished good should be reserved against the Sales Order")
# Cancelling the manufacture releases the finished-good reservation.
manufacture.cancel()
self.assertEqual(so_fg_reserved(), 0, "Cancelling manufacture must release the SO reservation")
finally:
frappe.db.set_single_value("Stock Settings", "enable_stock_reservation", 0)
def test_send_to_subcontractor_can_consume_work_order_reserved_stock(self):
from erpnext.buying.doctype.purchase_order.mapper import make_subcontracting_order
from erpnext.controllers.subcontracting_controller import make_rm_stock_entry

View File

@@ -46,11 +46,8 @@ from erpnext.manufacturing.doctype.work_order.services.operations import (
from erpnext.manufacturing.doctype.work_order.services.required_items import (
RequiredItemsService,
)
from erpnext.manufacturing.doctype.work_order.services.status import (
StatusService,
)
from erpnext.manufacturing.doctype.work_order.services.stock_reservation import (
StockReservationService,
from erpnext.manufacturing.doctype.work_order.services.reservation import (
WorkOrderStockReservation,
cancel_stock_reservation_entries,
get_consumed_qty,
get_reserved_qty_for_production,
@@ -58,6 +55,9 @@ from erpnext.manufacturing.doctype.work_order.services.stock_reservation import
get_sre_details,
make_stock_reservation_entries,
)
from erpnext.manufacturing.doctype.work_order.services.status import (
StatusService,
)
from erpnext.stock.doctype.batch.batch import make_batch
from erpnext.stock.doctype.item.item import validate_end_of_life
from erpnext.stock.doctype.serial_no.serial_no import get_available_serial_nos
@@ -297,8 +297,8 @@ class WorkOrder(Document):
self.status = self.get_status()
self.validate_workstation_type()
self.reset_use_multi_level_bom()
StockReservationService(self).set_reserve_stock()
StockReservationService(self).validate_fg_warehouse_for_reservation()
WorkOrderStockReservation(self).set_reserve_stock()
WorkOrderStockReservation(self).validate_fg_warehouse_for_reservation()
self.validate_dates()
if self.source_warehouse:
@@ -311,7 +311,7 @@ class WorkOrder(Document):
):
self.set_required_items(reset_only_qty=len(self.get("required_items")))
StockReservationService(self).enable_auto_reserve_stock()
WorkOrderStockReservation(self).enable_auto_reserve_stock()
self.validate_operations_sequence()
self.validate_subcontracting_inward_order()
@@ -625,7 +625,7 @@ class WorkOrder(Document):
self.create_job_card_from_wo()
if self.reserve_stock:
StockReservationService(self).update_stock_reservation()
WorkOrderStockReservation(self).update_stock_reservation()
self.update_subcontracting_inward_order_received_items()
@@ -649,7 +649,7 @@ class WorkOrder(Document):
self.update_reserved_qty_for_production()
if self.reserve_stock:
StockReservationService(self).update_stock_reservation()
WorkOrderStockReservation(self).update_stock_reservation()
self.update_subcontracting_inward_order_received_items()

View File

@@ -376,12 +376,27 @@ def make_delivery_note(
dn_item.qty = flt(sre.reserved_qty) / flt(dn_item.get("conversion_factor", 1))
dn_item.warehouse = sre.warehouse
if (
not use_serial_batch_fields
and sre.reservation_based_on == "Serial and Batch"
and (sre.has_serial_no or sre.has_batch_no)
):
dn_item.serial_and_batch_bundle = get_ssb_bundle_for_voucher([sre]).name
if sre.reservation_based_on == "Serial and Batch" and (sre.has_serial_no or sre.has_batch_no):
if use_serial_batch_fields:
# Carry the reserved serial/batch in the row fields. A single field can't hold
# multiple batches, so fall back to a bundle in that case.
dn_item.use_serial_batch_fields = 1
sb_entries = frappe.get_all(
"Serial and Batch Entry",
filters={"parent": sre.name},
fields=["serial_no", "batch_no"],
)
serial_nos = [d.serial_no for d in sb_entries if d.serial_no]
batch_nos = list({d.batch_no for d in sb_entries if d.batch_no})
if serial_nos:
dn_item.serial_no = "\n".join(serial_nos)
if len(batch_nos) == 1:
dn_item.batch_no = batch_nos[0]
elif len(batch_nos) > 1:
dn_item.use_serial_batch_fields = 0
dn_item.serial_and_batch_bundle = get_ssb_bundle_for_voucher([sre]).name
else:
dn_item.serial_and_batch_bundle = get_ssb_bundle_for_voucher([sre]).name
target_doc.append("items", dn_item)
# Correct rows index.

View File

@@ -23,8 +23,8 @@ from erpnext.manufacturing.doctype.blanket_order.blanket_order import (
)
from erpnext.selling.doctype.customer.customer import check_credit_limit
from erpnext.selling.doctype.sales_order.services.delivery_schedule import DeliveryScheduleService
from erpnext.selling.doctype.sales_order.services.reservation import SalesOrderStockReservation
from erpnext.selling.doctype.sales_order.services.status import StatusService
from erpnext.selling.doctype.sales_order.services.stock_reservation import StockReservationService
from erpnext.selling.doctype.sales_order.services.subcontracting import SubcontractingService
from erpnext.stock.doctype.packed_item.packed_item import make_packing_list
from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import has_reserved_stock
@@ -226,7 +226,7 @@ class SalesOrder(SellingController):
self.validate_for_items()
self.validate_warehouse()
self.validate_drop_ship()
StockReservationService(self).validate_reserved_stock()
SalesOrderStockReservation(self).validate_reserved_stock()
self.validate_serial_no_based_delivery()
validate_against_blanket_order(self)
validate_inter_company_party(
@@ -248,7 +248,7 @@ class SalesOrder(SellingController):
self.reset_default_field_value("set_warehouse", "items", "warehouse")
if not self.get("is_subcontracted"):
StockReservationService(self).enable_auto_reserve_stock()
SalesOrderStockReservation(self).enable_auto_reserve_stock()
def set_has_unit_price_items(self):
"""
@@ -542,7 +542,7 @@ class SalesOrder(SellingController):
StatusService(self).update_status(status)
def update_reserved_qty(self, so_item_rows=None):
StockReservationService(self).update_reserved_qty(so_item_rows)
SalesOrderStockReservation(self).update_reserved_qty(so_item_rows)
def on_update_after_submit(self):
self.calculate_commission()
@@ -652,7 +652,7 @@ class SalesOrder(SellingController):
@frappe.whitelist()
def has_unreserved_stock(self, table_name: str = "items") -> dict:
"""Returns unreserved qty per item if there is any unreserved item in the Sales Order."""
return StockReservationService(self).has_unreserved_stock(table_name)
return SalesOrderStockReservation(self).has_unreserved_stock(table_name)
@frappe.whitelist()
def create_stock_reservation_entries(
@@ -662,14 +662,14 @@ class SalesOrder(SellingController):
notify: bool = True,
) -> None:
"""Creates Stock Reservation Entries for Sales Order Items."""
StockReservationService(self).create_stock_reservation_entries(
SalesOrderStockReservation(self).create_stock_reservation_entries(
items_details, from_voucher_type, notify
)
@frappe.whitelist()
def cancel_stock_reservation_entries(self, sre_list: list | None = None, notify: bool = True) -> None:
"""Cancel Stock Reservation Entries for Sales Order Items."""
StockReservationService(self).cancel_stock_reservation_entries(sre_list, notify)
SalesOrderStockReservation(self).cancel_stock_reservation_entries(sre_list, notify)
def set_missing_values(self, for_validate=False):
super().set_missing_values(for_validate)

View File

@@ -15,7 +15,7 @@ from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry impor
from erpnext.stock.stock_balance import get_reserved_qty, update_bin_qty
class StockReservationService:
class SalesOrderStockReservation:
def __init__(self, doc):
self.doc = doc

View File

@@ -15,7 +15,9 @@ from erpnext.stock.doctype.purchase_receipt.services.billing_status import Billi
from erpnext.stock.doctype.purchase_receipt.services.provisional_accounting import (
ProvisionalAccountingService,
)
from erpnext.stock.doctype.purchase_receipt.services.stock_reservation import StockReservationService
from erpnext.stock.doctype.purchase_receipt.services.reservation import (
PurchaseReceiptStockReservation,
)
form_grid_templates = {"items": "templates/form_grid/item_grid.html"}
@@ -374,7 +376,7 @@ class PurchaseReceipt(BuyingController):
self.make_gl_entries()
self.repost_future_sle_and_gle()
self.set_consumed_qty_in_subcontract_order()
StockReservationService(self).reserve_stock()
PurchaseReceiptStockReservation(self).reserve_stock()
self.update_received_qty_if_from_pp()
def update_received_qty_if_from_pp(self):

View File

@@ -10,7 +10,7 @@ from frappe.utils import get_datetime
from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import StockReservation
class StockReservationService:
class PurchaseReceiptStockReservation:
def __init__(self, doc):
self.doc = doc

View File

@@ -1061,8 +1061,8 @@ class StockEntry(StockController, SubcontractingInwardController):
return getattr(self, "_wo_doc", None)
def make_stock_reserve_for_wip_and_fg(self):
from erpnext.manufacturing.doctype.work_order.services.stock_reservation import (
StockReservationService,
from erpnext.manufacturing.doctype.work_order.services.reservation import (
WorkOrderStockReservation,
)
if self.is_stock_reserve_for_work_order():
@@ -1076,7 +1076,7 @@ class StockEntry(StockController, SubcontractingInwardController):
):
return
StockReservationService(pro_doc).set_reserved_qty_for_wip_and_fg(self)
WorkOrderStockReservation(pro_doc).set_reserved_qty_for_wip_and_fg(self)
def reserve_stock_for_subcontracting(self):
if self.purpose == "Send to Subcontractor" and frappe.get_value(
@@ -1103,8 +1103,8 @@ class StockEntry(StockController, SubcontractingInwardController):
)
def cancel_stock_reserve_for_wip_and_fg(self):
from erpnext.manufacturing.doctype.work_order.services.stock_reservation import (
StockReservationService,
from erpnext.manufacturing.doctype.work_order.services.reservation import (
WorkOrderStockReservation,
)
if self.is_stock_reserve_for_work_order():
@@ -1116,7 +1116,7 @@ class StockEntry(StockController, SubcontractingInwardController):
):
return
StockReservationService(pro_doc).cancel_reserved_qty_for_wip_and_fg(self)
WorkOrderStockReservation(pro_doc).cancel_reserved_qty_for_wip_and_fg(self)
def is_stock_reserve_for_work_order(self):
if (
@@ -1133,8 +1133,8 @@ class StockEntry(StockController, SubcontractingInwardController):
# purpose), so the owning Work Order is derived from the Subcontracting Order / Purchase Order
# that raised the transfer. Each such Work Order that reserves stock gets its reservation for
# the sent items released, so the negative-stock guard stops blocking the consumption.
from erpnext.manufacturing.doctype.work_order.services.stock_reservation import (
StockReservationService,
from erpnext.manufacturing.doctype.work_order.services.reservation import (
WorkOrderStockReservation,
)
if self.purpose != "Send to Subcontractor":
@@ -1142,7 +1142,7 @@ class StockEntry(StockController, SubcontractingInwardController):
for wo_name in self.get_reserved_work_orders_for_subcontracting():
pro_doc = frappe.get_doc("Work Order", wo_name)
StockReservationService(pro_doc).release_reserved_qty_for_subcontract_transfer()
WorkOrderStockReservation(pro_doc).release_reserved_qty_for_subcontract_transfer()
def get_reserved_work_orders_for_subcontracting(self):
job_cards = set()

View File

@@ -1587,7 +1587,7 @@ def create_stock_reservation_entries_for_so_items(
):
"""Creates Stock Reservation Entries for Sales Order Items."""
from erpnext.selling.doctype.sales_order.services.stock_reservation import get_unreserved_qty
from erpnext.selling.doctype.sales_order.services.reservation import get_unreserved_qty
if not from_voucher_type and (
sales_order.get("_action") == "submit"

View File

@@ -4,7 +4,7 @@
from random import randint
import frappe
from frappe.utils import today
from frappe.utils import flt, today
from erpnext.selling.doctype.sales_order.mapper import create_pick_list, make_delivery_note
from erpnext.selling.doctype.sales_order.test_sales_order import make_sales_order
@@ -313,6 +313,231 @@ class TestStockReservationEntry(ERPNextTestSuite):
for sre_detail in sre_details:
self.assertEqual(sre_detail.reserved_qty, sre_detail.delivered_qty)
@ERPNextTestSuite.change_settings(
"Stock Settings", {"enable_stock_reservation": 1, "allow_partial_reservation": 1}
)
def test_reservation_restored_on_delivery_note_cancel(self) -> None:
# Cancellation path (spec #1): delivering reserved stock via a Delivery Note marks the SRE
# delivered; cancelling that Delivery Note must restore the reservation (delivered_qty -> 0),
# otherwise the reserved stock is silently lost.
so = make_sales_order(
item_code=self.sr_item.name,
warehouse=self.warehouse,
qty=10,
rate=100,
do_not_submit=True,
)
so.reserve_stock = 1
so.items[0].reserve_stock = 1
so.save()
so.submit()
so.create_stock_reservation_entries()
def sre():
return frappe.get_all(
"Stock Reservation Entry",
filters={"voucher_no": so.name, "item_code": self.sr_item.name, "docstatus": 1},
fields=["reserved_qty", "delivered_qty", "status"],
)[0]
self.assertEqual(sre().reserved_qty, 10)
self.assertEqual(sre().delivered_qty, 0)
dn = make_delivery_note(so.name)
dn.submit()
after = sre()
self.assertEqual(after.delivered_qty, 10, "Delivery Note should mark the reservation delivered")
self.assertEqual(after.status, "Delivered")
dn.cancel()
restored = sre()
self.assertEqual(
restored.delivered_qty, 0, "Cancelling the Delivery Note must restore the reservation"
)
self.assertEqual(restored.status, "Reserved")
@ERPNextTestSuite.change_settings(
"Stock Settings", {"enable_stock_reservation": 1, "allow_negative_stock": 0}
)
def test_reserved_stock_cannot_be_delivered_against_a_different_sales_order(self) -> None:
# Spec #2: stock reserved for one Sales Order must not be deliverable through a Delivery Note
# raised for a different Sales Order. Pin allow_negative_stock off so the guard is enforced
# regardless of any global Stock Settings left enabled by other tests in the suite.
from erpnext.stock.stock_ledger import NegativeStockError
item_doc = make_item(properties={"is_stock_item": 1, "valuation_rate": 100})
item = item_doc.name
warehouse = self.warehouse
create_material_receipt(items={item: item_doc}, warehouse=warehouse, qty=10)
# SO-A reserves the entire available stock.
so_a = make_sales_order(item_code=item, warehouse=warehouse, qty=10, rate=100, do_not_submit=True)
so_a.reserve_stock = 1
so_a.items[0].reserve_stock = 1
so_a.save()
so_a.submit()
so_a.create_stock_reservation_entries()
self.assertTrue(has_reserved_stock("Sales Order", so_a.name))
# SO-B (a different order) for the same item must not be able to deliver the reserved stock.
so_b = make_sales_order(item_code=item, warehouse=warehouse, qty=10, rate=100, do_not_submit=True)
so_b.save()
so_b.submit()
dn = make_delivery_note(so_b.name)
dn.save()
self.assertRaises(NegativeStockError, dn.submit)
@ERPNextTestSuite.change_settings("Stock Settings", {"enable_stock_reservation": 1})
def test_stock_can_be_unreserved_and_reserved_against_another_sales_order(self) -> None:
# Spec #3: a user can manually unreserve stock from one Sales Order and reserve the same stock
# against another Sales Order.
item_doc = make_item(properties={"is_stock_item": 1, "valuation_rate": 100})
item = item_doc.name
warehouse = self.warehouse
create_material_receipt(items={item: item_doc}, warehouse=warehouse, qty=10)
so_a = make_sales_order(item_code=item, warehouse=warehouse, qty=10, rate=100, do_not_submit=True)
so_a.reserve_stock = 1
so_a.items[0].reserve_stock = 1
so_a.save()
so_a.submit()
so_a.create_stock_reservation_entries()
self.assertTrue(has_reserved_stock("Sales Order", so_a.name))
so_b = make_sales_order(item_code=item, warehouse=warehouse, qty=10, rate=100, do_not_submit=True)
so_b.reserve_stock = 1
so_b.items[0].reserve_stock = 1
so_b.save()
so_b.submit()
# With all stock reserved by SO-A, SO-B cannot reserve anything yet.
so_b.create_stock_reservation_entries()
self.assertFalse(has_reserved_stock("Sales Order", so_b.name))
# Manually unreserve SO-A, then SO-B can reserve the freed stock.
cancel_stock_reservation_entries("Sales Order", so_a.name)
self.assertFalse(has_reserved_stock("Sales Order", so_a.name))
so_b.create_stock_reservation_entries()
self.assertTrue(has_reserved_stock("Sales Order", so_b.name))
@ERPNextTestSuite.change_settings(
"Stock Settings",
{
"enable_stock_reservation": 1,
"auto_reserve_serial_and_batch": 1,
"pick_serial_and_batch_based_on": "FIFO",
},
)
def test_serial_and_batch_reservation_can_be_unreserved(self) -> None:
# Spec #9 (cancellation): an auto-reserved serial/batch Sales Order reservation pins specific
# serial/batch entries on the SRE (`sb_entries`). Manually unreserving it must cancel the SRE
# (and its pinned entries) and free the stock for another order.
items_details = create_items()
create_material_receipt(items_details, self.warehouse, qty=10)
serial_item = next(
name for name, p in items_details.items() if p.get("has_serial_no") and not p.get("has_batch_no")
)
batch_item = next(
name for name, p in items_details.items() if p.get("has_batch_no") and not p.get("has_serial_no")
)
item_list = [
{"item_code": serial_item, "warehouse": self.warehouse, "qty": 10, "rate": 100},
{"item_code": batch_item, "warehouse": self.warehouse, "qty": 10, "rate": 100},
]
so = make_sales_order(item_list=item_list, warehouse=self.warehouse)
so.create_stock_reservation_entries()
so.load_from_db()
def sre_row(so_item):
return frappe.get_all(
"Stock Reservation Entry",
filters={"voucher_no": so.name, "voucher_detail_no": so_item, "docstatus": 1},
fields=["name", "reserved_qty", "reservation_based_on"],
)
# Each item is reserved on a Serial-and-Batch basis with the specific entries pinned.
self.assertTrue(has_reserved_stock("Sales Order", so.name))
for item in so.items:
rows = sre_row(item.name)
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0].reserved_qty, 10)
self.assertEqual(rows[0].reservation_based_on, "Serial and Batch")
pinned = frappe.get_all(
"Serial and Batch Entry",
filters={"parent": rows[0].name, "parentfield": "sb_entries"},
)
self.assertGreaterEqual(len(pinned), 1, "serial/batch entries should be pinned on the SRE")
# Manually unreserve: the SRE (and its pinned entries) must be cancelled and the stock freed.
cancel_stock_reservation_entries("Sales Order", so.name)
self.assertFalse(has_reserved_stock("Sales Order", so.name))
for item in so.items:
self.assertEqual(sre_row(item.name), [])
# The freed serial/batch stock can be reserved by another Sales Order.
so_b = make_sales_order(item_list=item_list, warehouse=self.warehouse)
so_b.create_stock_reservation_entries()
self.assertTrue(has_reserved_stock("Sales Order", so_b.name))
@ERPNextTestSuite.change_settings(
"Stock Settings",
{
"enable_stock_reservation": 1,
"auto_reserve_serial_and_batch": 1,
"pick_serial_and_batch_based_on": "FIFO",
"use_serial_batch_fields": 1,
},
)
def test_serial_and_batch_reserved_stock_delivery_and_cancel(self) -> None:
# Regression: delivering a serial/batch reserved Sales Order used to crash with
# "Serial and Batch Bundle None not found" when the delivered serial/batch was carried in the
# row's serial_no/batch_no fields (use_serial_batch_fields) rather than a bundle. Delivery
# must mark the reservation delivered, and cancelling the Delivery Note must restore it.
items_details = create_items()
create_material_receipt(items_details, self.warehouse, qty=10)
serial_item = next(
name for name, p in items_details.items() if p.get("has_serial_no") and not p.get("has_batch_no")
)
batch_item = next(
name for name, p in items_details.items() if p.get("has_batch_no") and not p.get("has_serial_no")
)
item_list = [
{"item_code": serial_item, "warehouse": self.warehouse, "qty": 10, "rate": 100},
{"item_code": batch_item, "warehouse": self.warehouse, "qty": 10, "rate": 100},
]
so = make_sales_order(item_list=item_list, warehouse=self.warehouse)
so.create_stock_reservation_entries()
def sre_row(so_item):
return frappe.get_all(
"Stock Reservation Entry",
filters={"voucher_no": so.name, "voucher_detail_no": so_item, "docstatus": 1},
fields=["reserved_qty", "delivered_qty", "status"],
)[0]
# Deliver the reserved stock (serial/batch carried in row fields, no bundle).
dn = make_delivery_note(so.name, kwargs={"for_reserved_stock": True})
dn.save()
dn.submit()
for item in so.items:
row = sre_row(item.name)
self.assertEqual(
row.delivered_qty, 10, "delivery should mark the serial/batch reservation delivered"
)
self.assertEqual(row.status, "Delivered")
# Cancelling the Delivery Note restores the reservation.
dn.cancel()
for item in so.items:
row = sre_row(item.name)
self.assertEqual(row.delivered_qty, 0, "DN cancel must restore the serial/batch reservation")
self.assertEqual(row.status, "Reserved")
@ERPNextTestSuite.change_settings(
"Stock Settings",
{