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.

411 lines
17 KiB

  1. # -*- coding: utf-8 -*-
  2. # Copyright 2016 Tecnativa - Antonio Espinosa
  3. # Copyright 2017 Tecnativa - David Vidal
  4. # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
  5. from openerp.tests.common import TransactionCase
  6. from openerp.exceptions import UserError, ValidationError
  7. import mock
  8. import json
  9. _packagepath = 'openerp.addons.mail_tracking_mailgun'
  10. class TestMailgun(TransactionCase):
  11. def mail_send(self):
  12. mail = self.env['mail.mail'].create({
  13. 'subject': 'Test subject',
  14. 'email_from': 'from@example.com',
  15. 'email_to': self.recipient,
  16. 'body_html': '<p>This is a test message</p>',
  17. })
  18. mail.send()
  19. # Search tracking created
  20. tracking_email = self.env['mail.tracking.email'].search([
  21. ('mail_id', '=', mail.id),
  22. ])
  23. return mail, tracking_email
  24. def setUp(self):
  25. super(TestMailgun, self).setUp()
  26. self.recipient = u'to@example.com'
  27. self.mail, self.tracking_email = self.mail_send()
  28. self.api_key = u'key-12345678901234567890123456789012'
  29. self.domain = u'example.com'
  30. self.token = u'f1349299097a51b9a7d886fcb5c2735b426ba200ada6e9e149'
  31. self.timestamp = u'1471021089'
  32. self.signature = ('4fb6d4dbbe10ce5d620265dcd7a3c0b8'
  33. 'ca0dede1433103891bc1ae4086e9d5b2')
  34. self.env['ir.config_parameter'].set_param(
  35. 'mailgun.apikey', self.api_key)
  36. self.env['ir.config_parameter'].set_param(
  37. 'mail.catchall.domain', self.domain)
  38. self.env['ir.config_parameter'].set_param(
  39. 'mailgun.validation_key', self.api_key)
  40. self.event = {
  41. 'Message-Id': u'<xxx.xxx.xxx-openerp-xxx-res.partner@test_db>',
  42. 'X-Mailgun-Sid': u'WyIwNjgxZSIsICJ0b0BleGFtcGxlLmNvbSIsICI3MG'
  43. 'I0MWYiXQ==',
  44. 'token': self.token,
  45. 'timestamp': self.timestamp,
  46. 'signature': self.signature,
  47. 'domain': u'example.com',
  48. 'message-headers': u'[]',
  49. 'recipient': self.recipient,
  50. 'odoo_db': self.env.cr.dbname,
  51. 'tracking_email_id': u'%s' % self.tracking_email.id
  52. }
  53. self.metadata = {
  54. 'ip': '127.0.0.1',
  55. 'user_agent': False,
  56. 'os_family': False,
  57. 'ua_family': False,
  58. }
  59. self.partner = self.env['res.partner'].create({
  60. 'name': 'Mr. Odoo',
  61. 'email': 'mrodoo@example.com',
  62. })
  63. self.response = {
  64. "items": [{
  65. "log-level": "info",
  66. "id": "oXAVv5URCF-dKv8c6Sa7T",
  67. "timestamp": 1509119329.0,
  68. "message": {
  69. "headers": {
  70. "to": "test@test.com",
  71. "message-id": "test-id@f187c54734e8",
  72. "from": "Mr. Odoo <mrodoo@odoo.com>",
  73. "subject": "This is a test"
  74. },
  75. },
  76. "event": "delivered",
  77. "recipient": "to@example.com",
  78. }]
  79. }
  80. def event_search(self, event_type):
  81. event = self.env['mail.tracking.event'].search([
  82. ('tracking_email_id', '=', self.tracking_email.id),
  83. ('event_type', '=', event_type),
  84. ])
  85. self.assertTrue(event)
  86. return event
  87. def test_no_api_key(self):
  88. self.env['ir.config_parameter'].set_param('mailgun.apikey', '')
  89. self.test_event_delivered()
  90. with self.assertRaises(ValidationError):
  91. self.env['mail.tracking.email']._mailgun_values()
  92. def test_no_domain(self):
  93. self.env['ir.config_parameter'].set_param('mail.catchall.domain', '')
  94. self.test_event_delivered()
  95. with self.assertRaises(ValidationError):
  96. self.env['mail.tracking.email']._mailgun_values()
  97. def test_bad_signature(self):
  98. self.event.update({
  99. 'event': u'delivered',
  100. 'signature': u'bad_signature',
  101. })
  102. response = self.env['mail.tracking.email'].event_process(
  103. None, self.event, self.metadata)
  104. self.assertEqual('ERROR: Signature', response)
  105. def test_bad_event_type(self):
  106. self.event.update({
  107. 'event': u'bad_event',
  108. })
  109. response = self.env['mail.tracking.email'].event_process(
  110. None, self.event, self.metadata)
  111. self.assertEqual('ERROR: Event type not supported', response)
  112. def test_bad_db(self):
  113. self.event.update({
  114. 'event': u'delivered',
  115. 'odoo_db': u'bad_db',
  116. })
  117. response = self.env['mail.tracking.email'].event_process(
  118. None, self.event, self.metadata)
  119. self.assertEqual('ERROR: Invalid DB', response)
  120. def test_bad_ts(self):
  121. timestamp = u'7a' # Now time will be used instead
  122. signature = ('06cc05680f6e8110e59b41152b2d1c0f'
  123. '1045d755ef2880ff922344325c89a6d4')
  124. self.event.update({
  125. 'event': u'delivered',
  126. 'timestamp': timestamp,
  127. 'signature': signature,
  128. })
  129. response = self.env['mail.tracking.email'].event_process(
  130. None, self.event, self.metadata)
  131. self.assertEqual('OK', response)
  132. def test_tracking_not_found(self):
  133. self.event.update({
  134. 'event': u'delivered',
  135. 'tracking_email_id': u'bad_id',
  136. })
  137. response = self.env['mail.tracking.email'].event_process(
  138. None, self.event, self.metadata)
  139. self.assertEqual('ERROR: Tracking not found', response)
  140. # https://documentation.mailgun.com/user_manual.html#tracking-deliveries
  141. def test_event_delivered(self):
  142. self.event.update({
  143. 'event': u'delivered',
  144. })
  145. response = self.env['mail.tracking.email'].event_process(
  146. None, self.event, self.metadata)
  147. self.assertEqual('OK', response)
  148. event = self.event_search('delivered')
  149. self.assertEqual(event.timestamp, float(self.timestamp))
  150. self.assertEqual(event.recipient, self.recipient)
  151. # https://documentation.mailgun.com/user_manual.html#tracking-opens
  152. def test_event_opened(self):
  153. ip = u'127.0.0.1'
  154. user_agent = u'Odoo Test/8.0 Gecko Firefox/11.0'
  155. os_family = u'Linux'
  156. ua_family = u'Firefox'
  157. ua_type = u'browser'
  158. self.event.update({
  159. 'event': u'opened',
  160. 'city': u'Mountain View',
  161. 'country': u'US',
  162. 'region': u'CA',
  163. 'client-name': ua_family,
  164. 'client-os': os_family,
  165. 'client-type': ua_type,
  166. 'device-type': u'desktop',
  167. 'ip': ip,
  168. 'user-agent': user_agent,
  169. })
  170. response = self.env['mail.tracking.email'].event_process(
  171. None, self.event, self.metadata)
  172. self.assertEqual('OK', response)
  173. event = self.event_search('open')
  174. self.assertEqual(event.timestamp, float(self.timestamp))
  175. self.assertEqual(event.recipient, self.recipient)
  176. self.assertEqual(event.ip, ip)
  177. self.assertEqual(event.user_agent, user_agent)
  178. self.assertEqual(event.os_family, os_family)
  179. self.assertEqual(event.ua_family, ua_family)
  180. self.assertEqual(event.ua_type, ua_type)
  181. self.assertEqual(event.mobile, False)
  182. self.assertEqual(event.user_country_id.code, 'US')
  183. # https://documentation.mailgun.com/user_manual.html#tracking-clicks
  184. def test_event_clicked(self):
  185. ip = u'127.0.0.1'
  186. user_agent = u'Odoo Test/8.0 Gecko Firefox/11.0'
  187. os_family = u'Linux'
  188. ua_family = u'Firefox'
  189. ua_type = u'browser'
  190. url = u'https://odoo-community.org'
  191. self.event.update({
  192. 'event': u'clicked',
  193. 'city': u'Mountain View',
  194. 'country': u'US',
  195. 'region': u'CA',
  196. 'client-name': ua_family,
  197. 'client-os': os_family,
  198. 'client-type': ua_type,
  199. 'device-type': u'tablet',
  200. 'ip': ip,
  201. 'user-agent': user_agent,
  202. 'url': url,
  203. })
  204. response = self.env['mail.tracking.email'].event_process(
  205. None, self.event, self.metadata, event_type='click')
  206. self.assertEqual('OK', response)
  207. event = self.event_search('click')
  208. self.assertEqual(event.timestamp, float(self.timestamp))
  209. self.assertEqual(event.recipient, self.recipient)
  210. self.assertEqual(event.ip, ip)
  211. self.assertEqual(event.user_agent, user_agent)
  212. self.assertEqual(event.os_family, os_family)
  213. self.assertEqual(event.ua_family, ua_family)
  214. self.assertEqual(event.ua_type, ua_type)
  215. self.assertEqual(event.mobile, True)
  216. self.assertEqual(event.url, url)
  217. # https://documentation.mailgun.com/user_manual.html#tracking-unsubscribes
  218. def test_event_unsubscribed(self):
  219. ip = u'127.0.0.1'
  220. user_agent = u'Odoo Test/8.0 Gecko Firefox/11.0'
  221. os_family = u'Linux'
  222. ua_family = u'Firefox'
  223. ua_type = u'browser'
  224. self.event.update({
  225. 'event': u'unsubscribed',
  226. 'city': u'Mountain View',
  227. 'country': u'US',
  228. 'region': u'CA',
  229. 'client-name': ua_family,
  230. 'client-os': os_family,
  231. 'client-type': ua_type,
  232. 'device-type': u'mobile',
  233. 'ip': ip,
  234. 'user-agent': user_agent,
  235. })
  236. response = self.env['mail.tracking.email'].event_process(
  237. None, self.event, self.metadata)
  238. self.assertEqual('OK', response)
  239. event = self.event_search('unsub')
  240. self.assertEqual(event.timestamp, float(self.timestamp))
  241. self.assertEqual(event.recipient, self.recipient)
  242. self.assertEqual(event.ip, ip)
  243. self.assertEqual(event.user_agent, user_agent)
  244. self.assertEqual(event.os_family, os_family)
  245. self.assertEqual(event.ua_family, ua_family)
  246. self.assertEqual(event.ua_type, ua_type)
  247. self.assertEqual(event.mobile, True)
  248. # https://documentation.mailgun.com/
  249. # user_manual.html#tracking-spam-complaints
  250. def test_event_complained(self):
  251. self.event.update({
  252. 'event': u'complained',
  253. })
  254. response = self.env['mail.tracking.email'].event_process(
  255. None, self.event, self.metadata)
  256. self.assertEqual('OK', response)
  257. event = self.event_search('spam')
  258. self.assertEqual(event.timestamp, float(self.timestamp))
  259. self.assertEqual(event.recipient, self.recipient)
  260. self.assertEqual(event.error_type, 'spam')
  261. # https://documentation.mailgun.com/user_manual.html#tracking-bounces
  262. def test_event_bounced(self):
  263. code = u'550'
  264. error = (u"5.1.1 The email account does not exist.\n"
  265. "5.1.1 double-checking the recipient's email address")
  266. notification = u"Please, check recipient's email address"
  267. self.event.update({
  268. 'event': u'bounced',
  269. 'code': code,
  270. 'error': error,
  271. 'notification': notification,
  272. })
  273. response = self.env['mail.tracking.email'].event_process(
  274. None, self.event, self.metadata)
  275. self.assertEqual('OK', response)
  276. event = self.event_search('hard_bounce')
  277. self.assertEqual(event.timestamp, float(self.timestamp))
  278. self.assertEqual(event.recipient, self.recipient)
  279. self.assertEqual(event.error_type, code)
  280. self.assertEqual(event.error_description, error)
  281. self.assertEqual(event.error_details, notification)
  282. # https://documentation.mailgun.com/user_manual.html#tracking-failures
  283. def test_event_dropped(self):
  284. reason = u'hardfail'
  285. code = u'605'
  286. description = u'Not delivering to previously bounced address'
  287. self.event.update({
  288. 'event': u'dropped',
  289. 'reason': reason,
  290. 'code': code,
  291. 'description': description,
  292. })
  293. response = self.env['mail.tracking.email'].event_process(
  294. None, self.event, self.metadata)
  295. self.assertEqual('OK', response)
  296. event = self.event_search('reject')
  297. self.assertEqual(event.timestamp, float(self.timestamp))
  298. self.assertEqual(event.recipient, self.recipient)
  299. self.assertEqual(event.error_type, reason)
  300. self.assertEqual(event.error_description, code)
  301. self.assertEqual(event.error_details, description)
  302. @mock.patch(_packagepath + '.models.res_partner.requests')
  303. def test_email_validity(self, mock_request):
  304. self.partner.email_bounced = False
  305. self.partner.email = 'info@tecnativa.com'
  306. mock_request.get.return_value.apparent_encoding = 'ascii'
  307. mock_request.get.return_value.status_code = 200
  308. mock_request.get.return_value.content = json.dumps({
  309. 'is_valid': True,
  310. 'mailbox_verification': 'true',
  311. }, ensure_ascii=True)
  312. self.partner.check_email_validity()
  313. self.assertFalse(self.partner.email_bounced)
  314. self.partner.email = 'xoxoxoxo@tecnativa.com'
  315. # Not a valid mailbox
  316. mock_request.get.return_value.content = json.dumps({
  317. 'is_valid': True,
  318. 'mailbox_verification': 'false',
  319. }, ensure_ascii=True)
  320. with self.assertRaises(UserError):
  321. self.partner.check_email_validity()
  322. # Not a valid mail address
  323. mock_request.get.return_value.content = json.dumps({
  324. 'is_valid': False,
  325. 'mailbox_verification': 'false',
  326. }, ensure_ascii=True)
  327. with self.assertRaises(UserError):
  328. self.partner.check_email_validity()
  329. # Unable to fully validate
  330. mock_request.get.return_value.content = json.dumps({
  331. 'is_valid': True,
  332. 'mailbox_verification': 'unknown',
  333. }, ensure_ascii=True)
  334. with self.assertRaises(UserError):
  335. self.partner.check_email_validity()
  336. self.assertTrue(self.partner.email_bounced)
  337. @mock.patch(_packagepath + '.models.res_partner.requests')
  338. def test_email_validity_exceptions(self, mock_request):
  339. mock_request.get.return_value.status_code = 404
  340. with self.assertRaises(UserError):
  341. self.partner.check_email_validity()
  342. self.env['ir.config_parameter'].set_param('mailgun.validation_key', '')
  343. with self.assertRaises(UserError):
  344. self.partner.check_email_validity()
  345. @mock.patch(_packagepath + '.models.res_partner.requests')
  346. def test_bounced(self, mock_request):
  347. self.partner.email_bounced = True
  348. mock_request.get.return_value.status_code = 404
  349. self.partner.check_email_bounced()
  350. self.assertFalse(self.partner.email_bounced)
  351. mock_request.get.return_value.status_code = 200
  352. self.partner.force_set_bounced()
  353. self.partner.check_email_bounced()
  354. self.assertTrue(self.partner.email_bounced)
  355. mock_request.delete.return_value.status_code = 200
  356. self.partner.force_unset_bounced()
  357. self.assertFalse(self.partner.email_bounced)
  358. def test_email_bounced_set(self):
  359. message_number = len(self.partner.message_ids) + 1
  360. self.partner._email_bounced_set('test_error', self.event)
  361. self.assertEqual(len(self.partner.message_ids), message_number)
  362. self.partner.email = ""
  363. self.partner._email_bounced_set('test_error', self.event)
  364. self.assertEqual(len(self.partner.message_ids), message_number)
  365. @mock.patch(_packagepath + '.models.mail_tracking_email.requests')
  366. def test_manual_check(self, mock_request):
  367. mock_request.get.return_value.content = json.dumps(self.response,
  368. ensure_ascii=True)
  369. mock_request.get.return_value.apparent_encoding = 'ascii'
  370. mock_request.get.return_value.status_code = 200
  371. self.tracking_email.action_manual_check_mailgun()
  372. event = self.env['mail.tracking.event'].search(
  373. [('mailgun_id', '=', self.response['items'][0]['id'])])
  374. self.assertEqual(event.event_type, self.response['items'][0]['event'])
  375. @mock.patch(_packagepath + '.models.mail_tracking_email.requests')
  376. def test_manual_check_exceptions(self, mock_request):
  377. mock_request.get.return_value.status_code = 404
  378. with self.assertRaises(ValidationError):
  379. self.tracking_email.action_manual_check_mailgun()
  380. mock_request.get.return_value.status_code = 200
  381. mock_request.get.return_value.content = json.dumps('{}',
  382. ensure_ascii=True)
  383. mock_request.get.return_value.apparent_encoding = 'ascii'
  384. with self.assertRaises(ValidationError):
  385. self.tracking_email.action_manual_check_mailgun()