fix: Stock Reservation blocks Subcontracting operation within the same Work Order

This commit is contained in:
Rohit Waghchaure
2026-06-09 16:31:19 +05:30
parent 7904385b90
commit 1c0dace3d6
4 changed files with 320 additions and 0 deletions

View File

@@ -877,6 +877,9 @@ class JobCard(Document):
)
def validate_time_logs_present(self):
if self.track_semi_finished_goods and self.is_subcontracted:
return
if not self.time_logs:
frappe.throw(
_("Time logs are required for {0} {1}").format(

View File

@@ -3414,6 +3414,180 @@ class TestWorkOrder(ERPNextTestSuite):
self.assertRaises(frappe.ValidationError, transfer_entry.submit)
@ERPNextTestSuite.change_settings(
"Buying Settings",
{"backflush_raw_materials_of_subcontract_based_on": "Material Transferred for Subcontract"},
)
def test_send_to_subcontractor_can_consume_work_order_reserved_stock(self):
from erpnext.buying.doctype.purchase_order.purchase_order import make_subcontracting_order
from erpnext.controllers.subcontracting_controller import make_rm_stock_entry
from erpnext.manufacturing.doctype.job_card.job_card import make_subcontracting_po
from erpnext.stock.doctype.stock_entry.stock_entry_utils import (
make_stock_entry as make_stock_entry_test_record,
)
from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
company = "_Test Company"
warehouse = "Stores - _TC"
supplier_warehouse = create_warehouse("Test S2S Supplier WH", company=company)
fabric = make_item("Test S2S Fabric", {"is_stock_item": 1}).name
stitched = make_item("Test S2S Stitched Shirt", {"is_stock_item": 1}).name
tshirt = make_item("Test S2S T-Shirt", {"is_stock_item": 1, "is_sub_contracted_item": 1}).name
service_item = make_item("Test S2S Ironing Service", {"is_stock_item": 0}).name
# Semi-FG BOM: Stitched Shirt from Fabric.
sfg_bom = frappe.new_doc("BOM")
sfg_bom.company = company
sfg_bom.item = stitched
sfg_bom.quantity = 1
sfg_bom.append("items", {"item_code": fabric, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
# Subcontracting BOM: how to make the final T-Shirt at the supplier (consuming Stitched Shirt).
tshirt_from_stitched = frappe.new_doc("BOM")
tshirt_from_stitched.company = company
tshirt_from_stitched.item = tshirt
tshirt_from_stitched.quantity = 1
tshirt_from_stitched.append("items", {"item_code": stitched, "qty": 1})
tshirt_from_stitched.insert()
tshirt_from_stitched.submit()
if not frappe.db.exists("Subcontracting BOM", {"finished_good": tshirt}):
frappe.get_doc(
{
"doctype": "Subcontracting BOM",
"finished_good": tshirt,
"finished_good_qty": 1,
"service_item": service_item,
"service_item_qty": 1,
"finished_good_bom": tshirt_from_stitched.name,
"is_active": 1,
}
).insert()
if not frappe.db.exists("Workstation", "Test S2S Workstation"):
make_workstation(workstation="Test S2S Workstation", production_capacity=1)
for op in ("Test S2S Stitching", "Test S2S Ironing"):
if not frappe.db.exists("Operation", op):
make_operation(operation=op, workstation="Test S2S Workstation")
# Final BOM for T-Shirt: internal Stitching op (produces Stitched Shirt) + subcontracted Ironing.
fg_bom = frappe.new_doc("BOM")
fg_bom.company = company
fg_bom.item = tshirt
fg_bom.quantity = 1
fg_bom.with_operations = 1
fg_bom.track_semi_finished_goods = 1
fg_bom.append("items", {"item_code": fabric, "qty": 1})
fg_bom.append(
"operations",
{
"operation": "Test S2S Stitching",
"workstation": "Test S2S Workstation",
"finished_good": stitched,
"finished_good_qty": 1,
"bom_no": sfg_bom.name,
"time_in_mins": 60,
"sequence_id": 1,
},
)
fg_bom.append(
"operations",
{
"operation": "Test S2S Ironing",
"workstation": "Test S2S Workstation",
"finished_good": tshirt,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"is_subcontracted": 1,
"bom_no": tshirt_from_stitched.name,
"time_in_mins": 60,
"sequence_id": 2,
},
)
fg_bom.append("items", {"item_code": stitched, "qty": 1, "operation_row_id": 2})
fg_bom.insert()
fg_bom.submit()
make_stock_entry_test_record(item_code=fabric, target=warehouse, qty=10, basic_rate=100)
wo = make_wo_order_test_record(
production_item=tshirt,
qty=10,
bom_no=fg_bom.name,
reserve_stock=1,
skip_transfer=1,
source_warehouse=warehouse,
wip_warehouse=warehouse,
fg_warehouse=warehouse,
do_not_save=True,
)
wo.operations[0].time_in_mins = 60
wo.operations[1].time_in_mins = 60
wo.save()
wo.submit()
# Complete the internal Stitching job card -> Stitched Shirt is produced into WIP and reserved.
stitching_jc = frappe.get_doc(
"Job Card",
frappe.db.get_value("Job Card", {"work_order": wo.name, "operation": "Test S2S Stitching"}),
)
stitching_jc.append(
"time_logs",
{
"from_time": "2024-01-01 08:00:00",
"to_time": "2024-01-01 09:00:00",
"completed_qty": stitching_jc.for_quantity,
},
)
stitching_jc.submit()
manufacturing_entry = frappe.get_doc(stitching_jc.make_stock_entry_for_semi_fg_item())
manufacturing_entry.submit()
sre_name = frappe.db.get_value(
"Stock Reservation Entry",
{"voucher_no": wo.name, "item_code": stitched, "warehouse": warehouse, "docstatus": 1},
)
self.assertTrue(sre_name, "Work Order should have reserved the semi-finished good")
# Subcontract the Ironing operation: Job Card -> Subcontracting PO -> Subcontracting Order.
ironing_jc = frappe.db.get_value("Job Card", {"work_order": wo.name, "operation": "Test S2S Ironing"})
po = frappe.get_doc(make_subcontracting_po(ironing_jc))
po.supplier = "_Test Supplier"
po.supplier_warehouse = supplier_warehouse
po.schedule_date = nowdate()
for item in po.items:
item.schedule_date = nowdate()
po.insert()
po.submit()
sco = make_subcontracting_order(po.name)
sco.supplier_warehouse = supplier_warehouse
for item in sco.supplied_items:
item.reserve_warehouse = warehouse
sco.insert()
sco.submit()
# Transfer the reserved Stitched Shirt to the subcontractor. This must NOT raise
# NegativeStockError ("reserved for other transactions").
ste = frappe.new_doc("Stock Entry").update(make_rm_stock_entry(sco.name))
ste.insert()
ste.submit()
# The reservation is freed: transferred_qty == sent qty and the SRE is Closed.
sre = frappe.get_doc("Stock Reservation Entry", sre_name)
self.assertEqual(sre.transferred_qty, 10)
self.assertEqual(sre.status, "Closed")
# Cancelling the transfer restores the reservation.
ste.cancel()
sre.reload()
self.assertEqual(sre.transferred_qty, 0)
self.assertEqual(sre.status, "Reserved")
def test_stock_reservation_for_batched_raw_material(self):
from erpnext.stock.doctype.stock_entry.stock_entry_utils import (
make_stock_entry as make_stock_entry_test_record,

View File

@@ -2138,6 +2138,98 @@ class WorkOrder(Document):
if sre_list:
cancel_stock_reservation_entries(self, sre_list)
def release_reserved_qty_for_subcontract_transfer(self):
"""Free this Work Order's own reservation for items sent to a subcontractor.
A ``Send to Subcontractor`` Stock Entry raised against a Work Order consumes stock that
the same Work Order reserved (e.g. the semi-finished item of a subcontracted operation).
The sent qty is recorded as ``transferred_qty`` on the matching Stock Reservation Entries
so the negative-stock guard stops treating it as reserved for "other transactions". The
figure is recomputed from every submitted ``Send to Subcontractor`` entry for the Work
Order, so it self-corrects on cancellation / reposting.
Note: only qty-based reservations are handled here; serial/batch reservations are left to
the existing material-transfer machinery.
"""
sent = self._subcontract_transferred_qty_by_item()
entries = frappe.get_all(
"Stock Reservation Entry",
filters={"voucher_no": self.name, "voucher_type": "Work Order", "docstatus": 1},
fields=["name", "item_code", "warehouse", "reservation_based_on"],
order_by="creation",
)
for entry in entries:
if entry.reservation_based_on == "Serial and Batch":
continue
key = (entry.item_code, entry.warehouse)
sre = frappe.get_doc("Stock Reservation Entry", entry.name)
# Cap at what is still reservable (qty not already delivered/consumed). Always set the
# value -- including back to 0 when nothing (or less) is now sent -- so cancelling a
# transfer restores the reservation.
available = flt(sre.reserved_qty) - flt(sre.consumed_qty) - flt(sre.delivered_qty)
qty_to_set = max(min(flt(sent.get(key, 0.0)), available), 0.0)
if key in sent:
sent[key] = flt(sent[key]) - qty_to_set
if flt(sre.transferred_qty) == qty_to_set:
continue
sre.db_set("transferred_qty", qty_to_set, update_modified=False)
sre.update_status()
sre.update_reserved_stock_in_bin()
def _subcontract_transferred_qty_by_item(self):
"""Qty sent to subcontractors for this Work Order, keyed by (item_code, source warehouse).
The transfer Stock Entries are linked to the Work Order through its subcontracted Job Cards
(Job Card -> Subcontracting Order / Purchase Order -> Send to Subcontractor entry), since the
entry itself does not retain ``work_order``. Only submitted (docstatus 1) entries contribute,
so a cancelled transfer drops out and the reservation is restored on the next recompute.
"""
job_cards = frappe.get_all(
"Job Card", filters={"work_order": self.name, "is_subcontracted": 1}, pluck="name"
)
if not job_cards:
return {}
sco_names = frappe.get_all(
"Subcontracting Order Item", filters={"job_card": ["in", job_cards]}, pluck="parent"
)
po_names = frappe.get_all(
"Purchase Order Item", filters={"job_card": ["in", job_cards]}, pluck="parent"
)
if not sco_names and not po_names:
return {}
ste = frappe.qb.DocType("Stock Entry")
ste_child = frappe.qb.DocType("Stock Entry Detail")
link = None
if sco_names:
link = ste.subcontracting_order.isin(list(set(sco_names)))
if po_names:
po_link = ste.purchase_order.isin(list(set(po_names)))
link = po_link if link is None else (link | po_link)
rows = (
frappe.qb.from_(ste)
.inner_join(ste_child)
.on(ste_child.parent == ste.name)
.select(ste_child.item_code, ste_child.s_warehouse, fn.Sum(ste_child.transfer_qty).as_("qty"))
.where(
(ste.docstatus == 1)
& (ste.purpose == "Send to Subcontractor")
& (ste_child.s_warehouse.isnotnull())
& link
)
.groupby(ste_child.item_code, ste_child.s_warehouse)
).run(as_dict=1)
return {(d.item_code, d.s_warehouse): flt(d.qty) for d in rows}
def remove_additional_items(self, stock_entry):
for row in stock_entry.items:
for item in self.required_items:

View File

@@ -538,6 +538,9 @@ class StockEntry(StockController, SubcontractingInwardController):
self.update_disassembled_order()
self.adjust_stock_reservation_entries_for_return()
self.update_stock_reservation_entries()
# Release the Work Order's own reservation for items being sent to the subcontractor
# before the negative-stock guard runs in update_stock_ledger().
self.update_wo_reservation_for_subcontracting()
self.update_stock_ledger()
self.make_stock_reserve_for_wip_and_fg()
self.reserve_stock_for_subcontracting()
@@ -589,6 +592,8 @@ class StockEntry(StockController, SubcontractingInwardController):
self.update_quality_inspection()
self.adjust_stock_reservation_entries_for_return()
self.update_stock_reservation_entries()
# Recompute (now excludes this cancelled entry) so the freed reservation is restored.
self.update_wo_reservation_for_subcontracting()
self.delete_auto_created_batches()
self.delete_linked_stock_entry()
@@ -2376,6 +2381,52 @@ class StockEntry(StockController, SubcontractingInwardController):
return False
def update_wo_reservation_for_subcontracting(self):
# A "Send to Subcontractor" entry never keeps its `work_order` (validate clears it for this
# 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.
if self.purpose != "Send to Subcontractor":
return
for wo_name in self.get_reserved_work_orders_for_subcontracting():
frappe.get_doc("Work Order", wo_name).release_reserved_qty_for_subcontract_transfer()
def get_reserved_work_orders_for_subcontracting(self):
job_cards = set()
if self.subcontracting_order:
job_cards.update(
frappe.get_all(
"Subcontracting Order Item",
filters={"parent": self.subcontracting_order},
pluck="job_card",
)
)
if self.purchase_order:
job_cards.update(
frappe.get_all(
"Purchase Order Item", filters={"parent": self.purchase_order}, pluck="job_card"
)
)
job_cards = {jc for jc in job_cards if jc}
if not job_cards:
return []
work_orders = frappe.get_all(
"Job Card", filters={"name": ["in", list(job_cards)]}, pluck="work_order"
)
reserved_work_orders = []
for work_order in set(work_orders):
if not work_order:
continue
if frappe.get_cached_value("Work Order", work_order, "reserve_stock"):
reserved_work_orders.append(work_order)
return reserved_work_orders
@frappe.whitelist()
def get_item_details(self, args: ItemDetailsCtx = None, for_update=False):
item = frappe.qb.DocType("Item")