refactor(stock): convert Delivery Trip contact/address lookups to qb

get_default_contact / get_default_address: raw correlated-subquery SELECTs over
Dynamic Link -> frappe.qb with a LEFT join (preserving the original
correlated-subquery semantics: a Dynamic Link whose parent Contact/Address is
missing still returns, with a NULL flag). Same result on MariaDB; valid under
Postgres.

Tests: pin the converted query output (real linked Contact/Address) and lock
the LEFT-join choice with an orphaned-Dynamic-Link case (fails under an inner
join).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Mihir Kandoi
2026-06-19 23:19:57 +05:30
parent 28f6994520
commit 7f81ffca23
2 changed files with 84 additions and 26 deletions

View File

@@ -315,19 +315,15 @@ def get_contact_and_address(name: str):
def get_default_contact(out, name):
contact_persons = frappe.db.sql(
"""
SELECT parent,
(SELECT is_primary_contact FROM tabContact c WHERE c.name = dl.parent) AS is_primary_contact
FROM
`tabDynamic Link` dl
WHERE
dl.link_doctype='Customer'
AND dl.link_name=%s
AND dl.parenttype = 'Contact'
""",
(name),
as_dict=1,
dl = frappe.qb.DocType("Dynamic Link")
contact = frappe.qb.DocType("Contact")
contact_persons = (
frappe.qb.from_(dl)
.left_join(contact)
.on(contact.name == dl.parent)
.select(dl.parent, contact.is_primary_contact)
.where((dl.link_doctype == "Customer") & (dl.link_name == name) & (dl.parenttype == "Contact"))
.run(as_dict=1)
)
if contact_persons:
@@ -341,19 +337,15 @@ def get_default_contact(out, name):
def get_default_address(out, name):
shipping_addresses = frappe.db.sql(
"""
SELECT parent,
(SELECT is_shipping_address FROM tabAddress a WHERE a.name=dl.parent) AS is_shipping_address
FROM
`tabDynamic Link` dl
WHERE
dl.link_doctype='Customer'
AND dl.link_name=%s
AND dl.parenttype = 'Address'
""",
(name),
as_dict=1,
dl = frappe.qb.DocType("Dynamic Link")
address = frappe.qb.DocType("Address")
shipping_addresses = (
frappe.qb.from_(dl)
.left_join(address)
.on(address.name == dl.parent)
.select(dl.parent, address.is_shipping_address)
.where((dl.link_doctype == "Customer") & (dl.link_name == name) & (dl.parenttype == "Address"))
.run(as_dict=1)
)
if shipping_addresses:

View File

@@ -8,6 +8,7 @@ from frappe.utils import add_days, flt, now_datetime, nowdate
import erpnext
from erpnext.stock.doctype.delivery_trip.delivery_trip import (
get_contact_and_address,
get_default_contact,
notify_customers,
)
from erpnext.tests.utils import ERPNextTestSuite
@@ -108,6 +109,71 @@ class TestDeliveryTrip(ERPNextTestSuite):
self.delivery_trip.save()
self.assertEqual(self.delivery_trip.status, "Completed")
def test_get_contact_and_address_returns_linked_contact_and_address(self):
"""get_contact_and_address (the converted Dynamic Link queries) must return a real Contact
and Address that are actually linked to the customer — pins the converted query's output."""
out = get_contact_and_address("_Test Customer")
self.assertTrue(out.contact_person and out.contact_person.parent)
self.assertTrue(frappe.db.exists("Contact", out.contact_person.parent))
self.assertTrue(
frappe.db.exists(
"Dynamic Link",
{
"parenttype": "Contact",
"parent": out.contact_person.parent,
"link_doctype": "Customer",
"link_name": "_Test Customer",
},
)
)
self.assertTrue(out.shipping_address and out.shipping_address.parent)
self.assertTrue(frappe.db.exists("Address", out.shipping_address.parent))
self.assertTrue(
frappe.db.exists(
"Dynamic Link",
{
"parenttype": "Address",
"parent": out.shipping_address.parent,
"link_doctype": "Customer",
"link_name": "_Test Customer",
},
)
)
def test_get_default_contact_keeps_orphaned_dynamic_link(self):
"""The converted get_default_contact uses a LEFT join, matching the original correlated
subquery: a Dynamic Link whose parent Contact no longer exists must STILL be returned
(is_primary_contact NULL). An inner join would silently drop it and return None."""
customer = "_Test Customer 2"
# A Contact linked to the customer, then orphan its Dynamic Link by deleting the Contact row.
contact = frappe.get_doc(
{
"doctype": "Contact",
"first_name": "_Test Orphan Link Contact",
"links": [{"link_doctype": "Customer", "link_name": customer}],
}
).insert()
orphan_parent = contact.name
frappe.db.delete("Contact", {"name": orphan_parent})
self.assertFalse(frappe.db.exists("Contact", orphan_parent))
self.assertTrue(
frappe.db.exists(
"Dynamic Link",
{"parenttype": "Contact", "parent": orphan_parent, "link_name": customer},
)
)
out = frappe._dict()
result = get_default_contact(out, customer)
# LEFT join keeps the orphaned-link row; an inner join would have returned None.
self.assertIsNotNone(result)
self.assertEqual(result.parent, orphan_parent)
self.assertIsNone(result.is_primary_contact)
def create_address(driver):
if not frappe.db.exists("Address", {"address_title": "_Test Address for Driver"}):