mirror of
https://github.com/frappe/erpnext.git
synced 2026-05-27 17:04:47 +00:00
refactor: using agentic AI
This commit is contained in:
Rohit Waghchaure
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
@@ -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(
|
||||
{
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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])
|
||||
|
||||
Reference in New Issue
Block a user