# Copyright 2016 Antonio Espinosa - # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html). import base64 import time import mock import psycopg2 import psycopg2.errorcodes from lxml import etree from odoo import http from odoo.tests.common import TransactionCase from odoo.tools import mute_logger from ..controllers.main import BLANK, MailTrackingController mock_send_email = "odoo.addons.base.models.ir_mail_server." "IrMailServer.send_email" class FakeUserAgent(object): browser = "Test browser" platform = "Test platform" def __str__(self): """Return name""" return "Test suite" class TestMailTracking(TransactionCase): def setUp(self, *args, **kwargs): super(TestMailTracking, self).setUp(*args, **kwargs) self.sender = self.env["res.partner"].create( {"name": "Test sender", "email": "sender@example.com"} ) self.recipient = self.env["res.partner"].create( {"name": "Test recipient", "email": "recipient@example.com"} ) self.last_request = http.request http.request = type( "obj", (object,), { "env": self.env, "cr": self.env.cr, "db": self.env.cr.dbname, "endpoint": type("obj", (object,), {"routing": []}), "httprequest": type( "obj", (object,), {"remote_addr": "123.123.123.123", "user_agent": FakeUserAgent()}, ), }, ) def tearDown(self, *args, **kwargs): http.request = self.last_request return super(TestMailTracking, self).tearDown(*args, **kwargs) def test_empty_email(self): self.recipient.write({"email_bounced": True}) self.recipient.write({"email": False}) self.assertEqual(False, self.recipient.email) self.assertEqual(False, self.recipient.email_bounced) self.recipient.write({"email_bounced": True}) self.recipient.write({"email": ""}) self.assertEqual(False, self.recipient.email_bounced) self.assertEqual(False, self.env["mail.tracking.email"].email_is_bounced(False)) self.assertEqual( 0.0, self.env["mail.tracking.email"].email_score_from_email(False) ) def test_recipient_address_compute(self): mail, tracking = self.mail_send(self.recipient.email) tracking.write({"recipient": False}) self.assertEqual(False, tracking.recipient_address) def test_message_post(self): # This message will generate a notification for recipient message = self.env["mail.message"].create( { "subject": "Message test", "author_id": self.sender.id, "email_from": self.sender.email, "message_type": "comment", "model": "res.partner", "res_id": self.recipient.id, "partner_ids": [(4, self.recipient.id)], "body": "

This is a test message

", } ) message._moderate_accept() # Search tracking created tracking_email = self.env["mail.tracking.email"].search( [ ("mail_message_id", "=", message.id), ("partner_id", "=", self.recipient.id), ] ) # The tracking email must be sent self.assertTrue(tracking_email) self.assertEqual(tracking_email.state, "sent") # message_dict read by web interface message_dict = message.message_format()[0] self.assertTrue(len(message_dict["partner_ids"]) > 0) # First partner is recipient partner_id = message_dict["partner_ids"][0] self.assertEqual(partner_id, self.recipient.id) status = message_dict["partner_trackings"][0] # Tracking status must be sent and # mail tracking must be the one search before self.assertEqual(status["status"], "sent") self.assertEqual(status["tracking_id"], tracking_email.id) self.assertEqual(status["recipient"], self.recipient.display_name) self.assertEqual(status["partner_id"], self.recipient.id) self.assertEqual(status["isCc"], False) # And now open the email metadata = { "ip": "127.0.0.1", "user_agent": "Odoo Test/1.0", "os_family": "linux", "ua_family": "odoo", } tracking_email.event_create("open", metadata) self.assertEqual(tracking_email.state, "opened") def test_message_post_partner_no_email(self): # Create message with recipient without defined email self.recipient.write({"email": False}) message = self.env["mail.message"].create( { "subject": "Message test", "author_id": self.sender.id, "email_from": self.sender.email, "message_type": "comment", "model": "res.partner", "res_id": self.recipient.id, "partner_ids": [(4, self.recipient.id)], "body": "

