From 961cbc3625f158a556c9e8c241f1e6aa44de3c9f Mon Sep 17 00:00:00 2001 From: Rohit Waghchaure Date: Thu, 21 May 2026 07:39:27 +0530 Subject: [PATCH] refactor: using agentic AI --- .../stock/doctype/stock_entry/stock_entry.js | 4 +- .../stock/doctype/stock_entry/stock_entry.py | 15 - .../stock_entry/stock_entry_handler/base.py | 41 ++ .../stock_entry_handler/disassemble.py | 32 +- .../stock_entry_handler/manufacturing.py | 131 +++--- .../material_receipt_issue.py | 23 +- .../stock_entry_handler/material_transfer.py | 71 +--- .../stock_entry_handler/serial_batch.py | 5 +- .../stock_entry_handler/subcontracting.py | 23 +- .../doctype/stock_entry/test_stock_entry.py | 384 ++++++++++++++++++ 10 files changed, 528 insertions(+), 201 deletions(-) create mode 100644 erpnext/stock/doctype/stock_entry/stock_entry_handler/base.py diff --git a/erpnext/stock/doctype/stock_entry/stock_entry.js b/erpnext/stock/doctype/stock_entry/stock_entry.js index a6448505065..3e5a8a10a8d 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry.js +++ b/erpnext/stock/doctype/stock_entry/stock_entry.js @@ -939,7 +939,7 @@ frappe.ui.form.on("Stock Entry", { if (frm.doc.purchase_order) { frm.set_value("subcontracting_order", ""); erpnext.utils.map_current_doc({ - method: "erpnext.stock.doctype.stock_entry.stock_entry.get_items_from_subcontract_order", + method: "erpnext.stock.doctype.stock_entry.stock_entry_handler.subcontracting.get_items_from_subcontract_order", source_name: frm.doc.purchase_order, target_doc: frm, freeze: true, @@ -951,7 +951,7 @@ frappe.ui.form.on("Stock Entry", { if (frm.doc.subcontracting_order) { frm.set_value("purchase_order", ""); erpnext.utils.map_current_doc({ - method: "erpnext.stock.doctype.stock_entry.stock_entry.get_items_from_subcontract_order", + method: "erpnext.stock.doctype.stock_entry.stock_entry_handler.subcontracting.get_items_from_subcontract_order", source_name: frm.doc.subcontracting_order, target_doc: frm, freeze: true, diff --git a/erpnext/stock/doctype/stock_entry/stock_entry.py b/erpnext/stock/doctype/stock_entry/stock_entry.py index 7ec1eede5c2..1daad3b0454 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry.py @@ -1722,18 +1722,3 @@ def get_warehouse_details(args: str | dict): "basic_rate": get_incoming_rate(args), } return ret - - -@frappe.whitelist() -def get_items_from_subcontract_order(source_name: str, target_doc: str | Document | None = None): - from erpnext.controllers.subcontracting_controller import make_rm_stock_entry - - if isinstance(target_doc, str): - target_doc = frappe.get_doc(json.loads(target_doc)) - - order_doctype = "Purchase Order" if target_doc.purchase_order else "Subcontracting Order" - target_doc = make_rm_stock_entry( - subcontract_order=source_name, order_doctype=order_doctype, target_doc=target_doc - ) - - return target_doc diff --git a/erpnext/stock/doctype/stock_entry/stock_entry_handler/base.py b/erpnext/stock/doctype/stock_entry/stock_entry_handler/base.py new file mode 100644 index 00000000000..4706be97914 --- /dev/null +++ b/erpnext/stock/doctype/stock_entry/stock_entry_handler/base.py @@ -0,0 +1,41 @@ +import frappe +from frappe import _ +from frappe.utils import flt + +from erpnext.manufacturing.doctype.bom.bom import get_backflush_based_on + + +class BaseStockEntry: + """Shared foundation for all stock entry purpose handlers. + + Provides common lazy-loaded work order document, backflush configuration, + and work order status validation used across multiple handler classes. + """ + + def __init__(self, se_doc): + self.doc = se_doc + + @property + def wo_doc(self): + if not getattr(self, "_wo_doc", None): + if self.doc.work_order: + self._wo_doc = frappe.get_doc("Work Order", self.doc.work_order) + return getattr(self, "_wo_doc", None) + + @property + def backflush_based_on(self): + return get_backflush_based_on(self.doc.bom_no) + + def _validate_work_order(self): + if not self.wo_doc: + return + + msg = "" + if flt(self.wo_doc.docstatus) != 1: + msg = _("Work Order {0} must be submitted").format(self.doc.work_order) + + if self.wo_doc.status == "Stopped": + msg = _("Transaction not allowed against stopped Work Order {0}").format(self.doc.work_order) + + if msg: + frappe.throw(msg) diff --git a/erpnext/stock/doctype/stock_entry/stock_entry_handler/disassemble.py b/erpnext/stock/doctype/stock_entry/stock_entry_handler/disassemble.py index ff9834917b7..7ea9fdc3280 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry_handler/disassemble.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry_handler/disassemble.py @@ -9,13 +9,16 @@ from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos from erpnext.stock.serial_batch_bundle import SerialBatchCreation from erpnext.stock.utils import get_combine_datetime -from .manufacturing import ceil_qty_if_uom_has_whole_number, get_bom_items, get_secondary_items +from .base import BaseStockEntry +from .manufacturing import ( + ceil_qty_if_uom_has_whole_number, + get_bom_items, + get_production_item_details, + get_secondary_items, +) -class DisassembleStockEntry: - def __init__(self, se_doc): - self.doc = se_doc - +class DisassembleStockEntry(BaseStockEntry): def validate(self): self.validate_warehouse() @@ -205,25 +208,8 @@ class DisassembleStockEntry: self.doc.append("items", item_args) - def get_production_item_details(self): - if self.doc.work_order: - production_item = frappe.get_cached_value("Work Order", self.doc.work_order, "production_item") - else: - production_item = frappe.get_cached_value("BOM", self.doc.bom_no, "item") - - item_details = frappe.get_cached_value( - "Item", - production_item, - ["item_name", "item_group", "description", "stock_uom", "name"], - as_dict=1, - ) - - return item_details - def add_finished_goods(self): - # Fininshed good will be removed from source warehouse - - item_details = self.get_production_item_details() + item_details = get_production_item_details(self.doc.work_order, self.doc.bom_no) item_details.update( { diff --git a/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py b/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py index 72011725d6f..f3dcc3b67ca 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py @@ -6,7 +6,7 @@ from frappe import _, bold from frappe.query_builder.functions import Sum from frappe.utils import ceil, cint, flt, get_link_to_form -from erpnext.manufacturing.doctype.bom.bom import add_additional_cost, get_backflush_based_on +from erpnext.manufacturing.doctype.bom.bom import add_additional_cost from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos from erpnext.stock.serial_batch_bundle import ( SerialBatchCreation, @@ -15,13 +15,11 @@ from erpnext.stock.serial_batch_bundle import ( get_serial_or_batch_items, ) +from .base import BaseStockEntry from .serial_batch import create_serial_and_batch_bundle -class BaseManufactureStockEntry: - def __init__(self, se_doc): - self.doc = se_doc - +class BaseManufactureStockEntry(BaseStockEntry): def set_default_warehouse(self): for row in self.doc.items: if ( @@ -64,17 +62,6 @@ class BaseManufactureStockEntry: title=_("Raw Materials Missing"), ) - @property - def wo_doc(self): - if not getattr(self, "_wo_doc", None): - if self.doc.work_order: - self._wo_doc = frappe.get_doc("Work Order", self.doc.work_order) - return getattr(self, "_wo_doc", None) - - @property - def backflush_based_on(self): - return get_backflush_based_on(self.doc.bom_no) - def get_item_dict(self, row): item_args = {} fields = [ @@ -148,23 +135,8 @@ class BaseManufactureStockEntry: (flt(self.doc.process_loss_qty) / flt(self.doc.fg_completed_qty)) * 100 ) - def get_production_item_details(self): - if self.doc.work_order: - production_item = frappe.get_cached_value("Work Order", self.doc.work_order, "production_item") - else: - production_item = frappe.get_cached_value("BOM", self.doc.bom_no, "item") - - item_details = frappe.get_cached_value( - "Item", - production_item, - ["item_name", "item_group", "description", "stock_uom", "name"], - as_dict=1, - ) - - return item_details - def add_finished_goods(self): - item_details = self.get_production_item_details() + item_details = get_production_item_details(self.doc.work_order, self.doc.bom_no) fg_item_qty = flt(self.doc.fg_completed_qty) - flt(self.doc.process_loss_qty) item_details.update( @@ -321,38 +293,7 @@ class ManufactureStockEntry(BaseManufactureStockEntry): if not rm_items: return - precision = frappe.get_precision("Stock Entry Detail", "qty") - bom_items = get_bom_items(self.doc.bom_no, self.doc.use_multi_level_bom) - - for row in bom_items: - row.qty = row.qty * self.doc.fg_completed_qty - if matched_item := self.get_matched_items(row.item_code): - if flt(row.qty, precision) != flt(matched_item.qty, precision): - frappe.throw( - _( - "For the item {0}, the consumed quantity should be {1} according to the BOM {2}." - ).format( - bold(row.item_code), - flt(row.qty), - get_link_to_form("BOM", self.doc.bom_no), - ), - title=_("Incorrect Component Quantity"), - ) - else: - frappe.throw( - _("According to the BOM {0}, the Item '{1}' is missing in the stock entry.").format( - get_link_to_form("BOM", self.doc.bom_no), bold(row.item_code) - ), - title=_("Missing Item"), - ) - - def get_matched_items(self, item_code): - items = [item for item in self.doc.items if item.s_warehouse] - for row in items: - if row.item_code == item_code or row.original_item == item_code: - return row - - return {} + _check_bom_component_qty(self.doc, get_bom_items(self.doc.bom_no, self.doc.use_multi_level_bom)) def validate_work_order(self): if not self.doc.work_order: @@ -746,17 +687,6 @@ class ManufactureStockEntry(BaseManufactureStockEntry): self.update_job_card_and_work_order() def update_job_card_and_work_order(self): - def _validate_work_order(pro_doc): - msg, title = "", "" - if flt(pro_doc.docstatus) != 1: - msg = _("Work Order {0} must be submitted").format(self.doc.work_order) - - if pro_doc.status == "Stopped": - msg = _("Transaction not allowed against stopped Work Order {0}").format(self.doc.work_order) - - if msg: - frappe.throw(_(msg), title=title) - if self.doc.job_card: job_doc = frappe.get_doc("Job Card", self.doc.job_card) job_doc.set_consumed_qty_in_job_card_item(self.doc) @@ -764,7 +694,7 @@ class ManufactureStockEntry(BaseManufactureStockEntry): job_doc.update_work_order() if self.doc.work_order: - _validate_work_order(self.wo_doc) + self._validate_work_order() if self.doc.fg_completed_qty: self.wo_doc.run_method("update_work_order_qty") @@ -827,6 +757,55 @@ class MaterialConsumptionForManufactureStockEntry(ManufactureStockEntry): self.add_raw_materials_based_on_transfer() +def get_production_item_details(work_order=None, bom_no=None): + production_item = ( + frappe.get_cached_value("Work Order", work_order, "production_item") + if work_order + else frappe.get_cached_value("BOM", bom_no, "item") + ) + return frappe.get_cached_value( + "Item", + production_item, + ["item_name", "item_group", "description", "stock_uom", "name"], + as_dict=1, + ) + + +def _check_bom_component_qty(doc, bom_items): + """Validate that stock entry items match BOM quantities.""" + precision = frappe.get_precision("Stock Entry Detail", "qty") + for row in bom_items: + row.qty = row.qty * doc.fg_completed_qty + matched_item = next( + ( + item + for item in doc.items + if item.s_warehouse + and (item.item_code == row.item_code or item.original_item == row.item_code) + ), + None, + ) + if matched_item: + if flt(row.qty, precision) != flt(matched_item.qty, precision): + frappe.throw( + _( + "For the item {0}, the consumed quantity should be {1} according to the BOM {2}." + ).format( + bold(row.item_code), + flt(row.qty), + get_link_to_form("BOM", doc.bom_no), + ), + title=_("Incorrect Component Quantity"), + ) + else: + frappe.throw( + _("According to the BOM {0}, the Item '{1}' is missing in the stock entry.").format( + get_link_to_form("BOM", doc.bom_no), bold(row.item_code) + ), + title=_("Missing Item"), + ) + + def get_bom_items(bom_no, use_multi_level_bom=None, qty=None, fetch_secondary_items=False): if use_multi_level_bom is None: use_multi_level_bom = frappe.get_cached_value("BOM", bom_no, "use_multi_level_bom") diff --git a/erpnext/stock/doctype/stock_entry/stock_entry_handler/material_receipt_issue.py b/erpnext/stock/doctype/stock_entry/stock_entry_handler/material_receipt_issue.py index 8908c0f9730..7ae15ed2ea2 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry_handler/material_receipt_issue.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry_handler/material_receipt_issue.py @@ -2,13 +2,11 @@ import frappe from frappe import _ from frappe.query_builder.functions import Sum +from .base import BaseStockEntry from .manufacturing import get_bom_items -class MaterialReceiptStockEntry: - def __init__(self, se_doc): - self.doc = se_doc - +class MaterialReceiptStockEntry(BaseStockEntry): def before_validate(self): self.set_default_warehouse() @@ -27,10 +25,7 @@ class MaterialReceiptStockEntry: frappe.throw(_("Target Warehouse is required for item {0}").format(row.item_code)) -class BaseMaterialIssueStockEntry: - def __init__(self, se_doc): - self.doc = se_doc - +class BaseMaterialIssueStockEntry(BaseStockEntry): def set_default_warehouse(self): for row in self.doc.items: row.t_warehouse = None @@ -65,12 +60,12 @@ class MaterialIssueStockEntry(BaseMaterialIssueStockEntry): self.doc.append("items", row) -def get_consumed_items(self): - """Get all raw materials consumed through consumption entries""" +def get_consumed_items(work_order): + """Get all raw materials consumed through consumption entries for a work order.""" parent = frappe.qb.DocType("Stock Entry") child = frappe.qb.DocType("Stock Entry Detail") - query = ( + return ( frappe.qb.from_(parent) .join(child) .on(parent.name == child.parent) @@ -82,9 +77,7 @@ def get_consumed_items(self): .where( (parent.docstatus == 1) & (parent.purpose == "Material Consumption for Manufacture") - & (parent.work_order == self.work_order) + & (parent.work_order == work_order) ) .groupby(child.item_code, child.original_item) - ) - - return query.run(as_dict=True) + ).run(as_dict=True) diff --git a/erpnext/stock/doctype/stock_entry/stock_entry_handler/material_transfer.py b/erpnext/stock/doctype/stock_entry/stock_entry_handler/material_transfer.py index 1d7b26653c3..c1255ba4d5b 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry_handler/material_transfer.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry_handler/material_transfer.py @@ -1,17 +1,13 @@ import frappe -from frappe import _, bold +from frappe import _ from frappe.query_builder.functions import Sum -from frappe.utils import cstr, flt, get_link_to_form +from frappe.utils import cstr, flt -from erpnext.manufacturing.doctype.bom.bom import get_backflush_based_on - -from .manufacturing import get_bom_items +from .base import BaseStockEntry +from .manufacturing import _check_bom_component_qty, get_bom_items -class BaseMaterialTransferStockEntry: - def __init__(self, se_doc): - self.doc = se_doc - +class BaseMaterialTransferStockEntry(BaseStockEntry): def set_default_warehouse(self): for row in self.doc.items: if not row.t_warehouse and self.doc.to_warehouse: @@ -69,17 +65,6 @@ class BaseMaterialTransferStockEntry: title=_("Invalid Source and Target Warehouse"), ) - @property - def wo_doc(self): - if not getattr(self, "_wo_doc", None): - if self.doc.work_order: - self._wo_doc = frappe.get_doc("Work Order", self.doc.work_order) - return getattr(self, "_wo_doc", None) - - @property - def backflush_based_on(self): - return get_backflush_based_on(self.doc.bom_no) - class MaterialTransferStockEntry(BaseMaterialTransferStockEntry): def before_validate(self): @@ -120,38 +105,7 @@ class MaterialTransferForManufactureStockEntry(BaseMaterialTransferStockEntry): if not self.doc.fg_completed_qty: return - precision = frappe.get_precision("Stock Entry Detail", "qty") - bom_items = get_bom_items(self.doc.bom_no, self.doc.use_multi_level_bom) - - for row in bom_items: - row.qty = row.qty * self.doc.fg_completed_qty - if matched_item := self.get_matched_items(row.item_code): - if flt(row.qty, precision) != flt(matched_item.qty, precision): - frappe.throw( - _( - "For the item {0}, the consumed quantity should be {1} according to the BOM {2}." - ).format( - bold(row.item_code), - flt(row.qty), - get_link_to_form("BOM", self.doc.bom_no), - ), - title=_("Incorrect Component Quantity"), - ) - else: - frappe.throw( - _("According to the BOM {0}, the Item '{1}' is missing in the stock entry.").format( - get_link_to_form("BOM", self.doc.bom_no), bold(row.item_code) - ), - title=_("Missing Item"), - ) - - def get_matched_items(self, item_code): - items = [item for item in self.doc.items if item.s_warehouse] - for row in items: - if row.item_code == item_code or row.original_item == item_code: - return row - - return {} + _check_bom_component_qty(self.doc, get_bom_items(self.doc.bom_no, self.doc.use_multi_level_bom)) def add_items(self): item_dict = self.get_pending_raw_materials() @@ -305,24 +259,13 @@ class MaterialTransferForManufactureStockEntry(BaseMaterialTransferStockEntry): self.update_job_card_and_work_order() def update_job_card_and_work_order(self): - def _validate_work_order(pro_doc): - msg, title = "", "" - if flt(pro_doc.docstatus) != 1: - msg = _("Work Order {0} must be submitted").format(self.doc.work_order) - - if pro_doc.status == "Stopped": - msg = _("Transaction not allowed against stopped Work Order {0}").format(self.doc.work_order) - - if msg: - frappe.throw(_(msg), title=title) - if self.doc.job_card: job_doc = frappe.get_doc("Job Card", self.doc.job_card) job_doc.set_transferred_qty(update_status=True) job_doc.set_transferred_qty_in_job_card_item(self.doc) if self.doc.work_order: - _validate_work_order(self.wo_doc) + self._validate_work_order() if self.doc.fg_completed_qty: if self.doc.docstatus == 1: diff --git a/erpnext/stock/doctype/stock_entry/stock_entry_handler/serial_batch.py b/erpnext/stock/doctype/stock_entry/stock_entry_handler/serial_batch.py index 0c985f035a9..dae9f5b8eb7 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry_handler/serial_batch.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry_handler/serial_batch.py @@ -8,11 +8,10 @@ from erpnext.manufacturing.doctype.bom.bom import get_backflush_based_on from erpnext.stock.serial_batch_bundle import SerialBatchCreation, get_serial_or_batch_items from erpnext.stock.utils import get_combine_datetime +from .base import BaseStockEntry -class StockEntrySABB: - def __init__(self, se_doc): - self.doc = se_doc +class StockEntrySABB(BaseStockEntry): def make_serial_and_batch_bundle_for_outward(self): serial_or_batch_items = get_serial_or_batch_items(self.doc.items) if not serial_or_batch_items: diff --git a/erpnext/stock/doctype/stock_entry/stock_entry_handler/subcontracting.py b/erpnext/stock/doctype/stock_entry/stock_entry_handler/subcontracting.py index 520d16c02f7..67cf3208fb5 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry_handler/subcontracting.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry_handler/subcontracting.py @@ -1,15 +1,17 @@ +import json + import frappe from frappe import _, bold +from frappe.model.document import Document from frappe.query_builder.functions import Sum from frappe.utils import flt from erpnext.stock.utils import get_bin +from .base import BaseStockEntry -class SendToSubcontractorStockEntry: - def __init__(self, se_doc): - self.doc = se_doc +class SendToSubcontractorStockEntry(BaseStockEntry): def validate(self): self.validate_subcontract_order() @@ -241,3 +243,18 @@ def get_supplied_items( supplied_item.total_supplied_qty = flt(supplied_item.supplied_qty) - flt(supplied_item.returned_qty) return supplied_item_details + + +@frappe.whitelist() +def get_items_from_subcontract_order(source_name: str, target_doc: str | Document | None = None): + from erpnext.controllers.subcontracting_controller import make_rm_stock_entry + + if isinstance(target_doc, str): + target_doc = frappe.get_doc(json.loads(target_doc)) + + order_doctype = "Purchase Order" if target_doc.purchase_order else "Subcontracting Order" + target_doc = make_rm_stock_entry( + subcontract_order=source_name, order_doctype=order_doctype, target_doc=target_doc + ) + + return target_doc diff --git a/erpnext/stock/doctype/stock_entry/test_stock_entry.py b/erpnext/stock/doctype/stock_entry/test_stock_entry.py index 59807e214b1..bb40f47765a 100644 --- a/erpnext/stock/doctype/stock_entry/test_stock_entry.py +++ b/erpnext/stock/doctype/stock_entry/test_stock_entry.py @@ -2495,6 +2495,390 @@ class TestStockEntry(ERPNextTestSuite): self.assertEqual(se.items[2].amount, 5) +class TestStockEntryCoverage(ERPNextTestSuite): + """Tests for functions previously lacking dedicated coverage.""" + + # ── ceil_qty_if_uom_has_whole_number ────────────────────────────────────── + + def test_ceil_qty_rounds_up_for_whole_number_uom(self): + from erpnext.stock.doctype.stock_entry.stock_entry_handler.manufacturing import ( + ceil_qty_if_uom_has_whole_number, + ) + + frappe.set_value("UOM", "Nos", "must_be_whole_number", 1) + self.assertEqual(ceil_qty_if_uom_has_whole_number(2.3, "Nos"), 3) + frappe.set_value("UOM", "Nos", "must_be_whole_number", 0) + + def test_ceil_qty_no_rounding_for_decimal_uom(self): + from erpnext.stock.doctype.stock_entry.stock_entry_handler.manufacturing import ( + ceil_qty_if_uom_has_whole_number, + ) + + frappe.set_value("UOM", "Nos", "must_be_whole_number", 0) + self.assertEqual(ceil_qty_if_uom_has_whole_number(2.3, "Nos"), 2.3) + + # ── get_uom_details ──────────────────────────────────────────────────────── + + def test_get_uom_details_returns_conversion_factor_and_transfer_qty(self): + from erpnext.stock.doctype.stock_entry.stock_entry import get_uom_details + + result = get_uom_details("_Test Item", "Nos", 5) + self.assertEqual(flt(result.get("conversion_factor")), 1.0) + self.assertEqual(flt(result.get("transfer_qty")), 5.0) + + # ── get_warehouse_details ────────────────────────────────────────────────── + + def test_get_warehouse_details_returns_actual_qty_and_rate(self): + import json + + from frappe.utils import nowdate, nowtime + + from erpnext.stock.doctype.stock_entry.stock_entry import get_warehouse_details + + make_stock_entry( + item_code="_Test Item", + target="_Test Warehouse - _TC", + qty=10, + basic_rate=100, + ) + + args = { + "item_code": "_Test Item", + "warehouse": "_Test Warehouse - _TC", + "posting_date": nowdate(), + "posting_time": nowtime(), + } + result = get_warehouse_details(json.dumps(args)) + self.assertGreater(result.get("actual_qty", 0), 0) + self.assertGreater(result.get("basic_rate", 0), 0) + + def test_get_warehouse_details_empty_args_returns_empty_dict(self): + from erpnext.stock.doctype.stock_entry.stock_entry import get_warehouse_details + + self.assertEqual(get_warehouse_details({}), {}) + + # ── get_work_order_details ───────────────────────────────────────────────── + + def test_get_work_order_details_returns_correct_fields(self): + from erpnext.stock.doctype.stock_entry.stock_entry import get_work_order_details + + bom_no = frappe.db.get_value("BOM", {"item": "_Test FG Item 2", "is_default": 1, "docstatus": 1}) + wo = frappe.new_doc("Work Order") + wo.update( + { + "company": "_Test Company", + "fg_warehouse": "_Test Warehouse 1 - _TC", + "production_item": "_Test FG Item 2", + "bom_no": bom_no, + "qty": 3.0, + "stock_uom": "_Test UOM", + "wip_warehouse": "_Test Warehouse - _TC", + } + ) + wo.insert() + wo.submit() + + result = get_work_order_details(wo.name, "_Test Company") + self.assertEqual(result["bom_no"], bom_no) + self.assertEqual(result["fg_completed_qty"], 3.0) + self.assertEqual(result["from_bom"], 1) + self.assertEqual(result["wip_warehouse"], wo.wip_warehouse) + + # ── get_production_item_details ──────────────────────────────────────────── + + def test_get_production_item_details_from_bom(self): + from erpnext.stock.doctype.stock_entry.stock_entry_handler.manufacturing import ( + get_production_item_details, + ) + + bom_no = frappe.db.get_value("BOM", {"item": "_Test FG Item 2", "is_default": 1, "docstatus": 1}) + result = get_production_item_details(bom_no=bom_no) + self.assertEqual(result.name, "_Test FG Item 2") + self.assertIsNotNone(result.stock_uom) + + def test_get_production_item_details_from_work_order(self): + from erpnext.stock.doctype.stock_entry.stock_entry_handler.manufacturing import ( + get_production_item_details, + ) + + bom_no = frappe.db.get_value("BOM", {"item": "_Test FG Item 2", "is_default": 1, "docstatus": 1}) + wo = frappe.new_doc("Work Order") + wo.update( + { + "company": "_Test Company", + "fg_warehouse": "_Test Warehouse 1 - _TC", + "production_item": "_Test FG Item 2", + "bom_no": bom_no, + "qty": 1.0, + "stock_uom": "_Test UOM", + "wip_warehouse": "_Test Warehouse - _TC", + } + ) + wo.insert() + wo.submit() + + result = get_production_item_details(work_order=wo.name) + self.assertEqual(result.name, "_Test FG Item 2") + + # ── get_bom_items ────────────────────────────────────────────────────────── + + def test_get_bom_items_returns_raw_materials_with_structure(self): + from erpnext.stock.doctype.stock_entry.stock_entry_handler.manufacturing import get_bom_items + + bom_no = frappe.db.get_value("BOM", {"item": "_Test FG Item 2", "is_default": 1, "docstatus": 1}) + items = get_bom_items(bom_no) + self.assertGreater(len(items), 0) + for item in items: + self.assertIn("item_code", item) + self.assertIn("qty", item) + + def test_get_bom_items_scales_qty_proportionally(self): + from erpnext.stock.doctype.stock_entry.stock_entry_handler.manufacturing import get_bom_items + + bom_no = frappe.db.get_value("BOM", {"item": "_Test FG Item 2", "is_default": 1, "docstatus": 1}) + items_1 = {i["item_code"]: i["qty"] for i in get_bom_items(bom_no, qty=1)} + items_2 = {i["item_code"]: i["qty"] for i in get_bom_items(bom_no, qty=2)} + for item_code, qty_at_1 in items_1.items(): + self.assertAlmostEqual(items_2[item_code], qty_at_1 * 2, places=4) + + # ── validate_sample_quantity ─────────────────────────────────────────────── + + @ERPNextTestSuite.change_settings( + "Stock Settings", {"sample_retention_warehouse": "_Test Warehouse 1 - _TC"} + ) + def test_validate_sample_quantity_raises_when_sample_exceeds_received_qty(self): + from erpnext.stock.doctype.stock_entry.stock_entry_handler.manufacturing import ( + validate_sample_quantity, + ) + + item = make_item( + "_Sample Qty Excess Item", + {"is_stock_item": 1, "retain_sample": 1, "sample_quantity": 2}, + ) + self.assertRaises(frappe.ValidationError, validate_sample_quantity, item.name, 10, 5) + + # ── get_expired_batches ──────────────────────────────────────────────────── + + def test_get_expired_batches_includes_expired_batch(self): + from erpnext.stock.doctype.batch.test_batch import make_new_batch + from erpnext.stock.doctype.stock_entry.stock_entry_handler.serial_batch import ( + get_expired_batches, + ) + + item = make_item("_Test Expired Batch Item", {"is_stock_item": 1, "has_batch_no": 1}) + batch = make_new_batch( + batch_id=frappe.generate_hash("", 5), + item_code=item.name, + expiry_date=add_days(today(), -1), + ) + + expired = get_expired_batches() + self.assertIn(batch.name, expired) + self.assertEqual(expired[batch.name].item, item.name) + + def test_get_expired_batches_excludes_future_batch(self): + from erpnext.stock.doctype.batch.test_batch import make_new_batch + from erpnext.stock.doctype.stock_entry.stock_entry_handler.serial_batch import ( + get_expired_batches, + ) + + item = make_item("_Test Future Batch Item", {"is_stock_item": 1, "has_batch_no": 1}) + future_batch = make_new_batch( + batch_id=frappe.generate_hash("", 5), + item_code=item.name, + expiry_date=add_days(today(), 10), + ) + + expired = get_expired_batches() + self.assertNotIn(future_batch.name, expired) + + # ── validate_source_stock_entry ──────────────────────────────────────────── + + def test_validate_source_stock_entry_skips_when_no_source(self): + se = frappe.new_doc("Stock Entry") + se.source_stock_entry = None + se.validate_source_stock_entry() # must not raise + + def test_validate_source_stock_entry_throws_on_work_order_mismatch(self): + source_se = make_stock_entry( + item_code="_Test Item", + target="_Test Warehouse - _TC", + qty=1, + basic_rate=100, + ) + frappe.db.set_value("Stock Entry", source_se.name, "work_order", "WO-FAKE-SOURCE-001") + + se = frappe.new_doc("Stock Entry") + se.source_stock_entry = source_se.name + se.work_order = "WO-FAKE-TARGET-999" + + self.assertRaises(frappe.ValidationError, se.validate_source_stock_entry) + + def test_validate_source_stock_entry_passes_with_matching_work_order(self): + source_se = make_stock_entry( + item_code="_Test Item", + target="_Test Warehouse - _TC", + qty=1, + basic_rate=100, + ) + frappe.db.set_value("Stock Entry", source_se.name, "work_order", "WO-SAME-001") + + se = frappe.new_doc("Stock Entry") + se.source_stock_entry = source_se.name + se.work_order = "WO-SAME-001" + se.validate_source_stock_entry() # must not raise + + # ── validate_job_card_fg_item ────────────────────────────────────────────── + + def test_validate_job_card_fg_item_skips_when_no_job_card(self): + se = frappe.new_doc("Stock Entry") + se.job_card = None + se.validate_job_card_fg_item() # must not raise + + def test_validate_job_card_fg_item_throws_when_fg_item_mismatches(self): + wrong_fg = make_item("_JC Wrong FG Item", {"is_stock_item": 1}).name + + jc_name = frappe.db.get_value("Job Card", {"docstatus": 1, "finished_good": ("!=", "")}) + if not jc_name: + return # skip if no suitable job card in test data + + jc = frappe.db.get_value("Job Card", jc_name, ["finished_good"], as_dict=1) + if jc.finished_good == wrong_fg: + return # skip if the wrong_fg happens to match + + se = frappe.new_doc("Stock Entry") + se.job_card = jc_name + se.append("items", {"item_code": wrong_fg, "is_finished_item": 1, "qty": 1}) + self.assertRaises(frappe.ValidationError, se.validate_job_card_fg_item) + + # ── validate_job_card_item ───────────────────────────────────────────────── + + def test_validate_job_card_item_skips_when_no_job_card(self): + se = frappe.new_doc("Stock Entry") + se.job_card = None + se.validate_job_card_item() # must not raise + + def test_validate_job_card_item_skips_for_manufacture_purpose(self): + se = frappe.new_doc("Stock Entry") + se.job_card = "SOME-JC-001" + se.purpose = "Manufacture" + se.validate_job_card_item() # must not raise even with a job card set + + @ERPNextTestSuite.change_settings("Manufacturing Settings", {"job_card_excess_transfer": 0}) + def test_validate_job_card_item_throws_when_job_card_item_ref_missing(self): + jc_name = frappe.db.get_value("Job Card", {"docstatus": 1}) + if not jc_name: + return # skip if no job cards in test data + + se = frappe.new_doc("Stock Entry") + se.job_card = jc_name + se.purpose = "Material Transfer for Manufacture" + se.append( + "items", + { + "item_code": "_Test Item", + "s_warehouse": "_Test Warehouse - _TC", + "qty": 1, + "job_card_item": None, + }, + ) + self.assertRaises(frappe.ValidationError, se.validate_job_card_item) + + # ── get_available_materials ──────────────────────────────────────────────── + + def test_get_available_materials_tracks_transferred_qty(self): + from erpnext.manufacturing.doctype.production_plan.test_production_plan import make_bom + from erpnext.manufacturing.doctype.work_order.work_order import ( + make_stock_entry as _make_stock_entry, + ) + from erpnext.stock.doctype.stock_entry.stock_entry_handler.disassemble import ( + get_available_materials, + ) + + fg = make_item("_AM FG Item", {"is_stock_item": 1}).name + rm = make_item("_AM RM Item", {"is_stock_item": 1}).name + source_wh = "_Test Warehouse - _TC" + wip_wh = "_Test Warehouse 1 - _TC" + + make_stock_entry(item_code=rm, target=source_wh, qty=20, purpose="Material Receipt") + + bom_no = make_bom(item=fg, raw_materials=[rm]).name + wo = frappe.new_doc("Work Order") + wo.update( + { + "company": "_Test Company", + "fg_warehouse": "_Test Warehouse 1 - _TC", + "production_item": fg, + "bom_no": bom_no, + "qty": 2.0, + "stock_uom": "Nos", + "wip_warehouse": wip_wh, + } + ) + wo.insert() + wo.submit() + + transfer_se = frappe.get_doc(_make_stock_entry(wo.name, "Material Transfer for Manufacture", 2)) + for d in transfer_se.items: + d.s_warehouse = source_wh + transfer_se.insert() + transfer_se.submit() + + materials = get_available_materials(wo.name) + self.assertGreater(len(materials), 0) + + key = (rm, wip_wh) + self.assertIn(key, materials) + self.assertGreater(materials[key].qty, 0) + + def test_get_available_materials_reduces_qty_after_consumption(self): + from erpnext.manufacturing.doctype.production_plan.test_production_plan import make_bom + from erpnext.manufacturing.doctype.work_order.work_order import ( + make_stock_entry as _make_stock_entry, + ) + from erpnext.stock.doctype.stock_entry.stock_entry_handler.disassemble import ( + get_available_materials, + ) + + fg = make_item("_AM2 FG Item", {"is_stock_item": 1}).name + rm = make_item("_AM2 RM Item", {"is_stock_item": 1}).name + source_wh = "_Test Warehouse - _TC" + wip_wh = "_Test Warehouse 1 - _TC" + + make_stock_entry(item_code=rm, target=source_wh, qty=20, purpose="Material Receipt") + + bom_no = make_bom(item=fg, raw_materials=[rm]).name + wo = frappe.new_doc("Work Order") + wo.update( + { + "company": "_Test Company", + "fg_warehouse": "_Test Warehouse 1 - _TC", + "production_item": fg, + "bom_no": bom_no, + "qty": 2.0, + "stock_uom": "Nos", + "wip_warehouse": wip_wh, + } + ) + wo.insert() + wo.submit() + + transfer_se = frappe.get_doc(_make_stock_entry(wo.name, "Material Transfer for Manufacture", 2)) + for d in transfer_se.items: + d.s_warehouse = source_wh + transfer_se.insert() + transfer_se.submit() + + manufacture_se = frappe.get_doc(_make_stock_entry(wo.name, "Manufacture", 2)) + manufacture_se.insert() + manufacture_se.submit() + + materials = get_available_materials(wo.name) + key = (rm, wip_wh) + if key in materials: + self.assertEqual(materials[key].qty, 0) + + def make_serialized_item(self, **args): args = frappe._dict(args) se = frappe.copy_doc(self.globalTestRecords["Stock Entry"][0])