refactor(postgres): port maintenance_visit queries to the query builder

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

erpnext/maintenance/doctype/maintenance_visit/maintenance_visit.py

@@ -143,9 +143,22 @@ class MaintenanceVisit(TransactionBase):
 						elif self.completion_status == "Partially Completed":
 							status = "Work In Progress"
 					else:
-						nm = frappe.db.sql(
+						mv = frappe.qb.DocType("Maintenance Visit")
-							"select t1.name, t1.mntc_date, t2.service_person, t2.work_done from `tabMaintenance Visit` t1, `tabMaintenance Visit Purpose` t2 where t2.parent = t1.name and t1.completion_status = 'Partially Completed' and t2.prevdoc_docname = %s and t1.name!=%s and t1.docstatus = 1 order by t1.name desc limit 1",
+						mvp = frappe.qb.DocType("Maintenance Visit Purpose")
-							(d.prevdoc_docname, self.name),
+						nm = (
+							frappe.qb.from_(mv)
+							.inner_join(mvp)
+							.on(mvp.parent == mv.name)
+							.select(mv.name, mv.mntc_date, mvp.service_person, mvp.work_done)
+							.where(
+								(mv.completion_status == "Partially Completed")
+								& (mvp.prevdoc_docname == d.prevdoc_docname)
+								& (mv.name != self.name)
+								& (mv.docstatus == 1)
+							)
+							.orderby(mv.name, order=frappe.qb.desc)
+							.limit(1)
+							.run()
 						)
 
 						if nm:
@@ -180,14 +193,27 @@ class MaintenanceVisit(TransactionBase):
 				# check_for_doctype = d.prevdoc_doctype
 
 		if check_for_docname:
-			check = frappe.db.sql(
+			mv = frappe.qb.DocType("Maintenance Visit")
-				"select t1.name from `tabMaintenance Visit` t1, `tabMaintenance Visit Purpose` t2 where t2.parent = t1.name and t1.name!=%s and t2.prevdoc_docname=%s and t1.docstatus = 1 and (t1.mntc_date > %s or (t1.mntc_date = %s and t1.mntc_time > %s))",
+			mvp = frappe.qb.DocType("Maintenance Visit Purpose")
-				(self.name, check_for_docname, self.mntc_date, self.mntc_date, self.mntc_time),
+			check = (
+				frappe.qb.from_(mv)
+				.inner_join(mvp)
+				.on(mvp.parent == mv.name)
+				.select(mv.name)
+				.where(
+					(mv.name != self.name)
+					& (mvp.prevdoc_docname == check_for_docname)
+					& (mv.docstatus == 1)
+					& (
+						(mv.mntc_date > self.mntc_date)
+						| ((mv.mntc_date == self.mntc_date) & (mv.mntc_time > self.mntc_time))
+					)
+				)
+				.run(pluck=True)
 			)
 
 			if check:
-				check_lst = [x[0] for x in check]
+				check_lst = ",".join(check)
-				check_lst = ",".join(check_lst)
 				frappe.throw(
 					_("Cancel Material Visits {0} before cancelling this Maintenance Visit").format(check_lst)
 				)

erpnext/maintenance/doctype/maintenance_visit/test_maintenance_visit.py

@@ -2,13 +2,139 @@
 # See license.txt
 
 import frappe
-from frappe.utils.data import today
+from frappe.utils.data import add_days, getdate, today
 
 from erpnext.tests.utils import ERPNextTestSuite
 
 
 class TestMaintenanceVisit(ERPNextTestSuite):
-	pass
+	def setUp(self):
+		self.sales_person = make_sales_person("_Test Maintenance Service Person")
+
+	def make_warranty_claim(self):
+		# Warranty Claim is not submittable; it provides a real target for the
+		# purposes-row Dynamic Link (prevdoc_doctype/prevdoc_docname).
+		claim = frappe.new_doc("Warranty Claim")
+		claim.status = "Open"
+		claim.complaint_date = today()
+		claim.customer = "_Test Customer"
+		claim.item_code = "_Test Item"
+		claim.complaint = "Device stopped working under warranty"
+		claim.company = "_Test Company"
+		claim.insert(ignore_permissions=True)
+		return claim
+
+	def make_visit(self, claim, completion_status, mntc_date=None, mntc_time=None, submit=True):
+		visit = frappe.new_doc("Maintenance Visit")
+		visit.company = "_Test Company"
+		visit.customer = "_Test Customer"
+		visit.mntc_date = mntc_date or today()
+		if mntc_time:
+			visit.mntc_time = mntc_time
+		visit.maintenance_type = "Unscheduled"
+		visit.completion_status = completion_status
+		visit.append(
+			"purposes",
+			{
+				"item_code": "_Test Item",
+				"service_person": self.sales_person.name,
+				"work_done": "Replaced the faulty component",
+				"description": "Warranty repair",
+				"prevdoc_doctype": "Warranty Claim",
+				"prevdoc_docname": claim.name,
+			},
+		)
+		visit.insert(ignore_permissions=True)
+		if submit:
+			visit.submit()
+		return visit
+
+	def test_cancel_blocked_when_later_visit_exists(self):
+		# check_if_last_visit's converted join query (B): cancelling an EARLIER
+		# submitted visit must be blocked while a LATER one (greater mntc_date)
+		# referencing the same prevdoc_docname is still active.
+		claim = self.make_warranty_claim()
+		earlier = self.make_visit(claim, "Partially Completed", mntc_date=today())
+		later = self.make_visit(claim, "Partially Completed", mntc_date=add_days(today(), 5))
+
+		# Sanity: both are submitted and share the prevdoc_docname the query keys on.
+		self.assertEqual(earlier.docstatus, 1)
+		self.assertEqual(later.docstatus, 1)
+		self.assertEqual(later.purposes[0].prevdoc_docname, claim.name)
+
+		# The throw originates in check_if_last_visit's query (B): a later visit exists.
+		self.assertRaisesRegex(frappe.ValidationError, later.name, earlier.cancel)
+
+	def test_cancel_blocked_by_same_date_later_time(self):
+		# Same converted query (B), time-tiebreak branch: equal mntc_date, but the
+		# blocking visit has a strictly greater mntc_time.
+		claim = self.make_warranty_claim()
+		earlier = self.make_visit(claim, "Partially Completed", mntc_date=today(), mntc_time="09:00:00")
+		later = self.make_visit(claim, "Partially Completed", mntc_date=today(), mntc_time="15:00:00")
+
+		self.assertRaisesRegex(frappe.ValidationError, later.name, earlier.cancel)
+
+	def test_cancel_allowed_for_latest_visit(self):
+		# The latest visit has no later sibling -> query (B) returns nothing ->
+		# cancellation proceeds and the visit is marked Cancelled.
+		claim = self.make_warranty_claim()
+		earlier = self.make_visit(claim, "Partially Completed", mntc_date=today())
+		later = self.make_visit(claim, "Partially Completed", mntc_date=add_days(today(), 5))
+
+		later.cancel()
+
+		self.assertEqual(frappe.db.get_value("Maintenance Visit", later.name, "docstatus"), 2)
+		self.assertEqual(frappe.db.get_value("Maintenance Visit", later.name, "status"), "Cancelled")
+		# The earlier one is untouched and still submitted.
+		self.assertEqual(frappe.db.get_value("Maintenance Visit", earlier.name, "docstatus"), 1)
+
+	def test_cancel_reopens_claim_to_work_in_progress_from_prior_partial(self):
+		# Drives the status-update query (A) inside update_customer_issue(flag=0).
+		# A submitted "Partially Completed" visit (prior) exists for the claim; when
+		# a LATER "Fully Completed" visit is cancelled, query (A) finds that prior
+		# partial visit and the Warranty Claim is reopened to "Work In Progress"
+		# carrying the prior visit's resolution data.
+		claim = self.make_warranty_claim()
+		prior = self.make_visit(claim, "Partially Completed", mntc_date=today())
+		latest = self.make_visit(claim, "Fully Completed", mntc_date=add_days(today(), 3))
+
+		# After submitting the "Fully Completed" visit the claim is Closed.
+		self.assertEqual(frappe.db.get_value("Warranty Claim", claim.name, "status"), "Closed")
+
+		# Cancelling the latest visit: no later sibling blocks it, so cancel runs
+		# update_customer_issue(0), which executes query (A) and reopens the claim.
+		latest.cancel()
+
+		claim.reload()
+		self.assertEqual(claim.status, "Work In Progress")
+		# Resolution data is back-filled from the prior partial visit found by query (A).
+		self.assertEqual(claim.resolved_by, self.sales_person.name)
+		self.assertEqual(claim.resolution_details, prior.purposes[0].work_done)
+		self.assertEqual(getdate(claim.resolution_date), getdate(prior.mntc_date))
+
+	def test_cancel_reopens_claim_to_open_when_no_prior_partial(self):
+		# Inverse of query (A): a lone "Fully Completed" visit with no prior
+		# "Partially Completed" sibling -> query (A) returns nothing -> the claim
+		# is reset to "Open" with cleared resolution fields on cancel.
+		claim = self.make_warranty_claim()
+		visit = self.make_visit(claim, "Fully Completed", mntc_date=today())
+		self.assertEqual(frappe.db.get_value("Warranty Claim", claim.name, "status"), "Closed")
+
+		visit.cancel()
+
+		claim.reload()
+		self.assertEqual(claim.status, "Open")
+		self.assertIsNone(claim.resolved_by)
+		self.assertIsNone(claim.resolution_details)
+		self.assertIsNone(claim.resolution_date)
+
+
+def make_sales_person(name):
+	sales_person = frappe.get_doc({"doctype": "Sales Person", "sales_person_name": name})
+	sales_person.insert(ignore_if_duplicate=True)
+	if not sales_person.name:
+		sales_person = frappe.get_doc("Sales Person", {"sales_person_name": name})
+	return sales_person
 
 
 def make_maintenance_visit():
@@ -33,10 +159,3 @@ def make_maintenance_visit():
 	mv.insert(ignore_permissions=True)
 
 	return mv
-
-
-def make_sales_person(name):
-	sales_person = frappe.get_doc({"doctype": "Sales Person", "sales_person_name": name})
-	sales_person.insert(ignore_if_duplicate=True)
-
-	return sales_person