This is a test message

", } ) message._moderate_accept() # Search tracking created tracking_email = self.env["mail.tracking.email"].search( [ ("mail_message_id", "=", message.id), ("partner_id", "=", self.recipient.id), ] ) # No email should generate a error state: no_recipient self.assertEqual(tracking_email.state, "error") self.assertEqual(tracking_email.error_type, "no_recipient") self.assertFalse(self.recipient.email_bounced) def _check_partner_trackings_cc(self, message): message_dict = message.message_format()[0] self.assertEqual(len(message_dict["partner_trackings"]), 3) # mail cc foundPartner = False foundNoPartner = False for tracking in message_dict["partner_trackings"]: if tracking["partner_id"] == self.sender.id: foundPartner = True self.assertTrue(tracking["isCc"]) elif tracking["recipient"] == "unnamed@test.com": foundNoPartner = True self.assertFalse(tracking["partner_id"]) self.assertTrue(tracking["isCc"]) elif tracking["partner_id"] == self.recipient.id: self.assertFalse(tracking["isCc"]) self.assertTrue(foundPartner) self.assertTrue(foundNoPartner) def test_email_cc(self): sender_user = self.env["res.users"].create( { "name": "Sender User Test", "partner_id": self.sender.id, "login": "sender-test", } ) # pylint: disable=C8107 message = self.recipient.with_user(sender_user).message_post( body="

This is a test message

", cc="Dominique Pinon , sender@example.com", ) # suggested recipients recipients = self.recipient._message_get_suggested_recipients() suggested_mails = {email[1] for email in recipients[self.recipient.id]} self.assertIn("unnamed@test.com", suggested_mails) self.assertEqual(len(recipients[self.recipient.id]), 3) # Repeated Cc recipients message = self.env["mail.message"].create( { "subject": "Message test", "author_id": self.sender.id, "email_from": self.sender.email, "message_type": "comment", "model": "res.partner", "res_id": self.recipient.id, "partner_ids": [(4, self.recipient.id)], "email_cc": "Dominique Pinon , sender@example.com" ", recipient@example.com", "body": "

This is another test message

", } ) message._moderate_accept() recipients = self.recipient._message_get_suggested_recipients() self.assertEqual(len(recipients[self.recipient.id]), 3) self._check_partner_trackings_cc(message) def _check_partner_trackings_to(self, message): message_dict = message.message_format()[0] self.assertEqual(len(message_dict["partner_trackings"]), 3) # mail cc foundPartner = False foundNoPartner = False for tracking in message_dict["partner_trackings"]: if tracking["partner_id"] == self.sender.id: foundPartner = True elif tracking["recipient"] == "support+unnamed@test.com": foundNoPartner = True self.assertFalse(tracking["partner_id"]) self.assertTrue(foundPartner) self.assertTrue(foundNoPartner) def test_email_to(self): sender_user = self.env["res.users"].create( { "name": "Sender User Test", "partner_id": self.sender.id, "login": "sender-test", } ) # pylint: disable=C8107 message = self.recipient.with_user(sender_user).message_post( body="

This is a test message

", to="Dominique Pinon , sender@example.com", ) # suggested recipients recipients = self.recipient._message_get_suggested_recipients() suggested_mails = {email[1] for email in recipients[self.recipient.id]} self.assertIn("support+unnamed@test.com", suggested_mails) self.assertEqual(len(recipients[self.recipient.id]), 3) # Repeated To recipients message = self.env["mail.message"].create( { "subject": "Message test", "author_id": self.sender.id, "email_from": self.sender.email, "message_type": "comment", "model": "res.partner", "res_id": self.recipient.id, "partner_ids": [(4, self.recipient.id)], "email_to": "Dominique Pinon " ", sender@example.com, recipient@example.com", "body": "

This is another test message

