diff --git a/erpnext/bulk_transaction/doctype/bulk_transaction_log/bulk_transaction_log.py b/erpnext/bulk_transaction/doctype/bulk_transaction_log/bulk_transaction_log.py
index fbe9d7fcf7d..6ffb23659c8 100644
--- a/erpnext/bulk_transaction/doctype/bulk_transaction_log/bulk_transaction_log.py
+++ b/erpnext/bulk_transaction/doctype/bulk_transaction_log/bulk_transaction_log.py
@@ -30,10 +30,7 @@ class BulkTransactionLog(Document):
def load_from_db(self):
log_detail = qb.DocType("Bulk Transaction Log Detail")
- has_records = frappe.db.sql(
- "select exists (select * from `tabBulk Transaction Log Detail` where date = %s);",
- (self.name,),
- )[0][0]
+ has_records = frappe.db.exists("Bulk Transaction Log Detail", {"date": self.name})
if not has_records:
raise frappe.DoesNotExistError
diff --git a/erpnext/bulk_transaction/doctype/bulk_transaction_log/test_bulk_transaction_log.py b/erpnext/bulk_transaction/doctype/bulk_transaction_log/test_bulk_transaction_log.py
index e3909ea619a..641ddc9bac6 100644
--- a/erpnext/bulk_transaction/doctype/bulk_transaction_log/test_bulk_transaction_log.py
+++ b/erpnext/bulk_transaction/doctype/bulk_transaction_log/test_bulk_transaction_log.py
@@ -1,11 +1,76 @@
-# Copyright (c) 2023, Frappe Technologies Pvt. Ltd. and Contributors
-# See license.txt
-
-# import frappe
+# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+import frappe
+from frappe.utils import nowtime, random_string
from erpnext.tests.utils import ERPNextTestSuite
class TestBulkTransactionLog(ERPNextTestSuite):
- pass
+ def _make_log_doc(self, date):
+ # "Bulk Transaction Log" is a virtual doctype named by date; build the doc
+ # in-memory and drive load_from_db() directly to exercise the converted query.
+ doc = frappe.new_doc("Bulk Transaction Log")
+ doc.name = date
+ return doc
+
+ def _insert_detail(self, date, status="Success"):
+ detail = frappe.get_doc(
+ {
+ "doctype": "Bulk Transaction Log Detail",
+ "from_doctype": "Sales Order",
+ "to_doctype": "Sales Invoice",
+ "transaction_name": "_Test BTLD " + random_string(8),
+ "date": date,
+ "time": nowtime(),
+ "transaction_status": status,
+ }
+ )
+ # transaction_name is a Dynamic Link (options=from_doctype); the converted
+ # query never reads it, so skip link validation rather than create real txns.
+ detail.insert(ignore_permissions=True, ignore_links=True)
+ return detail
+
+ def test_load_raises_when_no_detail_rows(self):
+ # A date with zero Bulk Transaction Log Detail rows must not resolve to a log.
+ date = "2024-01-01"
+ self.assertFalse(
+ frappe.db.exists("Bulk Transaction Log Detail", {"date": date}),
+ "precondition: no detail rows for this date",
+ )
+
+ doc = self._make_log_doc(date)
+ self.assertRaises(frappe.DoesNotExistError, doc.load_from_db)
+
+ def test_load_succeeds_and_aggregates_after_detail_inserted(self):
+ date = "2024-02-02"
+
+ # Initially absent -> load_from_db must raise.
+ self.assertRaises(frappe.DoesNotExistError, self._make_log_doc(date).load_from_db)
+
+ # Insert detail rows for this date: 2 succeeded, 1 failed.
+ self._insert_detail(date, "Success")
+ self._insert_detail(date, "Success")
+ self._insert_detail(date, "Failed")
+
+ # Now the exists() check passes and load_from_db() populates aggregates.
+ doc = self._make_log_doc(date)
+ doc.load_from_db()
+
+ self.assertEqual(doc.date, date)
+ self.assertEqual(doc.succeeded, 2)
+ self.assertEqual(doc.failed, 1)
+ self.assertEqual(doc.log_entries, 3)
+
+ def test_load_isolated_per_date(self):
+ # Detail rows on a different date must not satisfy the lookup for our date.
+ other_date = "2024-03-03"
+ self._insert_detail(other_date, "Success")
+
+ target_date = "2024-04-04"
+ self.assertFalse(
+ frappe.db.exists("Bulk Transaction Log Detail", {"date": target_date}),
+ "target date has no rows; rows on another date must not leak in",
+ )
+ self.assertRaises(frappe.DoesNotExistError, self._make_log_doc(target_date).load_from_db)
diff --git a/erpnext/quality_management/doctype/quality_procedure/quality_procedure.py b/erpnext/quality_management/doctype/quality_procedure/quality_procedure.py
index b98442fe92a..41e4412f799 100644
--- a/erpnext/quality_management/doctype/quality_procedure/quality_procedure.py
+++ b/erpnext/quality_management/doctype/quality_procedure/quality_procedure.py
@@ -49,11 +49,8 @@ class QualityProcedure(NestedSet):
def on_trash(self):
# clear from child table (sub procedures)
- frappe.db.sql(
- """update `tabQuality Procedure Process`
- set `procedure`='' where `procedure`=%s""",
- self.name,
- )
+ qpp = frappe.qb.DocType("Quality Procedure Process")
+ frappe.qb.update(qpp).set(qpp["procedure"], "").where(qpp["procedure"] == self.name).run()
NestedSet.on_trash(self, allow_root_deletion=True)
def check_for_incorrect_child(self):
diff --git a/erpnext/quality_management/doctype/quality_procedure/test_quality_procedure.py b/erpnext/quality_management/doctype/quality_procedure/test_quality_procedure.py
index e4d847159f6..61e0da093cd 100644
--- a/erpnext/quality_management/doctype/quality_procedure/test_quality_procedure.py
+++ b/erpnext/quality_management/doctype/quality_procedure/test_quality_procedure.py
@@ -65,6 +65,40 @@ class TestQualityProcedure(ERPNextTestSuite):
child_qp.reload()
self.assertEqual(child_qp.parent_quality_procedure, None)
+ def test_on_trash_clears_referencing_process(self):
+ # Build a parent group with a sub-procedure. The parent's child table gets a
+ # `Quality Procedure Process` row whose `procedure` field points at the child.
+ child_qp = create_procedure(
+ {
+ "quality_procedure_name": "Test Child On Trash",
+ "is_group": 0,
+ }
+ )
+ create_procedure(
+ {
+ "quality_procedure_name": "Test Group On Trash",
+ "is_group": 1,
+ "processes": [dict(procedure=child_qp.name)],
+ }
+ )
+
+ # Sanity: a process row in the parent references the child by name.
+ referencing_rows = frappe.get_all(
+ "Quality Procedure Process",
+ filters={"procedure": child_qp.name},
+ pluck="name",
+ )
+ self.assertTrue(referencing_rows)
+
+ # Deleting the child runs on_trash() -> the converted UPDATE clears `procedure`.
+ child_qp.delete()
+
+ for row_name in referencing_rows:
+ self.assertEqual(
+ frappe.db.get_value("Quality Procedure Process", row_name, "procedure"),
+ "",
+ )
+
def remove_child_from_old_parent(self):
child_qp = create_procedure(
{
diff --git a/erpnext/telephony/doctype/call_log/call_log.py b/erpnext/telephony/doctype/call_log/call_log.py
index 781fb844777..9c00e5aeb6d 100644
--- a/erpnext/telephony/doctype/call_log/call_log.py
+++ b/erpnext/telephony/doctype/call_log/call_log.py
@@ -7,6 +7,8 @@ from frappe import _
from frappe.contacts.doctype.contact.contact import get_contact_with_phone_number
from frappe.core.doctype.dynamic_link.dynamic_link import deduplicate_dynamic_links
from frappe.model.document import Document
+from frappe.query_builder import Case
+from frappe.query_builder.functions import Sum
from erpnext.crm.doctype.lead.lead import get_lead_with_phone_number
from erpnext.crm.doctype.utils import get_scheduled_employees_for_popup, strip_number
@@ -168,22 +170,22 @@ def link_existing_conversations(doc, state):
number = strip_number(number)
if not number:
continue
- logs = frappe.db.sql_list(
- """
- SELECT cl.name FROM `tabCall Log` cl
- LEFT JOIN `tabDynamic Link` dl
- ON cl.name = dl.parent
- WHERE (cl.`from` like %(phone_number)s or cl.`to` like %(phone_number)s)
- GROUP BY cl.name
- HAVING SUM(
- CASE
- WHEN dl.link_doctype = %(doctype)s AND dl.link_name = %(docname)s
- THEN 1
- ELSE 0
- END
- )=0
- """,
- dict(phone_number=f"%{number}", docname=doc.name, doctype=doc.doctype),
+ cl = frappe.qb.DocType("Call Log")
+ dl = frappe.qb.DocType("Dynamic Link")
+ logs = (
+ frappe.qb.from_(cl)
+ .left_join(dl)
+ .on(cl.name == dl.parent)
+ .select(cl.name)
+ .where(cl["from"].like(f"%{number}") | cl["to"].like(f"%{number}"))
+ .groupby(cl.name)
+ .having(
+ Sum(
+ Case().when((dl.link_doctype == doc.doctype) & (dl.link_name == doc.name), 1).else_(0)
+ )
+ == 0
+ )
+ .run(pluck=True)
)
if logs:
for log in logs:
diff --git a/erpnext/telephony/doctype/call_log/test_call_log.py b/erpnext/telephony/doctype/call_log/test_call_log.py
index 1db1390c5ee..3dbb014eb9e 100644
--- a/erpnext/telephony/doctype/call_log/test_call_log.py
+++ b/erpnext/telephony/doctype/call_log/test_call_log.py
@@ -1,9 +1,112 @@
-# Copyright (c) 2020, Frappe Technologies Pvt. Ltd. and Contributors
-# See license.txt
-# import frappe
+# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+import random
+import string
+
+import frappe
+
+from erpnext.telephony.doctype.call_log.call_log import link_existing_conversations
from erpnext.tests.utils import ERPNextTestSuite
class TestCallLog(ERPNextTestSuite):
- pass
+ def setUp(self):
+ # A fresh, unused 8-digit suffix guarantees the controller's before_insert
+ # auto-linking (Contact/Lead lookup) finds nothing, so the only Dynamic Link
+ # rows present are the ones this test creates.
+ self.number = "98" + "".join(random.choices(string.digits, k=8))
+
+ # Two Call Logs that share the same phone number: one via `to`, one via
+ # `from`. Both must be matched by the `from LIKE | to LIKE` predicate, and
+ # the leading "+91" / "0" prefixes exercise strip_number normalisation
+ # (the trailing digits still end with self.number).
+ self.linked_log = self._make_call_log(to=f"+91{self.number}", type="Incoming")
+ self.unlinked_log = self._make_call_log(**{"from": f"0{self.number}", "type": "Outgoing"})
+
+ # The target doc the existing conversations get linked to. A real Contact
+ # (with a name + matching phone) is required so link_existing_conversations
+ # accepts it (doctype == "Contact") and reads doc.phone_nos / doc.name.
+ self.contact = frappe.get_doc(
+ {
+ "doctype": "Contact",
+ "first_name": f"_Test Caller {self.number}",
+ "phone_nos": [{"phone": self.number, "is_primary_phone": 1}],
+ }
+ )
+ # Suppress the Contact's own after_insert auto-link hook during insert, so
+ # the test controls exactly which log is pre-linked (the hook is invoked
+ # explicitly via _run_linker once the fixtures are in place).
+ self.contact.flags.ignore_auto_link_call_log = True
+ self.contact.insert(ignore_permissions=True)
+
+ # Pre-link ONLY one of the two logs to the contact via a Dynamic Link row.
+ # The converted HAVING SUM(CASE ...) == 0 filter must therefore exclude
+ # this log and return the other when link_existing_conversations runs.
+ self._add_link(self.linked_log, "Contact", self.contact.name)
+
+ def _make_call_log(self, **kwargs):
+ doc = frappe.get_doc({"doctype": "Call Log", "id": frappe.generate_hash(length=10), **kwargs})
+ doc.insert(ignore_permissions=True)
+ return doc.name
+
+ def _add_link(self, call_log, link_doctype, link_name):
+ doc = frappe.get_doc("Call Log", call_log)
+ doc.append("links", {"link_doctype": link_doctype, "link_name": link_name})
+ doc.save(ignore_permissions=True)
+
+ def _run_linker(self):
+ # Clear the flag set during insert so the explicit call actually runs the
+ # converted LEFT JOIN / GROUP BY / HAVING query path.
+ self.contact.flags.ignore_auto_link_call_log = False
+ link_existing_conversations(self.contact, "Open")
+
+ def _contact_links_of(self, call_log):
+ return frappe.get_all(
+ "Dynamic Link",
+ filters={"parenttype": "Call Log", "parent": call_log, "link_doctype": "Contact"},
+ fields=["link_name"],
+ pluck="link_name",
+ )
+
+ def test_links_previously_unlinked_log(self):
+ """The converted query's HAVING == 0 returns the log NOT yet linked to the
+ contact, so link_existing_conversations adds the Contact link to it."""
+ self.assertEqual(self._contact_links_of(self.unlinked_log), [], "precondition")
+
+ self._run_linker()
+
+ self.assertEqual(
+ self._contact_links_of(self.unlinked_log),
+ [self.contact.name],
+ "Previously-unlinked log matching the number must gain the Contact link",
+ )
+
+ def test_already_linked_log_is_not_relinked(self):
+ """The HAVING SUM(CASE ...) == 0 must EXCLUDE the already-linked log from the returned set,
+ so link_existing_conversations never re-saves it. Asserting only the link count is not enough
+ (validate() -> deduplicate_dynamic_links strips a duplicate either way), so pin the HAVING by
+ asserting the already-linked log was never touched: an excluded log is not in `logs`, so
+ add_link()/save() never runs and its `modified` timestamp is unchanged."""
+ self.assertEqual(self._contact_links_of(self.linked_log), [self.contact.name], "precondition")
+ modified_before = frappe.db.get_value("Call Log", self.linked_log, "modified")
+
+ self._run_linker()
+
+ # Excluded by HAVING -> never re-saved -> modified unchanged. (If HAVING were dropped/inverted
+ # the log would be returned, re-saved, and modified would bump.)
+ self.assertEqual(
+ frappe.db.get_value("Call Log", self.linked_log, "modified"),
+ modified_before,
+ "Already-linked log must be excluded by HAVING and never re-saved",
+ )
+ self.assertEqual(self._contact_links_of(self.linked_log), [self.contact.name])
+
+ def test_log_not_matching_number_is_untouched(self):
+ """A log whose from/to does not contain the number is excluded by the
+ from/to LIKE predicate and must stay unlinked."""
+ other = self._make_call_log(**{"from": "+919999999999", "to": "+918888888888", "type": "Outgoing"})
+
+ self._run_linker()
+
+ self.assertEqual(self._contact_links_of(other), [], "Log not matching the number must stay unlinked")
diff --git a/erpnext/utilities/doctype/rename_tool/rename_tool.py b/erpnext/utilities/doctype/rename_tool/rename_tool.py
index fc09cda8ed8..972c87deaac 100644
--- a/erpnext/utilities/doctype/rename_tool/rename_tool.py
+++ b/erpnext/utilities/doctype/rename_tool/rename_tool.py
@@ -29,9 +29,8 @@ class RenameTool(Document):
@frappe.whitelist()
@deprecated
def get_doctypes():
- return frappe.db.sql_list(
- """select name from tabDocType
- where allow_rename=1 and module!='Core' order by name"""
+ return frappe.get_all(
+ "DocType", filters={"allow_rename": 1, "module": ["!=", "Core"]}, order_by="name", pluck="name"
)
diff --git a/erpnext/utilities/test_transaction_base.py b/erpnext/utilities/test_transaction_base.py
new file mode 100644
index 00000000000..6d297d02c38
--- /dev/null
+++ b/erpnext/utilities/test_transaction_base.py
@@ -0,0 +1,77 @@
+# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+
+import frappe
+from frappe.utils import now_datetime, random_string
+
+from erpnext.tests.utils import ERPNextTestSuite
+from erpnext.utilities.transaction_base import delete_events
+
+
+class TestDeleteEvents(ERPNextTestSuite):
+ def _make_event(self, reference_doctype, reference_docname):
+ # Insert a bare Event, then attach the Event Participants child row directly.
+ # reference_docname is a Dynamic Link that would otherwise be validated against a
+ # real target doc on save; db_insert keeps the test self-contained with arbitrary
+ # (random, guaranteed-unique) docnames while still populating exactly the columns
+ # delete_events joins/filters on (parent, reference_doctype, reference_docname).
+ event = frappe.get_doc(
+ {
+ "doctype": "Event",
+ "subject": "Test Event " + random_string(10),
+ "starts_on": now_datetime(),
+ "event_type": "Private",
+ }
+ ).insert(ignore_permissions=True)
+
+ participant = frappe.new_doc("Event Participants")
+ participant.name = frappe.generate_hash(length=10)
+ participant.flags.name_set = True
+ participant.parent = event.name
+ participant.parenttype = "Event"
+ participant.parentfield = "event_participants"
+ participant.idx = 1
+ participant.reference_doctype = reference_doctype
+ participant.reference_docname = reference_docname
+ participant.db_insert()
+
+ return event.name
+
+ def test_delete_events_removes_matching_and_keeps_others(self):
+ # Two distinct, real reference_docnames so the filter has something to discriminate on.
+ match_name = "Match " + random_string(10)
+ other_name = "Other " + random_string(10)
+ event_match = self._make_event("Customer", match_name)
+ event_other = self._make_event("Customer", other_name)
+
+ # Sanity: both exist before deletion (otherwise the assertions below are tautological).
+ self.assertTrue(frappe.db.exists("Event", event_match))
+ self.assertTrue(frappe.db.exists("Event", event_other))
+
+ delete_events("Customer", match_name)
+
+ # Only the Event whose participant matches BOTH reference_doctype and
+ # reference_docname must be deleted.
+ self.assertFalse(frappe.db.exists("Event", event_match))
+ self.assertTrue(frappe.db.exists("Event", event_other))
+
+ def test_delete_events_no_match_is_noop(self):
+ # When nothing matches, no Event may be deleted.
+ event = self._make_event("Customer", "Present " + random_string(10))
+ self.assertTrue(frappe.db.exists("Event", event))
+
+ delete_events("Customer", "Absent " + random_string(10))
+
+ self.assertTrue(frappe.db.exists("Event", event))
+
+ def test_delete_events_distinguishes_reference_doctype(self):
+ # Same docname under two different reference_doctypes: only the queried doctype
+ # is deleted, proving both predicates are ANDed together.
+ shared_name = "Shared " + random_string(10)
+ event_customer = self._make_event("Customer", shared_name)
+ event_supplier = self._make_event("Supplier", shared_name)
+
+ delete_events("Customer", shared_name)
+
+ self.assertFalse(frappe.db.exists("Event", event_customer))
+ self.assertTrue(frappe.db.exists("Event", event_supplier))
diff --git a/erpnext/utilities/transaction_base.py b/erpnext/utilities/transaction_base.py
index 4b51bc5ecd3..bd7bbcdd34b 100644
--- a/erpnext/utilities/transaction_base.py
+++ b/erpnext/utilities/transaction_base.py
@@ -582,19 +582,16 @@ class TransactionBase(StatusUpdater):
def delete_events(ref_type, ref_name):
+ event = frappe.qb.DocType("Event")
+ participant = frappe.qb.DocType("Event Participants")
events = (
- frappe.db.sql_list(
- """ SELECT
- distinct `tabEvent`.name
- from
- `tabEvent`, `tabEvent Participants`
- where
- `tabEvent`.name = `tabEvent Participants`.parent
- and `tabEvent Participants`.reference_doctype = %s
- and `tabEvent Participants`.reference_docname = %s
- """,
- (ref_type, ref_name),
- )
+ frappe.qb.from_(event)
+ .inner_join(participant)
+ .on(event.name == participant.parent)
+ .select(event.name)
+ .distinct()
+ .where((participant.reference_doctype == ref_type) & (participant.reference_docname == ref_name))
+ .run(pluck="name")
or []
)
diff --git a/erpnext/www/payment_setup_certification.py b/erpnext/www/payment_setup_certification.py
index 5d62d60f5eb..e8e9d51668c 100644
--- a/erpnext/www/payment_setup_certification.py
+++ b/erpnext/www/payment_setup_certification.py
@@ -1,4 +1,5 @@
import frappe
+from frappe.query_builder.functions import IfNull
no_cache = 1
@@ -12,13 +13,15 @@ def get_context(context):
def get_all_certifications_of_a_member():
"""Returns all certifications"""
all_certifications = []
- all_certifications = frappe.db.sql(
- """ select cc.name,cc.from_date,cc.to_date,ca.amount,ca.currency
- from `tabCertified Consultant` cc
- inner join `tabCertification Application` ca
- on cc.certification_application = ca.name
- where paid = 1 and email = %(user)s order by cc.to_date desc""",
- {"user": frappe.session.user},
- as_dict=True,
+ cc = frappe.qb.DocType("Certified Consultant")
+ ca = frappe.qb.DocType("Certification Application")
+ all_certifications = (
+ frappe.qb.from_(cc)
+ .inner_join(ca)
+ .on(cc.certification_application == ca.name)
+ .select(cc.name, cc.from_date, cc.to_date, ca.amount, ca.currency)
+ .where((cc.paid == 1) & (cc.email == frappe.session.user))
+ .orderby(IfNull(cc.to_date, "0001-01-01"), order=frappe.qb.desc)
+ .run(as_dict=True)
)
return all_certifications
diff --git a/erpnext/www/support/index.py b/erpnext/www/support/index.py
index 83fb8959d61..ddbb3af7c7f 100644
--- a/erpnext/www/support/index.py
+++ b/erpnext/www/support/index.py
@@ -1,4 +1,5 @@
import frappe
+from frappe.query_builder.functions import Count, Max
def get_context(context):
@@ -30,25 +31,27 @@ def get_context(context):
def get_favorite_articles_by_page_view():
- return frappe.db.sql(
- """
- SELECT
- t1.name as name,
- t1.title as title,
- t1.content as content,
- t1.route as route,
- t1.category as category,
- count(t1.route) as count
- FROM `tabHelp Article` AS t1
- INNER JOIN
- `tabWeb Page View` AS t2
- ON t1.route = t2.path
- WHERE t1.published = 1
- GROUP BY route
- ORDER BY count DESC
- LIMIT 6;
- """,
- as_dict=True,
+ ha = frappe.qb.DocType("Help Article")
+ wpv = frappe.qb.DocType("Web Page View")
+ return (
+ frappe.qb.from_(ha)
+ .inner_join(wpv)
+ .on(ha.route == wpv.path)
+ .select(
+ # route is the unique page URL, so there is one published article per route: Max() just
+ # returns that row's columns while keeping the GROUP BY route valid on postgres
+ Max(ha.name).as_("name"),
+ Max(ha.title).as_("title"),
+ Max(ha.content).as_("content"),
+ ha.route,
+ Max(ha.category).as_("category"),
+ Count(ha.route).as_("count"),
+ )
+ .where(ha.published == 1)
+ .groupby(ha.route)
+ .orderby(Count(ha.route), order=frappe.qb.desc)
+ .limit(6)
+ .run(as_dict=True)
)
diff --git a/erpnext/www/support/test_support_index.py b/erpnext/www/support/test_support_index.py
new file mode 100644
index 00000000000..52ed2133bc0
--- /dev/null
+++ b/erpnext/www/support/test_support_index.py
@@ -0,0 +1,89 @@
+# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+
+import frappe
+from frappe.utils import random_string
+
+from erpnext.tests.utils import ERPNextTestSuite
+from erpnext.www.support.index import get_favorite_articles_by_page_view
+
+
+class TestSupportIndex(ERPNextTestSuite):
+ def make_help_category(self):
+ category_name = "_Test Support Category " + random_string(8)
+ category = frappe.get_doc(
+ {
+ "doctype": "Help Category",
+ "category_name": category_name,
+ "published": 1,
+ }
+ ).insert(ignore_permissions=True)
+ return category.name
+
+ def make_help_article(self, category, route, title, content, published=1):
+ article = frappe.get_doc(
+ {
+ "doctype": "Help Article",
+ "title": title,
+ "category": category,
+ "content": content,
+ "route": route,
+ "published": published,
+ }
+ ).insert(ignore_permissions=True)
+ return article.name
+
+ def seed_page_views(self, path, count):
+ # Web Page View is in_create/read_only; insert the minimal row the
+ # converted JOIN reads (path == Help Article.route) directly.
+ for _ in range(count):
+ view = frappe.new_doc("Web Page View")
+ view.path = path
+ view.is_unique = "1"
+ view.flags.name_set = True
+ view.name = frappe.generate_hash("wpv", 12)
+ view.db_insert()
+
+ def test_favorite_articles_ordered_by_page_view_count(self):
+ category = self.make_help_category()
+
+ # Distinct, collision-free routes so other published articles in the DB
+ # can't masquerade as ours.
+ route_hi = "support-hi-" + random_string(10)
+ route_lo = "support-lo-" + random_string(10)
+
+ name_hi = self.make_help_article(
+ category, route_hi, "High Views Article", "High views content
"
+ )
+ name_lo = self.make_help_article(category, route_lo, "Low Views Article", "Low views content
")
+
+ # More views on route_hi than route_lo: a broken Count/GROUP BY/ORDER BY
+ # would not reproduce these exact counts or this ordering.
+ self.seed_page_views(route_hi, 3)
+ self.seed_page_views(route_lo, 1)
+
+ results = get_favorite_articles_by_page_view()
+
+ by_route = {row.route: row for row in results if row.route in (route_hi, route_lo)}
+
+ # Both of our routes are surfaced by the INNER JOIN on route == path.
+ self.assertIn(route_hi, by_route, "High-viewed route missing from results")
+ self.assertIn(route_lo, by_route, "Low-viewed route missing from results")
+
+ # Count(route) reflects the real number of seeded Web Page View rows.
+ self.assertEqual(by_route[route_hi]["count"], 3)
+ self.assertEqual(by_route[route_lo]["count"], 1)
+
+ # Max()-wrapped columns carry the article's own data (one row per route).
+ self.assertEqual(by_route[route_hi].name, name_hi)
+ self.assertEqual(by_route[route_hi].title, "High Views Article")
+ self.assertEqual(by_route[route_hi].category, category)
+ self.assertEqual(by_route[route_lo].name, name_lo)
+
+ # ORDER BY count desc: the higher-viewed route precedes the lower one.
+ ordered_routes = [row.route for row in results if row.route in (route_hi, route_lo)]
+ self.assertEqual(
+ ordered_routes,
+ [route_hi, route_lo],
+ "Results not ordered by page-view count descending",
+ )