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.
 
 
 
 

410 lines
17 KiB

# -*- coding: utf-8 -*-
# Copyright 2016 Tecnativa - Antonio Espinosa
# Copyright 2017 Tecnativa - David Vidal
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from openerp.tests.common import TransactionCase
from openerp.exceptions import UserError, ValidationError
import mock
import json
_packagepath = 'openerp.addons.mail_tracking_mailgun'
class TestMailgun(TransactionCase):
def mail_send(self):
mail = self.env['mail.mail'].create({
'subject': 'Test subject',
'email_from': 'from@example.com',
'email_to': self.recipient,
'body_html': '<p>This is a test message</p>',
})
mail.send()
# Search tracking created
tracking_email = self.env['mail.tracking.email'].search([
('mail_id', '=', mail.id),
])
return mail, tracking_email
def setUp(self):
super(TestMailgun, self).setUp()
self.recipient = u'to@example.com'
self.mail, self.tracking_email = self.mail_send()
self.api_key = u'key-12345678901234567890123456789012'
self.domain = u'example.com'
self.token = u'f1349299097a51b9a7d886fcb5c2735b426ba200ada6e9e149'
self.timestamp = u'1471021089'
self.signature = ('4fb6d4dbbe10ce5d620265dcd7a3c0b8'
'ca0dede1433103891bc1ae4086e9d5b2')
self.env['ir.config_parameter'].set_param(
'mailgun.apikey', self.api_key)
self.env['ir.config_parameter'].set_param(
'mail.catchall.domain', self.domain)
self.env['ir.config_parameter'].set_param(
'mailgun.validation_key', self.api_key)
self.event = {
'Message-Id': u'<xxx.xxx.xxx-openerp-xxx-res.partner@test_db>',
'X-Mailgun-Sid': u'WyIwNjgxZSIsICJ0b0BleGFtcGxlLmNvbSIsICI3MG'
'I0MWYiXQ==',
'token': self.token,
'timestamp': self.timestamp,
'signature': self.signature,
'domain': u'example.com',
'message-headers': u'[]',
'recipient': self.recipient,
'odoo_db': self.env.cr.dbname,
'tracking_email_id': u'%s' % self.tracking_email.id
}
self.metadata = {
'ip': '127.0.0.1',
'user_agent': False,
'os_family': False,
'ua_family': False,
}
self.partner = self.env['res.partner'].create({
'name': 'Mr. Odoo',
'email': 'mrodoo@example.com',
})
self.response = {
"items": [{
"log-level": "info",
"id": "oXAVv5URCF-dKv8c6Sa7T",
"timestamp": 1509119329.0,
"message": {
"headers": {
"to": "test@test.com",
"message-id": "test-id@f187c54734e8",
"from": "Mr. Odoo <mrodoo@odoo.com>",
"subject": "This is a test"
},
},
"event": "delivered"
}]
}
def event_search(self, event_type):
event = self.env['mail.tracking.event'].search([
('tracking_email_id', '=', self.tracking_email.id),
('event_type', '=', event_type),
])
self.assertTrue(event)
return event
def test_no_api_key(self):
self.env['ir.config_parameter'].set_param('mailgun.apikey', '')
self.test_event_delivered()
with self.assertRaises(ValidationError):
self.env['mail.tracking.email']._mailgun_values()
def test_no_domain(self):
self.env['ir.config_parameter'].set_param('mail.catchall.domain', '')
self.test_event_delivered()
with self.assertRaises(ValidationError):
self.env['mail.tracking.email']._mailgun_values()
def test_bad_signature(self):
self.event.update({
'event': u'delivered',
'signature': u'bad_signature',
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('ERROR: Signature', response)
def test_bad_event_type(self):
self.event.update({
'event': u'bad_event',
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('ERROR: Event type not supported', response)
def test_bad_db(self):
self.event.update({
'event': u'delivered',
'odoo_db': u'bad_db',
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('ERROR: Invalid DB', response)
def test_bad_ts(self):
timestamp = u'7a' # Now time will be used instead
signature = ('06cc05680f6e8110e59b41152b2d1c0f'
'1045d755ef2880ff922344325c89a6d4')
self.event.update({
'event': u'delivered',
'timestamp': timestamp,
'signature': signature,
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('OK', response)
def test_tracking_not_found(self):
self.event.update({
'event': u'delivered',
'tracking_email_id': u'bad_id',
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('ERROR: Tracking not found', response)
# https://documentation.mailgun.com/user_manual.html#tracking-deliveries
def test_event_delivered(self):
self.event.update({
'event': u'delivered',
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('OK', response)
event = self.event_search('delivered')
self.assertEqual(event.timestamp, float(self.timestamp))
self.assertEqual(event.recipient, self.recipient)
# https://documentation.mailgun.com/user_manual.html#tracking-opens
def test_event_opened(self):
ip = u'127.0.0.1'
user_agent = u'Odoo Test/8.0 Gecko Firefox/11.0'
os_family = u'Linux'
ua_family = u'Firefox'
ua_type = u'browser'
self.event.update({
'event': u'opened',
'city': u'Mountain View',
'country': u'US',
'region': u'CA',
'client-name': ua_family,
'client-os': os_family,
'client-type': ua_type,
'device-type': u'desktop',
'ip': ip,
'user-agent': user_agent,
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('OK', response)
event = self.event_search('open')
self.assertEqual(event.timestamp, float(self.timestamp))
self.assertEqual(event.recipient, self.recipient)
self.assertEqual(event.ip, ip)
self.assertEqual(event.user_agent, user_agent)
self.assertEqual(event.os_family, os_family)
self.assertEqual(event.ua_family, ua_family)
self.assertEqual(event.ua_type, ua_type)
self.assertEqual(event.mobile, False)
self.assertEqual(event.user_country_id.code, 'US')
# https://documentation.mailgun.com/user_manual.html#tracking-clicks
def test_event_clicked(self):
ip = u'127.0.0.1'
user_agent = u'Odoo Test/8.0 Gecko Firefox/11.0'
os_family = u'Linux'
ua_family = u'Firefox'
ua_type = u'browser'
url = u'https://odoo-community.org'
self.event.update({
'event': u'clicked',
'city': u'Mountain View',
'country': u'US',
'region': u'CA',
'client-name': ua_family,
'client-os': os_family,
'client-type': ua_type,
'device-type': u'tablet',
'ip': ip,
'user-agent': user_agent,
'url': url,
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata, event_type='click')
self.assertEqual('OK', response)
event = self.event_search('click')
self.assertEqual(event.timestamp, float(self.timestamp))
self.assertEqual(event.recipient, self.recipient)
self.assertEqual(event.ip, ip)
self.assertEqual(event.user_agent, user_agent)
self.assertEqual(event.os_family, os_family)
self.assertEqual(event.ua_family, ua_family)
self.assertEqual(event.ua_type, ua_type)
self.assertEqual(event.mobile, True)
self.assertEqual(event.url, url)
# https://documentation.mailgun.com/user_manual.html#tracking-unsubscribes
def test_event_unsubscribed(self):
ip = u'127.0.0.1'
user_agent = u'Odoo Test/8.0 Gecko Firefox/11.0'
os_family = u'Linux'
ua_family = u'Firefox'
ua_type = u'browser'
self.event.update({
'event': u'unsubscribed',
'city': u'Mountain View',
'country': u'US',
'region': u'CA',
'client-name': ua_family,
'client-os': os_family,
'client-type': ua_type,
'device-type': u'mobile',
'ip': ip,
'user-agent': user_agent,
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('OK', response)
event = self.event_search('unsub')
self.assertEqual(event.timestamp, float(self.timestamp))
self.assertEqual(event.recipient, self.recipient)
self.assertEqual(event.ip, ip)
self.assertEqual(event.user_agent, user_agent)
self.assertEqual(event.os_family, os_family)
self.assertEqual(event.ua_family, ua_family)
self.assertEqual(event.ua_type, ua_type)
self.assertEqual(event.mobile, True)
# https://documentation.mailgun.com/
# user_manual.html#tracking-spam-complaints
def test_event_complained(self):
self.event.update({
'event': u'complained',
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('OK', response)
event = self.event_search('spam')
self.assertEqual(event.timestamp, float(self.timestamp))
self.assertEqual(event.recipient, self.recipient)
self.assertEqual(event.error_type, 'spam')
# https://documentation.mailgun.com/user_manual.html#tracking-bounces
def test_event_bounced(self):
code = u'550'
error = (u"5.1.1 The email account does not exist.\n"
"5.1.1 double-checking the recipient's email address")
notification = u"Please, check recipient's email address"
self.event.update({
'event': u'bounced',
'code': code,
'error': error,
'notification': notification,
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('OK', response)
event = self.event_search('hard_bounce')
self.assertEqual(event.timestamp, float(self.timestamp))
self.assertEqual(event.recipient, self.recipient)
self.assertEqual(event.error_type, code)
self.assertEqual(event.error_description, error)
self.assertEqual(event.error_details, notification)
# https://documentation.mailgun.com/user_manual.html#tracking-failures
def test_event_dropped(self):
reason = u'hardfail'
code = u'605'
description = u'Not delivering to previously bounced address'
self.event.update({
'event': u'dropped',
'reason': reason,
'code': code,
'description': description,
})
response = self.env['mail.tracking.email'].event_process(
None, self.event, self.metadata)
self.assertEqual('OK', response)
event = self.event_search('reject')
self.assertEqual(event.timestamp, float(self.timestamp))
self.assertEqual(event.recipient, self.recipient)
self.assertEqual(event.error_type, reason)
self.assertEqual(event.error_description, code)
self.assertEqual(event.error_details, description)
@mock.patch(_packagepath + '.models.res_partner.requests')
def test_email_validity(self, mock_request):
self.partner.email_bounced = False
self.partner.email = 'info@tecnativa.com'
mock_request.get.return_value.apparent_encoding = 'ascii'
mock_request.get.return_value.status_code = 200
mock_request.get.return_value.content = json.dumps({
'is_valid': True,
'mailbox_verification': 'true',
}, ensure_ascii=True)
self.partner.check_email_validity()
self.assertFalse(self.partner.email_bounced)
self.partner.email = 'xoxoxoxo@tecnativa.com'
# Not a valid mailbox
mock_request.get.return_value.content = json.dumps({
'is_valid': True,
'mailbox_verification': 'false',
}, ensure_ascii=True)
with self.assertRaises(UserError):
self.partner.check_email_validity()
# Not a valid mail address
mock_request.get.return_value.content = json.dumps({
'is_valid': False,
'mailbox_verification': 'false',
}, ensure_ascii=True)
with self.assertRaises(UserError):
self.partner.check_email_validity()
# Unable to fully validate
mock_request.get.return_value.content = json.dumps({
'is_valid': True,
'mailbox_verification': 'unknown',
}, ensure_ascii=True)
with self.assertRaises(UserError):
self.partner.check_email_validity()
self.assertTrue(self.partner.email_bounced)
@mock.patch(_packagepath + '.models.res_partner.requests')
def test_email_validity_exceptions(self, mock_request):
mock_request.get.return_value.status_code = 404
with self.assertRaises(UserError):
self.partner.check_email_validity()
self.env['ir.config_parameter'].set_param('mailgun.validation_key', '')
with self.assertRaises(UserError):
self.partner.check_email_validity()
@mock.patch(_packagepath + '.models.res_partner.requests')
def test_bounced(self, mock_request):
self.partner.email_bounced = True
mock_request.get.return_value.status_code = 404
self.partner.check_email_bounced()
self.assertFalse(self.partner.email_bounced)
mock_request.get.return_value.status_code = 200
self.partner.force_set_bounced()
self.partner.check_email_bounced()
self.assertTrue(self.partner.email_bounced)
mock_request.delete.return_value.status_code = 200
self.partner.force_unset_bounced()
self.assertFalse(self.partner.email_bounced)
def test_email_bounced_set(self):
message_number = len(self.partner.message_ids) + 1
self.partner._email_bounced_set('test_error', self.event)
self.assertEqual(len(self.partner.message_ids), message_number)
self.partner.email = ""
self.partner._email_bounced_set('test_error', self.event)
self.assertEqual(len(self.partner.message_ids), message_number)
@mock.patch(_packagepath + '.models.mail_tracking_email.requests')
def test_manual_check(self, mock_request):
mock_request.get.return_value.content = json.dumps(self.response,
ensure_ascii=True)
mock_request.get.return_value.apparent_encoding = 'ascii'
mock_request.get.return_value.status_code = 200
self.tracking_email.action_manual_check_mailgun()
event = self.env['mail.tracking.event'].search(
[('mailgun_id', '=', self.response['items'][0]['id'])])
self.assertEqual(event.event_type, self.response['items'][0]['event'])
@mock.patch(_packagepath + '.models.mail_tracking_email.requests')
def test_manual_check_exceptions(self, mock_request):
mock_request.get.return_value.status_code = 404
with self.assertRaises(ValidationError):
self.tracking_email.action_manual_check_mailgun()
mock_request.get.return_value.status_code = 200
mock_request.get.return_value.content = json.dumps('{}',
ensure_ascii=True)
mock_request.get.return_value.apparent_encoding = 'ascii'
with self.assertRaises(ValidationError):
self.tracking_email.action_manual_check_mailgun()