", } ) message._moderate_accept() recipients = self.recipient._message_get_suggested_recipients() self.assertEqual(len(recipients[self.recipient.id]), 3) self._check_partner_trackings_to(message) # Catchall + Alias self.env["ir.config_parameter"].set_param("mail.catchall.domain", "test.com") self.env["mail.alias"].create( { "alias_model_id": self.env["ir.model"]._get("res.partner").id, "alias_name": "support+unnamed", } ) recipients = self.recipient._message_get_suggested_recipients() self.assertEqual(len(recipients[self.recipient.id]), 2) suggested_mails = {email[1] for email in recipients[self.recipient.id]} self.assertNotIn("support+unnamed@test.com", suggested_mails) def test_failed_message(self): MailMessageObj = self.env["mail.message"] # Create message mail, tracking = self.mail_send(self.recipient.email) self.assertFalse(tracking.mail_message_id.mail_tracking_needs_action) # Force error state tracking.state = "error" self.assertTrue(tracking.mail_message_id.mail_tracking_needs_action) failed_count = MailMessageObj.get_failed_count() self.assertTrue(failed_count > 0) values = tracking.mail_message_id.get_failed_messages() self.assertEqual(values[0]["id"], tracking.mail_message_id.id) messages = MailMessageObj.search([]) messages_failed = MailMessageObj.search( MailMessageObj._get_failed_message_domain() ) self.assertTrue(messages) self.assertTrue(messages_failed) self.assertTrue(len(messages) > len(messages_failed)) tracking.mail_message_id.set_need_action_done() self.assertFalse(tracking.mail_message_id.mail_tracking_needs_action) self.assertTrue(MailMessageObj.get_failed_count() < failed_count) # No author_id tracking.mail_message_id.author_id = False values = tracking.mail_message_id.get_failed_messages()[0] self.assertEqual(values["author"][0], -1) def test_resend_failed_message(self): # This message will generate a notification for recipient message = self.env["mail.message"].create( { "subject": "Message test", "author_id": self.sender.id, "email_from": self.sender.email, "message_type": "comment", "model": "res.partner", "res_id": self.recipient.id, "partner_ids": [(4, self.recipient.id)], "body": "

This is a test message

", } ) message._moderate_accept() # Search tracking created tracking_email = self.env["mail.tracking.email"].search( [ ("mail_message_id", "=", message.id), ("partner_id", "=", self.recipient.id), ] ) # Force error state tracking_email.state = "error" # Create resend mail wizard wizard = ( self.env["mail.resend.message"] .sudo() .with_context({"mail_message_to_resend": message.id}) .create({}) ) # Check failed recipient)s self.assertTrue(any(wizard.partner_ids)) self.assertEqual(self.recipient.email, wizard.partner_ids[0].email) # Resend message wizard.resend_mail_action() # Check tracking reset self.assertFalse(tracking_email.state) def mail_send(self, recipient): mail = self.env["mail.mail"].create( { "subject": "Test subject", "email_from": "from@domain.com", "email_to": recipient, "body_html": "

This is a test message

