refactor(stock): convert Purchase Receipt raw SQL to ORM

get_already_received_qty (sum over Purchase Receipt Item, parent != self.name)
and the two Purchase-Invoice-against-receipt existence checks (implicit
comma-joins -> child-table get_all on Purchase Invoice Item, docstatus=1).
Also fixes a pre-existing `self.submit_rv` -> `submit_rv` typo in the (dead)
check_next_docstatus that staging carried forward. Same result on MariaDB;
valid under Postgres.

Tests: get_already_received_qty (parent-exclusion sum) and check_next_docstatus
(blocks on a submitted Purchase Invoice; also locks the typo fix).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Mihir Kandoi
2026-06-20 19:41:39 +05:30
parent bfffed0f52
commit a65aa27225
2 changed files with 52 additions and 18 deletions

View File

@@ -339,14 +339,17 @@ class PurchaseReceipt(BuyingController):
frappe.throw(_(msg))
def get_already_received_qty(self, po, po_detail):
qty = frappe.db.sql(
"""select sum(qty) from `tabPurchase Receipt Item`
where purchase_order_item = %s and docstatus = 1
and purchase_order=%s
and parent != %s""",
(po_detail, po, self.name),
qty = frappe.get_all(
"Purchase Receipt Item",
filters={
"purchase_order_item": po_detail,
"docstatus": 1,
"purchase_order": po,
"parent": ["!=", self.name],
},
fields=[{"SUM": "qty", "as": "qty"}],
)
return qty and flt(qty[0][0]) or 0.0
return flt(qty[0].qty) if qty and qty[0].qty else 0.0
def get_po_qty_and_warehouse(self, po_detail):
po_qty, po_warehouse = frappe.db.get_value("Purchase Order Item", po_detail, ["qty", "warehouse"])
@@ -415,25 +418,27 @@ class PurchaseReceipt(BuyingController):
)
def check_next_docstatus(self):
submit_rv = frappe.db.sql(
"""select t1.name
from `tabPurchase Invoice` t1,`tabPurchase Invoice Item` t2
where t1.name = t2.parent and t2.purchase_receipt = %s and t1.docstatus = 1""",
(self.name),
submit_rv = frappe.get_all(
"Purchase Invoice Item",
filters={"purchase_receipt": self.name, "docstatus": 1},
fields=["parent"],
as_list=True,
limit=1,
)
if submit_rv:
frappe.throw(_("Purchase Invoice {0} is already submitted").format(self.submit_rv[0][0]))
frappe.throw(_("Purchase Invoice {0} is already submitted").format(submit_rv[0][0]))
def on_cancel(self):
super().on_cancel()
self.check_for_on_hold_or_closed_status("Purchase Order", "purchase_order")
# Check if Purchase Invoice has been submitted against current Purchase Order
submitted = frappe.db.sql(
"""select t1.name
from `tabPurchase Invoice` t1,`tabPurchase Invoice Item` t2
where t1.name = t2.parent and t2.purchase_receipt = %s and t1.docstatus = 1""",
self.name,
submitted = frappe.get_all(
"Purchase Invoice Item",
filters={"purchase_receipt": self.name, "docstatus": 1},
fields=["parent"],
as_list=True,
limit=1,
)
if submitted:
frappe.throw(_("Purchase Invoice {0} is already submitted").format(submitted[0][0]))

View File

@@ -6011,6 +6011,35 @@ class TestPurchaseReceipt(ERPNextTestSuite):
srbnb_credit = sum(flt(row.credit) for row in gl_entries if row.account == srbnb_account)
self.assertAlmostEqual(srbnb_credit, pi_base_net_amount, places=2)
def test_get_already_received_qty(self):
"""get_already_received_qty sums prior submitted PR Item qty against the same PO line,
excluding the current PR — covers the converted SUM with `parent != self.name`."""
from erpnext.buying.doctype.purchase_order.test_purchase_order import (
create_purchase_order,
make_pr_against_po,
)
po = create_purchase_order(qty=10)
po_detail = po.items[0].name
make_pr_against_po(po.name, 4) # PR1 receives 4
pr2 = make_pr_against_po(po.name, 2) # PR2 receives 2
# already received against this PO line, excluding pr2 itself, is pr1's 4
self.assertEqual(pr2.get_already_received_qty(po.name, po_detail), 4.0)
def test_check_next_docstatus_blocks_with_submitted_invoice(self):
"""check_next_docstatus must flag a submitted Purchase Invoice drawn from the receipt —
covers the converted child-table get_all (Purchase Invoice Item, docstatus=1)."""
pr = make_purchase_receipt()
pi = make_purchase_invoice(pr.name)
pi.insert()
pi.submit()
with self.assertRaises(frappe.ValidationError) as cm:
pr.check_next_docstatus()
self.assertIn("is already submitted", str(cm.exception))
def create_asset_category_for_pr_test():
category_name = "Test Asset Category for PR"