ext/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-10-29 22:57:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: resolve conflicts in backport of #60780

Browse Source
This commit is contained in:
Nabin Hait
2026-10-24 10:05:18 +05:30
parent e575c22987
commit ff4b64cee4
1 changed files with 11 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
erpnext/telephony/doctype/call_log/test_call_log.py
+11 -240
View File
@@ -1,185 +1,28 @@
# Copyright (c) 2020, Frappe Technologies Pvt. Ltd. and Contributors
# See license.txt
# import frappe
import unittest
import random
import string
from unittest.mock import patch
import frappe
from frappe.tests.utils import FrappeTestCase
from erpnext.setup.doctype.employee.test_employee import make_employee
<<<<<<< HEAD
class TestCallLog(unittest.TestCase):
pass
=======
class TestCallLog(ERPNextTestSuite):
class TestCallLog(FrappeTestCase):
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 tearDown(self):
frappe.db.rollback()
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_linking_does_not_commit(self):
with patch.object(frappe, "in_test", False), patch.object(frappe.db, "commit") as commit:
self._run_linker()
commit.assert_not_called()
self.assertEqual(self._contact_links_of(self.unlinked_log), [self.contact.name])
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")
def test_agent_can_save_summary_of_own_call(self):
user = "test_call_agent@example.com"
employee = make_employee(user, company="_Test Company")
if not frappe.db.exists("Telephony Call Type", "_Test Call Type"):
frappe.get_doc({"doctype": "Telephony Call Type", "call_type": "_Test Call Type"}).submit()
call_log = self._make_call_log(
**{"from": "+919999999999", "type": "Outgoing"},
call_received_by=employee,
employee_user_id=user,
)
frappe.set_user(user)
try:
add_call_summary_and_call_type(call_log, "Wants a quote", "_Test Call Type")
finally:
frappe.set_user("Administrator")
self.assertEqual(
frappe.db.get_value("Call Log", call_log, ["type_of_call", "summary"]),
("_Test Call Type", "Wants a quote"),
)
def test_timeline_shows_only_permitted_call_logs(self):
self.assertEqual(len(get_linked_call_logs("Contact", self.contact.name)), 1)
user = "test_call_log_sales_user@example.com"
if not frappe.db.exists("User", user):
frappe.get_doc(
{
"doctype": "User",
"email": user,
"first_name": "Sales",
"send_welcome_email": 0,
"roles": [{"role": "Sales User"}],
}
).insert(ignore_permissions=True)
frappe.set_user(user)
try:
self.assertEqual(get_linked_call_logs("Contact", self.contact.name), [])
finally:
frappe.set_user("Administrator")
def test_transferred_call_is_stored_with_the_new_agent(self):
first_agent = make_employee(
"test_call_agent_a@example.com", company="_Test Company", cell_number=f"+91{self.number}1"
)
second_agent = make_employee(
"test_call_agent_b@example.com", company="_Test Company", cell_number=f"+91{self.number}2"
)
call_log = frappe.get_doc(
"Call Log", self._make_call_log(to=f"+91{self.number}1", type="Incoming", status="Ringing")
)
self.assertEqual(call_log.call_received_by, first_agent)
call_log.to = f"+91{self.number}2"
call_log.save(ignore_permissions=True)
self.assertEqual(
frappe.db.get_value("Call Log", call_log.name, ["call_received_by", "employee_user_id"]),
(second_agent, "test_call_agent_b@example.com"),
)
def test_call_status_events_go_only_to_agents_of_the_call(self):
agent = "test_call_agent_a@example.com"
make_employee(agent, company="_Test Company", cell_number=f"+91{self.number}1")
@@ -200,75 +43,3 @@ class TestCallLog(ERPNextTestSuite):
ended = [c for c in publish.call_args_list if c.args[0] == f"call_{call_log.id}_ended"]
self.assertEqual([c.kwargs.get("user") for c in ended], [agent])
def test_summary_can_be_saved_without_call_type(self):
add_call_summary_and_call_type(self.linked_log, "No type chosen", None)
self.assertEqual(frappe.db.get_value("Call Log", self.linked_log, "summary"), "No type chosen")
def test_call_from_customer_contact_links_the_customer(self):
number = "97" + "".join(random.choices(string.digits, k=8))
contact = frappe.get_doc(
{
"doctype": "Contact",
"first_name": f"_Test Customer Caller {number}",
"phone_nos": [{"phone": f"+91{number}", "is_primary_phone": 1}],
"links": [{"link_doctype": "Customer", "link_name": "_Test Customer"}],
}
).insert(ignore_permissions=True)
call_log = frappe.get_doc(
"Call Log", self._make_call_log(**{"from": f"+91{number}"}, type="Incoming")
)
self.assertEqual(call_log.customer, "_Test Customer")
self.assertIn(("Customer", "_Test Customer"), [(d.link_doctype, d.link_name) for d in call_log.links])
self.assertIn(("Contact", contact.name), [(d.link_doctype, d.link_name) for d in call_log.links])
def test_past_calls_are_linked_to_new_lead_and_new_contact_number(self):
number = "96" + "".join(random.choices(string.digits, k=8))
call_log = self._make_call_log(**{"from": f"+91{number}"}, type="Incoming")
lead = frappe.get_doc(
{"doctype": "Lead", "first_name": f"_Test Caller {number}", "mobile_no": f"+91{number}"}
).insert(ignore_permissions=True)
self.contact.reload()
self.contact.flags.ignore_auto_link_call_log = False
self.contact.append("phone_nos", {"phone": f"+91{number}"})
self.contact.save(ignore_permissions=True)
links = frappe.get_all(
"Dynamic Link",
filters={"parenttype": "Call Log", "parent": call_log},
fields=["link_doctype", "link_name"],
as_list=True,
)
self.assertIn(("Lead", lead.name), links)
self.assertIn(("Contact", self.contact.name), links)
def test_new_contact_number_links_past_calls_to_customer_and_lead_of_contact(self):
number = "95" + "".join(random.choices(string.digits, k=8))
call_log = self._make_call_log(**{"from": f"+91{number}"}, type="Incoming")
lead = frappe.get_doc({"doctype": "Lead", "first_name": f"_Test Caller {number}"}).insert(
ignore_permissions=True
)
contact = frappe.get_doc(
{
"doctype": "Contact",
"first_name": f"_Test Customer Caller {number}",
"links": [
{"link_doctype": "Customer", "link_name": "_Test Customer"},
{"link_doctype": "Lead", "link_name": lead.name},
],
}
).insert(ignore_permissions=True)
contact.append("phone_nos", {"phone": f"+91{number}", "is_primary_phone": 1})
contact.save(ignore_permissions=True)
call_log = frappe.get_doc("Call Log", call_log)
links = [(d.link_doctype, d.link_name) for d in call_log.links]
self.assertEqual(call_log.customer, "_Test Customer")
self.assertIn(("Customer", "_Test Customer"), links)
self.assertIn(("Lead", lead.name), links)
>>>>>>> 4201e50 (fix(telephony): send call status events to the agents of the call)