", } ) mail.send() # Search tracking created tracking_email = self.env["mail.tracking.email"].search( [("mail_id", "=", mail.id)] ) return mail, tracking_email def test_mail_send(self): controller = MailTrackingController() db = self.env.cr.dbname image = base64.b64decode(BLANK) mail, tracking = self.mail_send(self.recipient.email) self.assertEqual(mail.email_to, tracking.recipient) self.assertEqual(mail.email_from, tracking.sender) with mock.patch("odoo.http.db_filter") as mock_client: mock_client.return_value = True res = controller.mail_tracking_open(db, tracking.id, tracking.token) self.assertEqual(image, res.response[0]) # Two events: sent and open self.assertEqual(2, len(tracking.tracking_event_ids)) # Fake event: tracking_email_id = False res = controller.mail_tracking_open(db, False, False) self.assertEqual(image, res.response[0]) # Two events again because no tracking_email_id found for False self.assertEqual(2, len(tracking.tracking_event_ids)) def test_mail_tracking_open(self): controller = MailTrackingController() db = self.env.cr.dbname with mock.patch("odoo.http.db_filter") as mock_client: mock_client.return_value = True mail, tracking = self.mail_send(self.recipient.email) # Tracking is in sent or delivered state. But no token give. # Don't generates tracking event controller.mail_tracking_open(db, tracking.id) self.assertEqual(1, len(tracking.tracking_event_ids)) tracking.write({"state": "opened"}) # Tracking isn't in sent or delivered state. # Don't generates tracking event controller.mail_tracking_open(db, tracking.id, tracking.token) self.assertEqual(1, len(tracking.tracking_event_ids)) tracking.write({"state": "sent"}) # Tracking is in sent or delivered state and a token is given. # Generates tracking event controller.mail_tracking_open(db, tracking.id, tracking.token) self.assertEqual(2, len(tracking.tracking_event_ids)) # Generate new email due concurrent event filter mail, tracking = self.mail_send(self.recipient.email) tracking.write({"token": False}) # Tracking is in sent or delivered state but a token is given for a # record that doesn't have a token. # Don't generates tracking event controller.mail_tracking_open(db, tracking.id, "tokentest") self.assertEqual(1, len(tracking.tracking_event_ids)) # Tracking is in sent or delivered state and not token is given for # a record that doesn't have a token. # Generates tracking event controller.mail_tracking_open(db, tracking.id, False) self.assertEqual(2, len(tracking.tracking_event_ids)) def test_concurrent_open(self): mail, tracking = self.mail_send(self.recipient.email) ts = time.time() metadata = { "ip": "127.0.0.1", "user_agent": "Odoo Test/1.0", "os_family": "linux", "ua_family": "odoo", "timestamp": ts, } # First open event tracking.event_create("open", metadata) opens = tracking.tracking_event_ids.filtered(lambda r: r.event_type == "open") self.assertEqual(len(opens), 1) # Concurrent open event metadata["timestamp"] = ts + 2 tracking.event_create("open", metadata) opens = tracking.tracking_event_ids.filtered(lambda r: r.event_type == "open") self.assertEqual(len(opens), 1) # Second open event metadata["timestamp"] = ts + 350 tracking.event_create("open", metadata) opens = tracking.tracking_event_ids.filtered(lambda r: r.event_type == "open") self.assertEqual(len(opens), 2) def test_concurrent_click(self): mail, tracking = self.mail_send(self.recipient.email) ts = time.time() metadata = { "ip": "127.0.0.1", "user_agent": "Odoo Test/1.0", "os_family": "linux", "ua_family": "odoo", "timestamp": ts, "url": "https://www.example.com/route/1", } # First click event (URL 1) tracking.event_create("click", metadata) opens = tracking.tracking_event_ids.filtered(lambda r: r.event_type == "click") self.assertEqual(len(opens), 1) # Concurrent click event (URL 1) metadata["timestamp"] = ts + 2 tracking.event_create("click", metadata) opens = tracking.tracking_event_ids.filtered(lambda r: r.event_type == "click") self.assertEqual(len(opens), 1) # Second click event (URL 1) metadata["timestamp"] = ts + 350 tracking.event_create("click", metadata) opens = tracking.tracking_event_ids.filtered(lambda r: r.event_type == "click") self.assertEqual(len(opens), 2) # Concurrent click event (URL 2) metadata["timestamp"] = ts + 2 metadata["url"] = "https://www.example.com/route/2" tracking.event_create("click", metadata) opens = tracking.tracking_event_ids.filtered(lambda r: r.event_type == "click") self.assertEqual(len(opens), 3) @mute_logger("odoo.addons.mail.models.mail_mail") def test_smtp_error(self): with mock.patch(mock_send_email) as mock_func: mock_func.side_effect = Warning("Test error") mail, tracking = self.mail_send(self.recipient.email) self.assertEqual("error", tracking.state) self.assertEqual("Warning", tracking.error_type) self.assertEqual("Test error", tracking.error_description) self.assertTrue(self.recipient.email_bounced) def test_partner_email_change(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("open", {}) orig_score = self.recipient.email_score orig_count = self.recipient.tracking_emails_count orig_email = self.recipient.email self.recipient.email = orig_email + "2" self.assertEqual(50.0, self.recipient.email_score) self.assertEqual(0, self.recipient.tracking_emails_count) self.recipient.email = orig_email self.assertEqual(orig_score, self.recipient.email_score) self.assertEqual(orig_count, self.recipient.tracking_emails_count) def test_process_hard_bounce(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("hard_bounce", {}) self.assertEqual("bounced", tracking.state) self.assertTrue(self.recipient.email_score < 50.0) def test_process_soft_bounce(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("soft_bounce", {}) self.assertEqual("soft-bounced", tracking.state) self.assertTrue(self.recipient.email_score < 50.0) def test_process_delivered(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("delivered", {}) self.assertEqual("delivered", tracking.state) self.assertTrue(self.recipient.email_score > 50.0) def test_process_deferral(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("deferral", {}) self.assertEqual("deferred", tracking.state) def test_process_spam(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("spam", {}) self.assertEqual("spam", tracking.state) self.assertTrue(self.recipient.email_score < 50.0) def test_process_unsub(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("unsub", {}) self.assertEqual("unsub", tracking.state) self.assertTrue(self.recipient.email_score < 50.0) def test_process_reject(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("reject", {}) self.assertEqual("rejected", tracking.state) self.assertTrue(self.recipient.email_score < 50.0) def test_process_open(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("open", {}) self.assertEqual("opened", tracking.state) self.assertTrue(self.recipient.email_score > 50.0) def test_process_click(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("click", {}) self.assertEqual("opened", tracking.state) self.assertTrue(self.recipient.email_score > 50.0) def test_process_several_bounce(self): for _i in range(1, 10): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("hard_bounce", {}) self.assertEqual("bounced", tracking.state) self.assertEqual(0.0, self.recipient.email_score) def test_bounce_new_partner(self): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("hard_bounce", {}) new_partner = self.env["res.partner"].create({"name": "Test New Partner"}) new_partner.email = self.recipient.email self.assertTrue(new_partner.email_bounced) def test_recordset_email_score(self): """For backwords compatibility sake""" trackings = self.env["mail.tracking.email"] for _i in range(11): mail, tracking = self.mail_send(self.recipient.email) tracking.event_create("click", {}) trackings |= tracking self.assertEqual(100.0, trackings.email_score()) def test_db(self): db = self.env.cr.dbname controller = MailTrackingController() with mock.patch("odoo.http.db_filter") as mock_client: mock_client.return_value = True with self.assertRaises(psycopg2.OperationalError): controller.mail_tracking_event("not_found_db") none = controller.mail_tracking_event(db) self.assertEqual(b"NONE", none.response[0]) none = controller.mail_tracking_event(db, "open") self.assertEqual(b"NONE", none.response[0]) class TestMailTrackingViews(TransactionCase): def test_fields_view_get(self): result = self.env["res.partner"].fields_view_get( view_id=self.env.ref("base.view_partner_form").id, view_type="form" ) doc = etree.XML(result["arch"]) nodes = doc.xpath( "//field[@name='failed_message_ids'" " and @widget='mail_failed_message']" ) self.assertTrue(nodes) result = self.env["res.partner"].fields_view_get( view_id=self.env.ref("base.view_res_partner_filter").id, view_type="search" ) doc = etree.XML(result["arch"]) nodes = doc.xpath("//filter[@name='failed_message_ids']") self.assertTrue(nodes)