From 4719ad9b91b8cd377b409ab025a71d5ffb81de2b Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Tue, 8 Sep 2026 09:41:24 +0530 Subject: [PATCH] fix(manufacturing): apply safety stock once across Production Plan rows (v16) (#58832) --- .../production_plan/production_plan.py | 61 ++++--- .../production_plan/test_production_plan.py | 163 ++++++++++++++++++ 2 files changed, 197 insertions(+), 27 deletions(-) diff --git a/erpnext/manufacturing/doctype/production_plan/production_plan.py b/erpnext/manufacturing/doctype/production_plan/production_plan.py index cb1e2a14a98..b46709e4b38 100644 --- a/erpnext/manufacturing/doctype/production_plan/production_plan.py +++ b/erpnext/manufacturing/doctype/production_plan/production_plan.py @@ -1450,42 +1450,18 @@ def get_material_request_items( bin_dict, consumed_qty, ): - required_qty = 0 - item_code = row.get("item_code") - - if not ignore_existing_ordered_qty or bin_dict.get("projected_qty", 0) < 0: - required_qty = flt(row.get("qty")) - else: - key = (item_code, warehouse) - available_qty = flt(bin_dict.get("projected_qty", 0)) - consumed_qty[key] - if available_qty > 0: - required_qty = max(0, flt(row.get("qty")) - available_qty) - consumed_qty[key] += min(flt(row.get("qty")), available_qty) - else: - required_qty = flt(row.get("qty")) + required_qty = _required_qty_for_mr( + row, ignore_existing_ordered_qty, warehouse, bin_dict, consumed_qty, include_safety_stock + ) if doc.get("consider_minimum_order_qty") and required_qty > 0 and required_qty < row["min_order_qty"]: required_qty = row["min_order_qty"] item_group_defaults = get_item_group_defaults(row.item_code, company) - if not row["purchase_uom"]: - row["purchase_uom"] = row["stock_uom"] - - if row["purchase_uom"] != row["stock_uom"]: - if not (row["conversion_factor"] or frappe.flags.show_qty_in_stock_uom): - frappe.throw( - _("UOM Conversion factor ({0} -> {1}) not found for item: {2}").format( - row["purchase_uom"], row["stock_uom"], row.item_code - ) - ) - if frappe.db.get_value("UOM", row["purchase_uom"], "must_be_whole_number"): required_qty = ceil(required_qty) - if include_safety_stock: - required_qty += flt(row["safety_stock"]) - item_details = frappe.get_cached_value("Item", row.item_code, ["purchase_uom", "stock_uom"], as_dict=1) conversion_factor = 1.0 @@ -1606,6 +1582,37 @@ def get_sales_orders(self): return open_so +def _required_qty_for_mr( + row, ignore_existing_ordered_qty, warehouse, bin_dict, consumed_qty, include_safety_stock +): + safety_stock = flt(row["safety_stock"]) if include_safety_stock else 0 + qty = flt(row.get("qty")) + projected_qty = max(0, flt(bin_dict.get("projected_qty"))) if ignore_existing_ordered_qty else 0 + + key = (row.get("item_code"), warehouse) + available_qty = projected_qty - consumed_qty[key] + required_qty = max(0, qty - (available_qty - safety_stock)) + consumed_qty[key] += qty - required_qty + return _adjust_required_qty_for_uom(row, required_qty) + + +def _adjust_required_qty_for_uom(row, required_qty): + if not row["purchase_uom"]: + row["purchase_uom"] = row["stock_uom"] + + if row["purchase_uom"] != row["stock_uom"]: + if not (row["conversion_factor"] or frappe.flags.show_qty_in_stock_uom): + frappe.throw( + _("UOM Conversion factor ({0} -> {1}) not found for item: {2}").format( + row["purchase_uom"], row["stock_uom"], row.item_code + ) + ) + + if frappe.db.get_value("UOM", row["purchase_uom"], "must_be_whole_number"): + required_qty = ceil(required_qty) + return required_qty + + @frappe.whitelist() def get_bin_details(row, company, for_warehouse=None, all_warehouse=False): if isinstance(row, str): diff --git a/erpnext/manufacturing/doctype/production_plan/test_production_plan.py b/erpnext/manufacturing/doctype/production_plan/test_production_plan.py index 369ba0b7ef6..9053c03de85 100644 --- a/erpnext/manufacturing/doctype/production_plan/test_production_plan.py +++ b/erpnext/manufacturing/doctype/production_plan/test_production_plan.py @@ -110,6 +110,169 @@ class TestProductionPlan(ERPNextTestSuite): pln = frappe.get_doc("Production Plan", pln.name) pln.cancel() + def _plan_for_safety_stock(self, rm_item, qty_per_order, bom_quantity=1): + fg_item = make_item(properties={"is_stock_item": 1}).name + make_bom( + item=fg_item, + raw_materials=[rm_item], + source_warehouse="_Test Warehouse - _TC", + quantity=bom_quantity, + ) + + pln = create_production_plan( + item_code=fg_item, + ignore_existing_ordered_qty=1, + do_not_save=1, + skip_getting_mr_items=1, + ) + pln.get_items_from = "Sales Order" + for _ in range(2): + so = make_sales_order(item_code=fg_item, qty=qty_per_order) + pln.append( + "sales_orders", + { + "sales_order": so.name, + "sales_order_date": so.transaction_date, + "customer": so.customer, + "grand_total": so.grand_total, + }, + ) + pln.get_items() + return pln + + def test_safety_stock_added_once_for_repeated_raw_material(self): + rm_item = make_item(properties={"is_stock_item": 1, "safety_stock": 10, "valuation_rate": 100}).name + make_stock_entry(item_code=rm_item, qty=100, rate=100, target="_Test Warehouse - _TC") + + pln = self._plan_for_safety_stock(rm_item, qty_per_order=50) + pln.include_safety_stock = 1 + + items = get_items_for_material_requests(pln.as_dict()) + quantities = sorted(flt(d.get("quantity")) for d in items if d.get("item_code") == rm_item) + self.assertEqual(quantities, [0, 10]) + + def test_safety_stock_added_once_with_negative_or_ignored_projected_qty(self): + from erpnext.stock.utils import get_or_make_bin + + rm_item = make_item(properties={"is_stock_item": 1, "safety_stock": 100}).name + bin_name = get_or_make_bin(rm_item, "_Test Warehouse - _TC") + pln = self._plan_for_safety_stock(rm_item, qty_per_order=250) + pln.po_items[1].planned_qty = 1000 + pln.include_safety_stock = 1 + + for projected_qty in (-5, 0, 200, 1500): + frappe.db.set_value("Bin", bin_name, "projected_qty", projected_qty) + for consider_projected_qty in (0, 1): + with self.subTest(projected_qty=projected_qty, consider_projected_qty=consider_projected_qty): + pln.ignore_existing_ordered_qty = consider_projected_qty + items = get_items_for_material_requests(pln.as_dict()) + expected_qty = [350, 1000] + if consider_projected_qty and projected_qty > 0: + expected_qty = [150, 1000] if projected_qty == 200 else [0, 0] + self.assertEqual([row["quantity"] for row in items], expected_qty) + self.assertEqual([row["required_bom_qty"] for row in items], [250, 1000]) + self.assertEqual([row["safety_stock"] for row in items], [100, 100]) + self.assertEqual( + [row["sales_order"] for row in items], [row.sales_order for row in pln.po_items] + ) + + def test_safety_stock_disabled_with_negative_projected_qty(self): + from erpnext.stock.utils import get_or_make_bin + + rm_item = make_item(properties={"is_stock_item": 1, "safety_stock": 100}).name + bin_name = get_or_make_bin(rm_item, "_Test Warehouse - _TC") + frappe.db.set_value("Bin", bin_name, "projected_qty", -5) + pln = self._plan_for_safety_stock(rm_item, qty_per_order=250) + pln.po_items[1].planned_qty = 1000 + + items = get_items_for_material_requests(pln.as_dict()) + self.assertEqual([row["quantity"] for row in items], [250, 1000]) + + def test_safety_stock_is_separate_for_each_item_and_warehouse(self): + from collections import defaultdict + + from erpnext.manufacturing.doctype.production_plan.production_plan import ( + _required_qty_for_mr, + ) + + row = frappe._dict(qty=250, safety_stock=100, purchase_uom="Nos", stock_uom="Nos") + items_and_warehouses = [ + ("Raw Material Item 1", "_Test Warehouse - _TC"), + ("Raw Material Item 1", "_Test Warehouse 1 - _TC"), + ("Raw Material Item 2", "_Test Warehouse - _TC"), + ] + for consider_projected_qty in (0, 1): + with self.subTest(consider_projected_qty=consider_projected_qty): + consumed_qty = defaultdict(float) + quantities = [] + for item_code, warehouse in items_and_warehouses * 2: + row.item_code = item_code + quantities.append( + _required_qty_for_mr( + row, consider_projected_qty, warehouse, {"projected_qty": -5}, consumed_qty, True + ) + ) + self.assertEqual(quantities, [350, 350, 350, 250, 250, 250]) + + def test_safety_stock_added_once_before_transferring_materials(self): + from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse + from erpnext.stock.utils import get_or_make_bin + + rm_item = make_item(properties={"is_stock_item": 1, "safety_stock": 100, "min_order_qty": 1234}).name + source_warehouse = create_warehouse("Safety Stock Source Warehouse", company="_Test Company") + make_stock_entry(item_code=rm_item, qty=2000, rate=100, target=source_warehouse) + bin_name = get_or_make_bin(rm_item, "_Test Warehouse - _TC") + frappe.db.set_value("Bin", bin_name, "projected_qty", -5) + pln = self._plan_for_safety_stock(rm_item, qty_per_order=250) + pln.po_items[1].planned_qty = 1000 + pln.for_warehouse = "_Test Warehouse - _TC" + pln.include_safety_stock = 1 + + items = get_items_for_material_requests(pln.as_dict(), warehouses=[{"warehouse": source_warehouse}]) + self.assertEqual([row["material_request_type"] for row in items], ["Material Transfer"] * 2) + self.assertEqual([row["quantity"] for row in items], [350, 1000]) + + def test_safety_stock_does_not_share_purchase_rounding_between_rows(self): + from erpnext.stock.utils import get_or_make_bin + + rm_item = make_item(properties={"is_stock_item": 1, "stock_uom": "Nos", "safety_stock": 1}).name + pln = self._plan_for_safety_stock(rm_item, qty_per_order=1, bom_quantity=2) + bin_name = get_or_make_bin(rm_item, "_Test Warehouse - _TC") + + for projected_qty in (0, 0.25, 0.75): + frappe.db.set_value("Bin", bin_name, "projected_qty", projected_qty) + for include_safety_stock in (0, 1): + for consider_projected_qty in (0, 1): + with self.subTest( + projected_qty=projected_qty, + include_safety_stock=include_safety_stock, + consider_projected_qty=consider_projected_qty, + ): + pln.include_safety_stock = include_safety_stock + pln.ignore_existing_ordered_qty = consider_projected_qty + items = get_items_for_material_requests(pln.as_dict()) + expected_qty = [2, 1] if include_safety_stock else [1, 1] + if consider_projected_qty and projected_qty == 0.75: + expected_qty = [1, 1] if include_safety_stock else [0, 1] + self.assertEqual([row["quantity"] for row in items], expected_qty) + self.assertEqual([row["required_bom_qty"] for row in items], [0.5, 0.5]) + self.assertEqual( + [row["sales_order"] for row in items], [row.sales_order for row in pln.po_items] + ) + + def test_safety_stock_with_fractional_minimum_uses_whole_purchase_uom(self): + rm_item = make_item( + properties={"is_stock_item": 1, "stock_uom": "Nos", "safety_stock": 0.5, "min_order_qty": 2.5} + ).name + pln = self._plan_for_safety_stock(rm_item, qty_per_order=1) + pln.set("po_items", [pln.po_items[0]]) + pln.include_safety_stock = 1 + pln.consider_minimum_order_qty = 1 + + items = get_items_for_material_requests(pln.as_dict()) + self.assertEqual(len(items), 1) + self.assertEqual(items[0]["quantity"], 3) + def test_production_plan_start_date(self): "Test if Work Order has same Planned Start Date as Prod Plan." planned_date = add_to_date(date=None, days=3)