You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

323 lines
13 KiB

  1. # -*- coding: utf-8 -*-
  2. # Copyright 2016 Antonio Espinosa - <antonio.espinosa@tecnativa.com>
  3. # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
  4. import mock
  5. from odoo.tools import mute_logger
  6. import time
  7. from odoo import http
  8. from odoo.tests.common import TransactionCase
  9. from ..controllers.main import MailTrackingController, BLANK
  10. mock_send_email = ('odoo.addons.base.ir.ir_mail_server.'
  11. 'IrMailServer.send_email')
  12. class FakeUserAgent(object):
  13. browser = 'Test browser'
  14. platform = 'Test platform'
  15. def __str__(self):
  16. """Return name"""
  17. return 'Test suite'
  18. class TestMailTracking(TransactionCase):
  19. def setUp(self, *args, **kwargs):
  20. super(TestMailTracking, self).setUp(*args, **kwargs)
  21. self.sender = self.env['res.partner'].create({
  22. 'name': 'Test sender',
  23. 'email': 'sender@example.com',
  24. })
  25. self.recipient = self.env['res.partner'].create({
  26. 'name': 'Test recipient',
  27. 'email': 'recipient@example.com',
  28. })
  29. self.last_request = http.request
  30. http.request = type('obj', (object,), {
  31. 'db': self.env.cr.dbname,
  32. 'env': self.env,
  33. 'endpoint': type('obj', (object,), {
  34. 'routing': [],
  35. }),
  36. 'httprequest': type('obj', (object,), {
  37. 'remote_addr': '123.123.123.123',
  38. 'user_agent': FakeUserAgent(),
  39. }),
  40. })
  41. def tearDown(self, *args, **kwargs):
  42. http.request = self.last_request
  43. return super(TestMailTracking, self).tearDown(*args, **kwargs)
  44. def test_email_lower(self):
  45. self.recipient.write({'email': 'UPPER@example.com'})
  46. self.assertEqual('upper@example.com', self.recipient.email)
  47. def test_empty_email(self):
  48. self.recipient.write({'email_bounced': True})
  49. self.recipient.write({'email': False})
  50. self.assertEqual(False, self.recipient.email)
  51. self.assertEqual(False, self.recipient.email_bounced)
  52. self.recipient.write({'email_bounced': True})
  53. self.recipient.write({'email': ''})
  54. self.assertEqual(False, self.recipient.email)
  55. self.assertEqual(False, self.recipient.email_bounced)
  56. self.assertEqual(
  57. False,
  58. self.env['mail.tracking.email'].email_is_bounced(False))
  59. self.assertEqual(
  60. 0.,
  61. self.env['mail.tracking.email'].email_score_from_email(False))
  62. def test_recipient_address_compute(self):
  63. mail, tracking = self.mail_send(self.recipient.email)
  64. tracking.write({'recipient': False})
  65. self.assertEqual(False, tracking.recipient_address)
  66. def test_message_post(self):
  67. # This message will generate a notification for recipient
  68. message = self.env['mail.message'].create({
  69. 'subject': 'Message test',
  70. 'author_id': self.sender.id,
  71. 'email_from': self.sender.email,
  72. 'message_type': 'comment',
  73. 'model': 'res.partner',
  74. 'res_id': self.recipient.id,
  75. 'partner_ids': [(4, self.recipient.id)],
  76. 'body': '<p>This is a test message</p>',
  77. })
  78. # Search tracking created
  79. tracking_email = self.env['mail.tracking.email'].search([
  80. ('mail_message_id', '=', message.id),
  81. ('partner_id', '=', self.recipient.id),
  82. ])
  83. # The tracking email must be sent
  84. self.assertTrue(tracking_email)
  85. self.assertEqual(tracking_email.state, 'sent')
  86. # message_dict read by web interface
  87. message_dict = message.message_format()[0]
  88. self.assertTrue(len(message_dict['partner_ids']) > 0)
  89. # First partner is recipient
  90. partner_id = message_dict['partner_ids'][0][0]
  91. self.assertEqual(partner_id, self.recipient.id)
  92. status = message_dict['partner_trackings'][0]
  93. # Tracking status must be sent and
  94. # mail tracking must be the one search before
  95. self.assertEqual(status[0], 'sent')
  96. self.assertEqual(status[1], tracking_email.id)
  97. self.assertEqual(status[2], self.recipient.display_name)
  98. self.assertEqual(status[3], self.recipient.id)
  99. # And now open the email
  100. metadata = {
  101. 'ip': '127.0.0.1',
  102. 'user_agent': 'Odoo Test/1.0',
  103. 'os_family': 'linux',
  104. 'ua_family': 'odoo',
  105. }
  106. tracking_email.event_create('open', metadata)
  107. self.assertEqual(tracking_email.state, 'opened')
  108. def mail_send(self, recipient):
  109. mail = self.env['mail.mail'].create({
  110. 'subject': 'Test subject',
  111. 'email_from': 'from@domain.com',
  112. 'email_to': recipient,
  113. 'body_html': '<p>This is a test message</p>',
  114. })
  115. mail.send()
  116. # Search tracking created
  117. tracking_email = self.env['mail.tracking.email'].search([
  118. ('mail_id', '=', mail.id),
  119. ])
  120. return mail, tracking_email
  121. def test_mail_send(self):
  122. controller = MailTrackingController()
  123. db = self.env.cr.dbname
  124. image = BLANK
  125. mail, tracking = self.mail_send(self.recipient.email)
  126. self.assertEqual(mail.email_to, tracking.recipient)
  127. self.assertEqual(mail.email_from, tracking.sender)
  128. res = controller.mail_tracking_open(db, tracking.id)
  129. self.assertEqual(image, res.response[0])
  130. # Two events: sent and open
  131. self.assertEqual(2, len(tracking.tracking_event_ids))
  132. # Fake event: tracking_email_id = False
  133. res = controller.mail_tracking_open(db, False)
  134. self.assertEqual(image, res.response[0])
  135. # Two events again because no tracking_email_id found for False
  136. self.assertEqual(2, len(tracking.tracking_event_ids))
  137. def test_concurrent_open(self):
  138. mail, tracking = self.mail_send(self.recipient.email)
  139. ts = time.time()
  140. metadata = {
  141. 'ip': '127.0.0.1',
  142. 'user_agent': 'Odoo Test/1.0',
  143. 'os_family': 'linux',
  144. 'ua_family': 'odoo',
  145. 'timestamp': ts,
  146. }
  147. # First open event
  148. tracking.event_create('open', metadata)
  149. opens = tracking.tracking_event_ids.filtered(
  150. lambda r: r.event_type == 'open'
  151. )
  152. self.assertEqual(len(opens), 1)
  153. # Concurrent open event
  154. metadata['timestamp'] = ts + 2
  155. tracking.event_create('open', metadata)
  156. opens = tracking.tracking_event_ids.filtered(
  157. lambda r: r.event_type == 'open'
  158. )
  159. self.assertEqual(len(opens), 1)
  160. # Second open event
  161. metadata['timestamp'] = ts + 350
  162. tracking.event_create('open', metadata)
  163. opens = tracking.tracking_event_ids.filtered(
  164. lambda r: r.event_type == 'open'
  165. )
  166. self.assertEqual(len(opens), 2)
  167. def test_concurrent_click(self):
  168. mail, tracking = self.mail_send(self.recipient.email)
  169. ts = time.time()
  170. metadata = {
  171. 'ip': '127.0.0.1',
  172. 'user_agent': 'Odoo Test/1.0',
  173. 'os_family': 'linux',
  174. 'ua_family': 'odoo',
  175. 'timestamp': ts,
  176. 'url': 'https://www.example.com/route/1',
  177. }
  178. # First click event (URL 1)
  179. tracking.event_create('click', metadata)
  180. opens = tracking.tracking_event_ids.filtered(
  181. lambda r: r.event_type == 'click'
  182. )
  183. self.assertEqual(len(opens), 1)
  184. # Concurrent click event (URL 1)
  185. metadata['timestamp'] = ts + 2
  186. tracking.event_create('click', metadata)
  187. opens = tracking.tracking_event_ids.filtered(
  188. lambda r: r.event_type == 'click'
  189. )
  190. self.assertEqual(len(opens), 1)
  191. # Second click event (URL 1)
  192. metadata['timestamp'] = ts + 350
  193. tracking.event_create('click', metadata)
  194. opens = tracking.tracking_event_ids.filtered(
  195. lambda r: r.event_type == 'click'
  196. )
  197. self.assertEqual(len(opens), 2)
  198. # Concurrent click event (URL 2)
  199. metadata['timestamp'] = ts + 2
  200. metadata['url'] = 'https://www.example.com/route/2'
  201. tracking.event_create('click', metadata)
  202. opens = tracking.tracking_event_ids.filtered(
  203. lambda r: r.event_type == 'click'
  204. )
  205. self.assertEqual(len(opens), 3)
  206. @mute_logger('odoo.addons.mail.models.mail_mail')
  207. def test_smtp_error(self):
  208. with mock.patch(mock_send_email) as mock_func:
  209. mock_func.side_effect = Warning('Test error')
  210. mail, tracking = self.mail_send(self.recipient.email)
  211. self.assertEqual('error', tracking.state)
  212. self.assertEqual('Warning', tracking.error_type)
  213. self.assertEqual('Test error', tracking.error_description)
  214. self.assertTrue(self.recipient.email_bounced)
  215. def test_partner_email_change(self):
  216. mail, tracking = self.mail_send(self.recipient.email)
  217. tracking.event_create('open', {})
  218. orig_score = self.recipient.email_score
  219. orig_count = self.recipient.tracking_emails_count
  220. orig_email = self.recipient.email
  221. self.recipient.email = orig_email + '2'
  222. self.assertEqual(50.0, self.recipient.email_score)
  223. self.assertEqual(0, self.recipient.tracking_emails_count)
  224. self.recipient.email = orig_email
  225. self.assertEqual(orig_score, self.recipient.email_score)
  226. self.assertEqual(orig_count, self.recipient.tracking_emails_count)
  227. def test_process_hard_bounce(self):
  228. mail, tracking = self.mail_send(self.recipient.email)
  229. tracking.event_create('hard_bounce', {})
  230. self.assertEqual('bounced', tracking.state)
  231. self.assertTrue(self.recipient.email_score < 50.0)
  232. def test_process_soft_bounce(self):
  233. mail, tracking = self.mail_send(self.recipient.email)
  234. tracking.event_create('soft_bounce', {})
  235. self.assertEqual('soft-bounced', tracking.state)
  236. self.assertTrue(self.recipient.email_score < 50.0)
  237. def test_process_delivered(self):
  238. mail, tracking = self.mail_send(self.recipient.email)
  239. tracking.event_create('delivered', {})
  240. self.assertEqual('delivered', tracking.state)
  241. self.assertTrue(self.recipient.email_score > 50.0)
  242. def test_process_deferral(self):
  243. mail, tracking = self.mail_send(self.recipient.email)
  244. tracking.event_create('deferral', {})
  245. self.assertEqual('deferred', tracking.state)
  246. def test_process_spam(self):
  247. mail, tracking = self.mail_send(self.recipient.email)
  248. tracking.event_create('spam', {})
  249. self.assertEqual('spam', tracking.state)
  250. self.assertTrue(self.recipient.email_score < 50.0)
  251. def test_process_unsub(self):
  252. mail, tracking = self.mail_send(self.recipient.email)
  253. tracking.event_create('unsub', {})
  254. self.assertEqual('unsub', tracking.state)
  255. self.assertTrue(self.recipient.email_score < 50.0)
  256. def test_process_reject(self):
  257. mail, tracking = self.mail_send(self.recipient.email)
  258. tracking.event_create('reject', {})
  259. self.assertEqual('rejected', tracking.state)
  260. self.assertTrue(self.recipient.email_score < 50.0)
  261. def test_process_open(self):
  262. mail, tracking = self.mail_send(self.recipient.email)
  263. tracking.event_create('open', {})
  264. self.assertEqual('opened', tracking.state)
  265. self.assertTrue(self.recipient.email_score > 50.0)
  266. def test_process_click(self):
  267. mail, tracking = self.mail_send(self.recipient.email)
  268. tracking.event_create('click', {})
  269. self.assertEqual('opened', tracking.state)
  270. self.assertTrue(self.recipient.email_score > 50.0)
  271. def test_process_several_bounce(self):
  272. for i in range(1, 10):
  273. mail, tracking = self.mail_send(self.recipient.email)
  274. tracking.event_create('hard_bounce', {})
  275. self.assertEqual('bounced', tracking.state)
  276. self.assertEqual(0.0, self.recipient.email_score)
  277. def test_recordset_email_score(self):
  278. """For backwords compatibility sake"""
  279. trackings = self.env['mail.tracking.email']
  280. for i in range(11):
  281. mail, tracking = self.mail_send(self.recipient.email)
  282. tracking.event_create('click', {})
  283. trackings |= tracking
  284. self.assertEqual(100.0, trackings.email_score())
  285. def test_db(self):
  286. db = self.env.cr.dbname
  287. controller = MailTrackingController()
  288. not_found = controller.mail_tracking_all('not_found_db')
  289. self.assertEqual(b'NOT FOUND', not_found.response[0])
  290. none = controller.mail_tracking_all(db)
  291. self.assertEqual(b'NONE', none.response[0])
  292. none = controller.mail_tracking_event(db, 'open')
  293. self.assertEqual(b'NONE', none.response[0])