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.

317 lines
12 KiB

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