diff --git a/frappe/core/doctype/communication/test_communication.py b/frappe/core/doctype/communication/test_communication.py index 48a6e9b636..aa349aaa30 100644 --- a/frappe/core/doctype/communication/test_communication.py +++ b/frappe/core/doctype/communication/test_communication.py @@ -1,12 +1,14 @@ # Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors # License: MIT. See LICENSE +from datetime import timedelta from typing import TYPE_CHECKING import frappe from frappe.core.doctype.communication.communication import Communication, get_emails, parse_email -from frappe.core.doctype.communication.email import add_attachments, make +from frappe.core.doctype.communication.email import add_attachments, make, undo_email_send from frappe.email.doctype.email_queue.email_queue import EmailQueue from frappe.tests import IntegrationTestCase +from frappe.utils import now_datetime if TYPE_CHECKING: from frappe.contacts.doctype.contact.contact import Contact @@ -438,6 +440,85 @@ class TestCommunicationEmailMixin(IntegrationTestCase): self.assertEqual(attached_file.file_name, file_name) self.assertEqual(attached_file.get_content(), file_content) + def test_undo_email_send(self): + """Undo should delete Communication and Email Queue, and return original data.""" + comm = self.new_communication(recipients=["to@test.com"]) + comm.sent_or_received = "Sent" + comm.save(ignore_permissions=True) + + eq = frappe.get_doc( + { + "doctype": "Email Queue", + "sender": "Test ", + "message": "Test message", + "status": "Not Sent", + "priority": 1, + "communication": comm.name, + "recipients": [{"recipient": "to@test.com", "status": "Not Sent"}], + } + ).insert(ignore_permissions=True) + + result = undo_email_send(comm.name) + + self.assertFalse(frappe.db.exists("Communication", comm.name)) + self.assertFalse(frappe.db.exists("Email Queue", eq.name)) + self.assertFalse(frappe.db.exists("Email Queue Recipient", {"parent": eq.name})) + self.assertEqual(result["subject"], comm.subject) + self.assertEqual(result["recipients"], comm.recipients) + + def test_undo_email_send_fails_for_different_user(self): + """Undo should fail if the current user is not the owner.""" + comm = self.new_communication(recipients=["to@test.com"]) + comm.sent_or_received = "Sent" + comm.save(ignore_permissions=True) + frappe.db.set_value("Communication", comm.name, "owner", "other@test.com") + + with self.assertRaises(frappe.exceptions.ValidationError): + undo_email_send(comm.name) + + self.assertTrue(frappe.db.exists("Communication", comm.name)) + + def test_undo_email_send_fails_after_time_window(self): + """Undo should fail if the 10-second window has passed.""" + comm = self.new_communication(recipients=["to@test.com"]) + comm.sent_or_received = "Sent" + comm.save(ignore_permissions=True) + frappe.db.set_value( + "Communication", + comm.name, + "creation", + now_datetime() - timedelta(seconds=15), + update_modified=False, + ) + + with self.assertRaises(frappe.exceptions.ValidationError): + undo_email_send(comm.name) + + self.assertTrue(frappe.db.exists("Communication", comm.name)) + + def test_undo_email_send_fails_if_already_sent(self): + """Undo should fail if Email Queue status is not 'Not Sent'.""" + comm = self.new_communication(recipients=["to@test.com"]) + comm.sent_or_received = "Sent" + comm.save(ignore_permissions=True) + + frappe.get_doc( + { + "doctype": "Email Queue", + "sender": "Test ", + "message": "Test message", + "status": "Sent", + "priority": 1, + "communication": comm.name, + "recipients": [{"recipient": "to@test.com", "status": "Sent"}], + } + ).insert(ignore_permissions=True) + + with self.assertRaises(frappe.exceptions.ValidationError): + undo_email_send(comm.name) + + self.assertTrue(frappe.db.exists("Communication", comm.name)) + def create_email_account() -> "EmailAccount": frappe.delete_doc_if_exists("Email Account", "_Test Comm Account 1")