refactor(stock): extract QualityInspectionService from StockController

Move quality-inspection validation (validate_inspection + validate_qi_presence/
submission/rejection) into erpnext/stock/services/quality_inspection.py as a
delegating service. validate_inspection keeps a controller delegator (called from
validate() and 3 other doctypes); the three row-level helpers are internal-only.
The whitelisted module fns check_item_quality_inspection / make_quality_inspections
stay in stock_controller (stable endpoint paths).

Behaviour-preserving: ledger snapshots + quality inspection suite stay green.
This commit is contained in:
Nabin Hait
2026-06-04 15:54:34 +05:30
parent b447cbc3c1
commit 926bdf5a20
2 changed files with 112 additions and 85 deletions

View File

@@ -438,92 +438,9 @@ class StockController(AccountsController):
)
def validate_inspection(self):
"""Checks if quality inspection is set/ is valid for Items that require inspection."""
inspection_fieldname_map = {
"Purchase Receipt": "inspection_required_before_purchase",
"Purchase Invoice": "inspection_required_before_purchase",
"Subcontracting Receipt": "inspection_required_before_purchase",
"Sales Invoice": "inspection_required_before_delivery",
"Delivery Note": "inspection_required_before_delivery",
}
inspection_required_fieldname = inspection_fieldname_map.get(self.doctype)
from erpnext.stock.services.quality_inspection import QualityInspectionService
# return if inspection is not required on document level
if (
(not inspection_required_fieldname and self.doctype != "Stock Entry")
or (self.doctype == "Stock Entry" and not self.inspection_required)
or (self.doctype in ["Sales Invoice", "Purchase Invoice"] and not self.update_stock)
):
return
for row in self.get("items"):
qi_required = False
if inspection_required_fieldname and frappe.get_cached_value(
"Item", row.item_code, inspection_required_fieldname
):
qi_required = True
elif self.doctype == "Stock Entry" and row.t_warehouse:
qi_required = True # inward stock needs inspection
if row.get("secondary_item_type") or row.get("is_legacy_scrap_item"):
continue
if qi_required: # validate row only if inspection is required on item level
if self.doctype in [
"Purchase Receipt",
"Purchase Invoice",
"Sales Invoice",
"Delivery Note",
] and frappe.get_single_value(
"Stock Settings", "allow_to_make_quality_inspection_after_purchase_or_delivery"
):
return
self.validate_qi_presence(row)
if self.docstatus == 1:
self.validate_qi_submission(row)
self.validate_qi_rejection(row)
def validate_qi_presence(self, row):
"""Check if QI is present on row level. Warn on save and stop on submit if missing."""
if not row.quality_inspection:
msg = _("Row #{0}: Quality Inspection is required for Item {1}").format(
row.idx, frappe.bold(row.item_code)
)
if self.docstatus == 1:
frappe.throw(msg, title=_("Inspection Required"), exc=QualityInspectionRequiredError)
else:
frappe.msgprint(msg, title=_("Inspection Required"), indicator="blue")
def validate_qi_submission(self, row):
"""Check if QI is submitted on row level, during submission"""
action = frappe.get_single_value("Stock Settings", "action_if_quality_inspection_is_not_submitted")
qa_docstatus = frappe.db.get_value("Quality Inspection", row.quality_inspection, "docstatus")
if qa_docstatus != 1:
link = frappe.utils.get_link_to_form("Quality Inspection", row.quality_inspection)
msg = _("Row #{0}: Quality Inspection {1} is not submitted for the item: {2}").format(
row.idx, link, row.item_code
)
if action == "Stop":
frappe.throw(msg, title=_("Inspection Submission"), exc=QualityInspectionNotSubmittedError)
else:
frappe.msgprint(msg, alert=True, indicator="orange")
def validate_qi_rejection(self, row):
"""Check if QI is rejected on row level, during submission"""
action = frappe.get_single_value("Stock Settings", "action_if_quality_inspection_is_rejected")
qa_status = frappe.db.get_value("Quality Inspection", row.quality_inspection, "status")
if qa_status == "Rejected":
link = frappe.utils.get_link_to_form("Quality Inspection", row.quality_inspection)
msg = _("Row #{0}: Quality Inspection {1} was rejected for item {2}").format(
row.idx, link, row.item_code
)
if action == "Stop":
frappe.throw(msg, title=_("Inspection Rejected"), exc=QualityInspectionRejectedError)
else:
frappe.msgprint(msg, alert=True, indicator="orange")
return QualityInspectionService(self).validate_inspection()
def update_blanket_order(self):
blanket_orders = list(set([d.blanket_order for d in self.items if d.blanket_order]))

