db69de1c11
Some checks failed
Frontend / Unit Tests & Coverage (push) Has been skipped
Regenerate POT file (translatable strings) / Regenerate POT file (develop) (push) Failing after 1m13s
Regenerate POT file (translatable strings) / Regenerate POT file (main-hotfix) (push) Failing after 6s
Server / Tests (frappe ${{ matrix.frappe-branch }}) (push) Has been skipped
Server / Resolve frappe matrix (push) Failing after 1s
Frontend / Lint (push) Failing after 1s
Frontend / Production Build (push) Failing after 1s
Server / Coverage Wrap Up (push) Has been skipped
Frontend / Coverage Wrap Up (push) Has been skipped
663 lines
24 KiB
Python
663 lines
24 KiB
Python
import time
|
|
from unittest.mock import patch
|
|
|
|
import frappe
|
|
from frappe.tests import IntegrationTestCase, UnitTestCase
|
|
|
|
from crm.utils import (
|
|
_get_communication_status,
|
|
_should_update_modified,
|
|
are_same_phone_number,
|
|
create_lead_from_incoming_email,
|
|
on_communication_update,
|
|
parse_phone_number,
|
|
seconds_to_duration,
|
|
)
|
|
|
|
|
|
class TestUtils(UnitTestCase):
|
|
def test_seconds_to_duration(self):
|
|
# 3661 seconds = 1 hour, 1 minute, and 1 second
|
|
self.assertEqual(seconds_to_duration(3661), "1h 1m 1s")
|
|
|
|
# 3660 seconds = 1 hour and 1 minute
|
|
self.assertEqual(seconds_to_duration(3660), "1h 1m")
|
|
|
|
# 3601 seconds = 1 hour and 1 second
|
|
self.assertEqual(seconds_to_duration(3601), "1h 1s")
|
|
|
|
# 0 seconds = 0s
|
|
self.assertEqual(seconds_to_duration(0), "0s")
|
|
|
|
def test_seconds_to_duration_single_unit(self):
|
|
"""Each single-unit branch (hours / minutes / seconds only) should format correctly."""
|
|
self.assertEqual(seconds_to_duration(3600), "1h") # hours only
|
|
self.assertEqual(seconds_to_duration(60), "1m") # minutes only
|
|
self.assertEqual(seconds_to_duration(1), "1s") # seconds only
|
|
|
|
def test_seconds_to_duration_minutes_and_seconds(self):
|
|
"""Minutes + seconds without hours should format correctly."""
|
|
self.assertEqual(seconds_to_duration(61), "1m 1s")
|
|
self.assertEqual(seconds_to_duration(90), "1m 30s")
|
|
|
|
def test_seconds_to_duration_falsy_values(self):
|
|
"""None and 0 are both falsy and should return '0s'."""
|
|
self.assertEqual(seconds_to_duration(None), "0s")
|
|
self.assertEqual(seconds_to_duration(0), "0s")
|
|
|
|
def test_seconds_to_duration_large_values(self):
|
|
"""Values larger than one hour should still compute correctly."""
|
|
self.assertEqual(seconds_to_duration(7322), "2h 2m 2s") # 2*3600 + 2*60 + 2
|
|
|
|
def test_are_same_phone_number_normalized_input(self):
|
|
# Indian number with country code and different formats
|
|
self.assertTrue(are_same_phone_number("+91 9845552671", "9845552671"))
|
|
self.assertTrue(are_same_phone_number("+91-984-555-2671", "9845552671"))
|
|
self.assertTrue(are_same_phone_number("+91 (984) 555-2671", "9845552671"))
|
|
self.assertTrue(are_same_phone_number("+91 984-555-2671", "9845552671"))
|
|
|
|
# US number with country code and different formats
|
|
self.assertTrue(are_same_phone_number("+1 415 555 2671", "4155552671", default_region="US"))
|
|
self.assertTrue(are_same_phone_number("+1-415-555-2671", "4155552671", default_region="US"))
|
|
self.assertTrue(are_same_phone_number("+1 (415) 555-2671", "4155552671", default_region="US"))
|
|
self.assertTrue(are_same_phone_number("+1 415-555-2671", "4155552671", default_region="US"))
|
|
|
|
def test_are_same_phone_number_invalid_input(self):
|
|
# Invalid numbers should return False
|
|
self.assertFalse(
|
|
are_same_phone_number("+1 415 555 2671", "4155552671")
|
|
) # Missing default region as US
|
|
self.assertFalse(
|
|
are_same_phone_number("+91 984-555-2671", "9845552671", default_region="US")
|
|
) # Wrong default region
|
|
self.assertFalse(are_same_phone_number("12345", "67890"))
|
|
self.assertFalse(are_same_phone_number("abc", "14155552671"))
|
|
|
|
def test_are_same_phone_number_different_numbers(self):
|
|
"""Two distinct valid numbers should not be considered same."""
|
|
self.assertFalse(are_same_phone_number("+91 9845552671", "+91 9845552672"))
|
|
self.assertFalse(are_same_phone_number("+1 415 555 2671", "+1 415 555 2672", default_region="US"))
|
|
|
|
def test_are_same_phone_number_validate_false(self):
|
|
"""With validate=False, structurally parseable but invalid numbers can still match."""
|
|
# These parse but fail is_valid_number; with validate=False they should still compare equal
|
|
self.assertTrue(
|
|
are_same_phone_number("+1 415 555 2671", "+1 415 555 2671", default_region="US", validate=False)
|
|
)
|
|
|
|
def test_parse_phone_number_valid_indian_number(self):
|
|
"""A valid Indian number should return success=True with correct metadata."""
|
|
result = parse_phone_number("+919845552671")
|
|
|
|
self.assertTrue(result["success"])
|
|
self.assertTrue(result["is_valid"])
|
|
self.assertEqual(result["country_code"], 91)
|
|
self.assertEqual(result["country"], "IN")
|
|
self.assertEqual(result["national_number"], "9845552671")
|
|
# All four formats must be present
|
|
self.assertIn("international", result["formats"])
|
|
self.assertIn("national", result["formats"])
|
|
self.assertIn("E164", result["formats"])
|
|
self.assertIn("RFC3966", result["formats"])
|
|
# E164 should start with +91
|
|
self.assertTrue(result["formats"]["E164"].startswith("+91"))
|
|
|
|
def test_parse_phone_number_valid_us_number(self):
|
|
"""A valid US number should return success=True with country_code 1."""
|
|
result = parse_phone_number("+14155552671")
|
|
|
|
self.assertTrue(result["success"])
|
|
self.assertTrue(result["is_valid"])
|
|
self.assertEqual(result["country_code"], 1)
|
|
self.assertEqual(result["country"], "US")
|
|
|
|
def test_parse_phone_number_invalid_string(self):
|
|
"""An unparseable string should return success=False with an error key."""
|
|
result = parse_phone_number("not-a-number")
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertIn("error", result)
|
|
|
|
def test_parse_phone_number_default_country_fallback(self):
|
|
"""A local number without a country prefix should be parsed using the default_country."""
|
|
result = parse_phone_number("9845552671", default_country="IN")
|
|
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["country"], "IN")
|
|
|
|
|
|
class TestUpdateModifiedTimestamp(IntegrationTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
# Patch frappe.enqueue to run update_modified_background synchronously in tests
|
|
self._enqueue_patch = patch("frappe.enqueue", self._immediate_enqueue)
|
|
self._enqueue_patch.start()
|
|
|
|
@staticmethod
|
|
def _immediate_enqueue(method, **kwargs):
|
|
# Only patch for update_modified_background
|
|
from crm.utils import update_modified_background
|
|
|
|
if method == update_modified_background or (
|
|
isinstance(method, str) and method.endswith("update_modified_background")
|
|
):
|
|
return update_modified_background(kwargs["doctype"], kwargs["docname"])
|
|
# fallback: do nothing
|
|
return None
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
# Ensure FCRM Settings has the flag enabled; disable both status hooks to isolate
|
|
# timestamp behaviour from communication-status side-effects.
|
|
frappe.db.set_single_value("FCRM Settings", "update_timestamp_on_new_communication", 1)
|
|
frappe.db.set_single_value("FCRM Settings", "auto_reopen_on_new_communication", 0)
|
|
frappe.db.set_single_value("FCRM Settings", "auto_mark_replied_on_response", 0)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
frappe.db.rollback()
|
|
super().tearDownClass()
|
|
|
|
def tearDown(self):
|
|
frappe.db.rollback()
|
|
self._enqueue_patch.stop()
|
|
super().tearDown()
|
|
|
|
def _make_lead(self):
|
|
lead = frappe.get_doc(
|
|
{
|
|
"doctype": "CRM Lead",
|
|
"first_name": "Test",
|
|
"email": "test.ts@example.com",
|
|
}
|
|
)
|
|
lead.insert(ignore_permissions=True)
|
|
return lead
|
|
|
|
def test_timestamp_updated_on_new_communication(self):
|
|
"""Inserting a Communication linked to a CRM Lead should bump its modified timestamp."""
|
|
lead = self._make_lead()
|
|
original_modified = frappe.db.get_value("CRM Lead", lead.name, "modified")
|
|
|
|
time.sleep(0.1)
|
|
|
|
comm = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Received",
|
|
"subject": "Hello",
|
|
"reference_doctype": "CRM Lead",
|
|
"reference_name": lead.name,
|
|
}
|
|
)
|
|
comm.insert(ignore_permissions=True)
|
|
|
|
updated_modified = frappe.db.get_value("CRM Lead", lead.name, "modified")
|
|
self.assertGreater(updated_modified, original_modified)
|
|
|
|
def test_timestamp_not_updated_when_setting_disabled(self):
|
|
"""When the setting is off the timestamp should NOT be updated."""
|
|
frappe.db.set_single_value("FCRM Settings", "update_timestamp_on_new_communication", 0)
|
|
|
|
lead = self._make_lead()
|
|
original_modified = frappe.db.get_value("CRM Lead", lead.name, "modified")
|
|
|
|
time.sleep(0.1)
|
|
|
|
comm = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Received",
|
|
"subject": "No update expected",
|
|
"reference_doctype": "CRM Lead",
|
|
"reference_name": lead.name,
|
|
}
|
|
)
|
|
comm.insert(ignore_permissions=True)
|
|
|
|
updated_modified = frappe.db.get_value("CRM Lead", lead.name, "modified")
|
|
self.assertEqual(updated_modified, original_modified)
|
|
|
|
# Restore
|
|
frappe.db.set_single_value("FCRM Settings", "update_timestamp_on_new_communication", 1)
|
|
|
|
def test_timestamp_updated_on_new_comment(self):
|
|
"""Inserting a Comment linked to a CRM Lead should bump its modified timestamp (via hook)."""
|
|
lead = self._make_lead()
|
|
original_modified = frappe.db.get_value("CRM Lead", lead.name, "modified")
|
|
|
|
time.sleep(0.1)
|
|
|
|
comment = frappe.get_doc(
|
|
{
|
|
"doctype": "Comment",
|
|
"comment_type": "Comment",
|
|
"reference_doctype": "CRM Lead",
|
|
"reference_name": lead.name,
|
|
"content": "A test comment",
|
|
}
|
|
)
|
|
comment.insert(ignore_permissions=True)
|
|
|
|
updated_modified = frappe.db.get_value("CRM Lead", lead.name, "modified")
|
|
self.assertGreater(updated_modified, original_modified)
|
|
|
|
def test_timestamp_not_updated_when_no_reference(self):
|
|
"""A Communication without a reference should not touch the DB."""
|
|
comm = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Received",
|
|
"subject": "No reference",
|
|
}
|
|
)
|
|
with patch("frappe.db.set_value") as mock_set_value:
|
|
# _should_update_modified returns False when no reference — set_value must not be called
|
|
result = _should_update_modified(comm)
|
|
self.assertFalse(result)
|
|
mock_set_value.assert_not_called()
|
|
|
|
def test_timestamp_direct_call_updates_lead(self):
|
|
"""Direct call to on_communication_update updates the reference doc modified timestamp."""
|
|
lead = self._make_lead()
|
|
|
|
time.sleep(0.1)
|
|
|
|
comm = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Received",
|
|
"subject": "Direct call",
|
|
"reference_doctype": "CRM Lead",
|
|
"reference_name": lead.name,
|
|
}
|
|
)
|
|
comm.insert(ignore_permissions=True)
|
|
|
|
before = frappe.db.get_value("CRM Lead", lead.name, "modified")
|
|
time.sleep(0.1)
|
|
on_communication_update(comm)
|
|
after = frappe.db.get_value("CRM Lead", lead.name, "modified")
|
|
self.assertGreaterEqual(after, before)
|
|
|
|
|
|
class TestUpdateCommunicationStatus(IntegrationTestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
frappe.db.set_single_value("FCRM Settings", "auto_reopen_on_new_communication", 1)
|
|
frappe.db.set_single_value("FCRM Settings", "auto_mark_replied_on_response", 1)
|
|
frappe.db.set_single_value("FCRM Settings", "update_timestamp_on_new_communication", 0)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
frappe.db.rollback()
|
|
super().tearDownClass()
|
|
|
|
def tearDown(self):
|
|
frappe.db.rollback()
|
|
frappe.db.set_single_value("FCRM Settings", "auto_reopen_on_new_communication", 1)
|
|
frappe.db.set_single_value("FCRM Settings", "auto_mark_replied_on_response", 1)
|
|
|
|
def _make_lead(self, suffix=""):
|
|
lead = frappe.get_doc(
|
|
{
|
|
"doctype": "CRM Lead",
|
|
"first_name": "CommTest" + suffix,
|
|
"email": f"commtest{suffix}@example.com",
|
|
}
|
|
)
|
|
lead.insert(ignore_permissions=True)
|
|
return lead
|
|
|
|
def _insert_communication(self, lead, sent_or_received):
|
|
comm = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": sent_or_received,
|
|
"subject": f"Test {sent_or_received}",
|
|
"reference_doctype": "CRM Lead",
|
|
"reference_name": lead.name,
|
|
}
|
|
)
|
|
comm.insert(ignore_permissions=True)
|
|
return comm
|
|
|
|
def test_status_set_to_open_on_received_communication(self):
|
|
"""Receiving an incoming communication should set the lead's communication_status to 'Open'."""
|
|
lead = self._make_lead("recv")
|
|
self._insert_communication(lead, "Received")
|
|
|
|
status = frappe.db.get_value("CRM Lead", lead.name, "communication_status")
|
|
self.assertEqual(status, "Open")
|
|
|
|
def test_status_set_to_replied_on_sent_communication(self):
|
|
"""Sending an outgoing communication should set the lead's communication_status to 'Replied'."""
|
|
lead = self._make_lead("sent")
|
|
self._insert_communication(lead, "Sent")
|
|
|
|
status = frappe.db.get_value("CRM Lead", lead.name, "communication_status")
|
|
self.assertEqual(status, "Replied")
|
|
|
|
def test_status_not_updated_when_reopen_setting_disabled(self):
|
|
"""When auto_reopen_on_new_communication is off, a Received communication should NOT set status."""
|
|
frappe.db.set_single_value("FCRM Settings", "auto_reopen_on_new_communication", 0)
|
|
frappe.db.set_single_value("FCRM Settings", "auto_mark_replied_on_response", 0)
|
|
|
|
lead = self._make_lead("reopen-off")
|
|
original_status = frappe.db.get_value("CRM Lead", lead.name, "communication_status")
|
|
self._insert_communication(lead, "Received")
|
|
|
|
status = frappe.db.get_value("CRM Lead", lead.name, "communication_status")
|
|
self.assertEqual(status, original_status)
|
|
|
|
def test_status_not_updated_when_replied_setting_disabled(self):
|
|
"""When auto_mark_replied_on_response is off, a Sent communication should NOT set status."""
|
|
frappe.db.set_single_value("FCRM Settings", "auto_mark_replied_on_response", 0)
|
|
frappe.db.set_single_value("FCRM Settings", "auto_reopen_on_new_communication", 0)
|
|
|
|
lead = self._make_lead("replied-off")
|
|
original_status = frappe.db.get_value("CRM Lead", lead.name, "communication_status")
|
|
self._insert_communication(lead, "Sent")
|
|
|
|
status = frappe.db.get_value("CRM Lead", lead.name, "communication_status")
|
|
self.assertEqual(status, original_status)
|
|
|
|
def test_only_last_communication_drives_status(self):
|
|
"""Only the most recent communication should determine the status."""
|
|
lead = self._make_lead("last")
|
|
|
|
self._insert_communication(lead, "Received")
|
|
self.assertEqual(frappe.db.get_value("CRM Lead", lead.name, "communication_status"), "Open")
|
|
|
|
self._insert_communication(lead, "Sent")
|
|
self.assertEqual(frappe.db.get_value("CRM Lead", lead.name, "communication_status"), "Replied")
|
|
|
|
self._insert_communication(lead, "Received")
|
|
self.assertEqual(frappe.db.get_value("CRM Lead", lead.name, "communication_status"), "Open")
|
|
|
|
def test_status_not_updated_when_no_reference(self):
|
|
"""A Communication with no reference doctype/name should not touch the DB."""
|
|
comm = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Received",
|
|
"subject": "No reference",
|
|
}
|
|
)
|
|
with patch("frappe.db.set_value") as mock_set_value:
|
|
on_communication_update(comm)
|
|
mock_set_value.assert_not_called()
|
|
|
|
def test_status_not_updated_for_non_communication_doctype(self):
|
|
"""Calling _get_communication_status with a non-Communication doc should return None."""
|
|
comment = frappe.get_doc(
|
|
{
|
|
"doctype": "Comment",
|
|
"comment_type": "Comment",
|
|
"reference_doctype": "CRM Lead",
|
|
"reference_name": "LEAD-0001",
|
|
"content": "test",
|
|
}
|
|
)
|
|
self.assertIsNone(_get_communication_status(comment))
|
|
|
|
def test_status_not_updated_for_unknown_sent_or_received_value(self):
|
|
"""A Communication with an unexpected sent_or_received value should be skipped."""
|
|
lead = self._make_lead("invalid")
|
|
|
|
comm = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Received",
|
|
"subject": "Setup",
|
|
"reference_doctype": "CRM Lead",
|
|
"reference_name": lead.name,
|
|
}
|
|
)
|
|
comm.insert(ignore_permissions=True)
|
|
|
|
comm.sent_or_received = "Unknown"
|
|
self.assertIsNone(_get_communication_status(comm))
|
|
|
|
status = frappe.db.get_value("CRM Lead", lead.name, "communication_status")
|
|
self.assertEqual(status, "Open")
|
|
|
|
|
|
class TestCreateLeadFromIncomingEmail(IntegrationTestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
frappe.db.set_single_value("FCRM Settings", "auto_reopen_on_new_communication", 0)
|
|
frappe.db.set_single_value("FCRM Settings", "auto_mark_replied_on_response", 0)
|
|
frappe.db.set_single_value("FCRM Settings", "update_timestamp_on_new_communication", 0)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
frappe.db.rollback()
|
|
super().tearDownClass()
|
|
|
|
def tearDown(self):
|
|
frappe.db.rollback()
|
|
frappe.db.set_single_value("FCRM Settings", "auto_reopen_on_new_communication", 0)
|
|
frappe.db.set_single_value("FCRM Settings", "auto_mark_replied_on_response", 0)
|
|
frappe.db.set_single_value("FCRM Settings", "update_timestamp_on_new_communication", 0)
|
|
|
|
def _make_email_account(self, create_lead=1):
|
|
"""Create a minimal incoming Email Account with the CRM custom field set."""
|
|
email_account = frappe.get_doc(
|
|
{
|
|
"doctype": "Email Account",
|
|
"email_account_name": "Test CRM Incoming",
|
|
"email_id": "test-crm-incoming@example.com",
|
|
"enable_incoming": 1,
|
|
"create_lead_from_incoming_email": create_lead,
|
|
}
|
|
)
|
|
email_account.flags.ignore_mandatory = True
|
|
email_account.flags.ignore_validate = True
|
|
email_account.insert(ignore_permissions=True)
|
|
return email_account
|
|
|
|
def _incoming_comm(self, sender, email_account_name, sender_full_name=None, **kwargs):
|
|
doc = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Received",
|
|
"subject": "Test Incoming Email",
|
|
"sender": sender,
|
|
"email_account": email_account_name,
|
|
**kwargs,
|
|
}
|
|
)
|
|
if sender_full_name:
|
|
doc.sender_full_name = sender_full_name
|
|
return doc
|
|
|
|
def test_lead_created_from_incoming_email(self):
|
|
"""An unreferenced incoming email should create a new CRM Lead."""
|
|
email_account = self._make_email_account()
|
|
doc = self._incoming_comm("newlead@example.com", email_account.name, sender_full_name="New Lead")
|
|
doc.insert(ignore_permissions=True)
|
|
|
|
self.assertTrue(frappe.db.exists("CRM Lead", {"email": "newlead@example.com"}))
|
|
|
|
def test_lead_not_created_when_setting_disabled(self):
|
|
"""When create_lead_from_incoming_email is off on the Email Account, no lead is created."""
|
|
email_account = self._make_email_account(create_lead=0)
|
|
doc = self._incoming_comm("disabled@example.com", email_account.name)
|
|
doc.insert(ignore_permissions=True)
|
|
|
|
self.assertFalse(frappe.db.exists("CRM Lead", {"email": "disabled@example.com"}))
|
|
|
|
def test_lead_not_created_when_no_email_account(self):
|
|
"""A Communication without an email_account should not create a lead."""
|
|
doc = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Received",
|
|
"subject": "No email account",
|
|
"sender": "noaccount@example.com",
|
|
}
|
|
)
|
|
with patch("frappe.new_doc") as mock_new_doc:
|
|
create_lead_from_incoming_email(doc)
|
|
mock_new_doc.assert_not_called()
|
|
|
|
def test_lead_not_created_when_communication_already_referenced(self):
|
|
"""A Communication with an existing reference should not create a new lead."""
|
|
email_account = self._make_email_account()
|
|
existing = frappe.get_doc(
|
|
{"doctype": "CRM Lead", "first_name": "RefLead", "email": "reflead@example.com"}
|
|
)
|
|
existing.insert(ignore_permissions=True)
|
|
|
|
doc = self._incoming_comm(
|
|
"referenced@example.com",
|
|
email_account.name,
|
|
reference_doctype="CRM Lead",
|
|
reference_name=existing.name,
|
|
)
|
|
doc.insert(ignore_permissions=True)
|
|
|
|
self.assertFalse(frappe.db.exists("CRM Lead", {"email": "referenced@example.com"}))
|
|
|
|
def test_lead_not_created_when_lead_already_exists_for_sender(self):
|
|
"""No duplicate lead should be created when a lead with the sender email already exists."""
|
|
email_account = self._make_email_account()
|
|
existing = frappe.get_doc(
|
|
{"doctype": "CRM Lead", "first_name": "Existing", "email": "dupe@example.com"}
|
|
)
|
|
existing.insert(ignore_permissions=True)
|
|
|
|
doc = self._incoming_comm("dupe@example.com", email_account.name)
|
|
doc.insert(ignore_permissions=True)
|
|
|
|
self.assertEqual(frappe.db.count("CRM Lead", {"email": "dupe@example.com"}), 1)
|
|
|
|
def test_lead_not_created_for_non_communication_doctype(self):
|
|
"""Calling the function with a non-Communication doc should be a no-op."""
|
|
comment = frappe.get_doc(
|
|
{
|
|
"doctype": "Comment",
|
|
"comment_type": "Comment",
|
|
"reference_doctype": "CRM Lead",
|
|
"reference_name": "LEAD-0001",
|
|
"content": "test",
|
|
}
|
|
)
|
|
with patch("frappe.new_doc") as mock_new_doc:
|
|
create_lead_from_incoming_email(comment)
|
|
mock_new_doc.assert_not_called()
|
|
|
|
def test_lead_not_created_for_sent_communication_with_non_communication_type(self):
|
|
"""A sent message with a non-Communication type should not create a lead."""
|
|
doc = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Notification",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Sent",
|
|
"subject": "Sent notification",
|
|
"sender": "sent@example.com",
|
|
}
|
|
)
|
|
with patch("frappe.new_doc") as mock_new_doc:
|
|
create_lead_from_incoming_email(doc)
|
|
mock_new_doc.assert_not_called()
|
|
|
|
def test_lead_first_name_from_sender_full_name(self):
|
|
"""The lead's first_name should come from sender_full_name when present."""
|
|
email_account = self._make_email_account()
|
|
doc = self._incoming_comm("fullname@example.com", email_account.name, sender_full_name="Jane Doe")
|
|
create_lead_from_incoming_email(doc)
|
|
|
|
lead = frappe.db.get_values(
|
|
"CRM Lead", {"email": "fullname@example.com"}, ["first_name", "last_name"], as_dict=True
|
|
)
|
|
self.assertEqual(lead, [{"first_name": "Jane", "last_name": "Doe"}])
|
|
|
|
def test_lead_first_name_falls_back_to_email_prefix(self):
|
|
"""When sender_full_name is absent, the email prefix should be used as first_name."""
|
|
email_account = self._make_email_account()
|
|
doc = self._incoming_comm("prefix@example.com", email_account.name)
|
|
create_lead_from_incoming_email(doc)
|
|
|
|
lead = frappe.db.get_value("CRM Lead", {"email": "prefix@example.com"}, "first_name")
|
|
self.assertEqual(lead, "prefix")
|
|
|
|
def test_lead_last_name_empty_for_single_word_sender_full_name(self):
|
|
"""When sender_full_name is a single word, last_name should be empty."""
|
|
email_account = self._make_email_account()
|
|
doc = self._incoming_comm("singlename@example.com", email_account.name, sender_full_name="Mononym")
|
|
create_lead_from_incoming_email(doc)
|
|
|
|
lead = frappe.db.get_values(
|
|
"CRM Lead", {"email": "singlename@example.com"}, ["first_name", "last_name"], as_dict=True
|
|
)
|
|
self.assertEqual(lead, [{"first_name": "Mononym", "last_name": ""}])
|
|
|
|
def test_communication_linked_back_to_created_lead(self):
|
|
"""After lead creation, the communication's reference_doctype and reference_name should point to the new lead."""
|
|
email_account = self._make_email_account()
|
|
doc = self._incoming_comm("linked@example.com", email_account.name, sender_full_name="Link Test")
|
|
create_lead_from_incoming_email(doc)
|
|
|
|
self.assertEqual(doc.reference_doctype, "CRM Lead")
|
|
lead_name = frappe.db.get_value("CRM Lead", {"email": "linked@example.com"}, "name")
|
|
self.assertEqual(doc.reference_name, lead_name)
|
|
|
|
def test_lead_source_set_to_email_when_source_exists(self):
|
|
"""Lead source should be set to 'Email' when the CRM Lead Source 'Email' exists."""
|
|
if not frappe.db.exists("CRM Lead Source", "Email"):
|
|
frappe.get_doc({"doctype": "CRM Lead Source", "name": "Email"}).insert(ignore_permissions=True)
|
|
|
|
email_account = self._make_email_account()
|
|
doc = self._incoming_comm("leadsource@example.com", email_account.name)
|
|
create_lead_from_incoming_email(doc)
|
|
|
|
source = frappe.db.get_value("CRM Lead", {"email": "leadsource@example.com"}, "source")
|
|
self.assertEqual(source, "Email")
|
|
|
|
def test_lead_created_for_sent_communication_with_communication_type(self):
|
|
"""A sent communication with communication_type='Communication' should still create a lead.
|
|
|
|
The guard condition uses AND: both sent_or_received != 'Received' AND
|
|
communication_type != 'Communication' must be true to bail out. When the
|
|
type IS 'Communication', the second condition is false and the function proceeds.
|
|
"""
|
|
email_account = self._make_email_account()
|
|
doc = frappe.get_doc(
|
|
{
|
|
"doctype": "Communication",
|
|
"communication_type": "Communication",
|
|
"communication_medium": "Email",
|
|
"sent_or_received": "Sent",
|
|
"subject": "Sent communication type",
|
|
"sender": "sentcomm@example.com",
|
|
"email_account": email_account.name,
|
|
}
|
|
)
|
|
create_lead_from_incoming_email(doc)
|
|
|
|
self.assertTrue(frappe.db.exists("CRM Lead", {"email": "sentcomm@example.com"}))
|