View File

@@ -0,0 +1,110 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
"""Quality inspection validation for stock transactions.
Extracted from ``StockController``. Validates that items requiring quality
inspection have a present / submitted / non-rejected Quality Inspection.
"""
import frappe
from frappe import _
class QualityInspectionService:
def __init__(self, doc) -> None:
self.doc = doc
def validate_inspection(self):
"""Checks if quality inspection is set/ is valid for Items that require inspection."""
inspection_fieldname_map = {
"Purchase Receipt": "inspection_required_before_purchase",
"Purchase Invoice": "inspection_required_before_purchase",
"Subcontracting Receipt": "inspection_required_before_purchase",
"Sales Invoice": "inspection_required_before_delivery",
"Delivery Note": "inspection_required_before_delivery",
}
inspection_required_fieldname = inspection_fieldname_map.get(self.doc.doctype)
# return if inspection is not required on document level
if (
(not inspection_required_fieldname and self.doc.doctype != "Stock Entry")
or (self.doc.doctype == "Stock Entry" and not self.doc.inspection_required)
or (self.doc.doctype in ["Sales Invoice", "Purchase Invoice"] and not self.doc.update_stock)
):
return
for row in self.doc.get("items"):
qi_required = False
if inspection_required_fieldname and frappe.get_cached_value(
"Item", row.item_code, inspection_required_fieldname
):
qi_required = True
elif self.doc.doctype == "Stock Entry" and row.t_warehouse:
qi_required = True # inward stock needs inspection
if row.get("secondary_item_type") or row.get("is_legacy_scrap_item"):
continue
if qi_required: # validate row only if inspection is required on item level
if self.doc.doctype in [
"Purchase Receipt",
"Purchase Invoice",
"Sales Invoice",
"Delivery Note",
] and frappe.get_single_value(
"Stock Settings", "allow_to_make_quality_inspection_after_purchase_or_delivery"
):
return
self.validate_qi_presence(row)
if self.doc.docstatus == 1:
self.validate_qi_submission(row)
self.validate_qi_rejection(row)
def validate_qi_presence(self, row):
"""Check if QI is present on row level. Warn on save and stop on submit if missing."""
from erpnext.controllers.stock_controller import QualityInspectionRequiredError
if not row.quality_inspection:
msg = _("Row #{0}: Quality Inspection is required for Item {1}").format(
row.idx, frappe.bold(row.item_code)
)
if self.doc.docstatus == 1:
frappe.throw(msg, title=_("Inspection Required"), exc=QualityInspectionRequiredError)
else:
frappe.msgprint(msg, title=_("Inspection Required"), indicator="blue")
def validate_qi_submission(self, row):
"""Check if QI is submitted on row level, during submission"""
from erpnext.controllers.stock_controller import QualityInspectionNotSubmittedError
action = frappe.get_single_value("Stock Settings", "action_if_quality_inspection_is_not_submitted")
qa_docstatus = frappe.db.get_value("Quality Inspection", row.quality_inspection, "docstatus")
if qa_docstatus != 1:
link = frappe.utils.get_link_to_form("Quality Inspection", row.quality_inspection)
msg = _("Row #{0}: Quality Inspection {1} is not submitted for the item: {2}").format(
row.idx, link, row.item_code
)
if action == "Stop":
frappe.throw(msg, title=_("Inspection Submission"), exc=QualityInspectionNotSubmittedError)
else:
frappe.msgprint(msg, alert=True, indicator="orange")
def validate_qi_rejection(self, row):
"""Check if QI is rejected on row level, during submission"""
from erpnext.controllers.stock_controller import QualityInspectionRejectedError
action = frappe.get_single_value("Stock Settings", "action_if_quality_inspection_is_rejected")
qa_status = frappe.db.get_value("Quality Inspection", row.quality_inspection, "status")
if qa_status == "Rejected":
link = frappe.utils.get_link_to_form("Quality Inspection", row.quality_inspection)
msg = _("Row #{0}: Quality Inspection {1} was rejected for item {2}").format(
row.idx, link, row.item_code
)
if action == "Stop":
frappe.throw(msg, title=_("Inspection Rejected"), exc=QualityInspectionRejectedError)
else:
frappe.msgprint(msg, alert=True, indicator="orange")