431 lines
14 KiB
Python
431 lines
14 KiB
Python
# This file is part of Tryton. The COPYRIGHT file at the top level of
|
|
# this repository contains the full copyright notices and license terms.
|
|
|
|
import json
|
|
import uuid
|
|
from email.message import EmailMessage
|
|
from unittest.mock import patch
|
|
|
|
import trytond.config as config
|
|
from trytond.modules.inbound_email.ir import (
|
|
REPLY_LINE, _get_channel_identifier)
|
|
from trytond.pool import Pool
|
|
from trytond.protocols.wrappers import HTTPStatus
|
|
from trytond.tests.test_tryton import (
|
|
ModuleTestCase, RouteTestCase, with_transaction)
|
|
from trytond.transaction import Transaction
|
|
|
|
|
|
class InboundEmailTestCase(ModuleTestCase):
|
|
"Test Inbound Email module"
|
|
module = 'inbound_email'
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
email_from = config.get('email', 'from', default='')
|
|
config.set('email', 'from', 'tryton@example.org')
|
|
cls.addClassCleanup(config.set, 'email', 'from', email_from)
|
|
config.add_section('inbound_email')
|
|
config.set('inbound_email', 'chat_reply_to', 'chat@example.org')
|
|
cls.addClassCleanup(config.remove_section, 'inbound_email')
|
|
|
|
def get_message(self):
|
|
message = EmailMessage()
|
|
message['From'] = "John Doe <john.doe@example.com>"
|
|
message['To'] = (
|
|
"Michael Scott <michael@example.com>, pam@example.com")
|
|
message['Cc'] = 'jim@example.com'
|
|
message['Subject'] = "The office"
|
|
|
|
message.set_content("Hello")
|
|
message.add_attachment(
|
|
b'bin', maintype='application', subtype='octet-stream',
|
|
filename='data.bin')
|
|
return message
|
|
|
|
def get_message_dict(self, headers):
|
|
return {
|
|
'from': 'john.doe@example.com',
|
|
'to': ['michael@example.com', 'pam@example.com'],
|
|
'cc': ['jim@example.com'],
|
|
'subject': 'The office',
|
|
'text': 'Hello\n',
|
|
'attachments': [{
|
|
'filename': 'data.bin',
|
|
'type': 'application/octet-stream',
|
|
'data': b'bin',
|
|
}],
|
|
'headers': dict(headers.items()),
|
|
}
|
|
|
|
@with_transaction()
|
|
def test_inbox_identifier(self):
|
|
"Test inbox identifier"
|
|
pool = Pool()
|
|
Inbox = pool.get('inbound.email.inbox')
|
|
|
|
inbox = Inbox(name="Test")
|
|
inbox.save()
|
|
|
|
self.assertFalse(inbox.identifier)
|
|
self.assertFalse(inbox.endpoint)
|
|
|
|
Inbox.new_identifier([inbox])
|
|
|
|
self.assertTrue(inbox.identifier)
|
|
self.assertTrue(inbox.endpoint)
|
|
|
|
Inbox.new_identifier([inbox])
|
|
|
|
self.assertFalse(inbox.identifier)
|
|
self.assertFalse(inbox.endpoint)
|
|
|
|
@with_transaction()
|
|
def test_email_from_webhook(self):
|
|
"Test email from webhook"
|
|
pool = Pool()
|
|
Email = pool.get('inbound.email')
|
|
Inbox = pool.get('inbound.email.inbox')
|
|
|
|
message = self.get_message()
|
|
inbox = Inbox()
|
|
data = message.as_bytes()
|
|
|
|
for data_type in ['raw', 'mailpace', 'postmark', 'sendgrid']:
|
|
with self.subTest(data_type=data_type):
|
|
emails = Email.from_webhook(inbox, data, data_type=data_type)
|
|
email, = emails
|
|
self.assertEqual(email.inbox, inbox)
|
|
self.assertEqual(email.data, data)
|
|
self.assertEqual(email.data_type, data_type)
|
|
|
|
with self.subTest(data_type='mailchimp'):
|
|
data = json.dumps({
|
|
'mandrill_events': [{
|
|
'event': 'inbound',
|
|
}, {
|
|
'event': 'inbound',
|
|
}],
|
|
})
|
|
emails = Email.from_webhook(inbox, data, data_type='mailchimp')
|
|
self.assertEqual(len(emails), 2)
|
|
email = emails[0]
|
|
self.assertEqual(email.inbox, inbox)
|
|
self.assertEqual(json.loads(email.data), {'event': 'inbound'})
|
|
self.assertEqual(email.data_type, 'mailchimp')
|
|
|
|
@with_transaction()
|
|
def test_email_as_dict_raw(self):
|
|
"Test raw email as dict"
|
|
pool = Pool()
|
|
Email = pool.get('inbound.email')
|
|
|
|
message = self.get_message()
|
|
email = Email(data=message.as_bytes(), data_type='raw')
|
|
|
|
self.assertDictEqual(email.as_dict(), self.get_message_dict(message))
|
|
|
|
@with_transaction()
|
|
def test_email_as_dict_mailchimp(self):
|
|
"Test mailchimp email as dict"
|
|
pool = Pool()
|
|
Email = pool.get('inbound.email')
|
|
|
|
message = self.get_message()
|
|
data = json.dumps({
|
|
'event': 'inbound',
|
|
'raw_msg': message.as_string(),
|
|
}).encode('utf-8')
|
|
email = Email(data=data, data_type='mailchimp')
|
|
|
|
self.assertDictEqual(email.as_dict(), self.get_message_dict(message))
|
|
|
|
@with_transaction()
|
|
def test_email_as_dict_mailpace(self):
|
|
"Test mailpace email as dict"
|
|
pool = Pool()
|
|
Email = pool.get('inbound.email')
|
|
|
|
message = self.get_message()
|
|
data = json.dumps({
|
|
'raw': message.as_string(),
|
|
}).encode('utf-8')
|
|
email = Email(data=data, data_type='mailpace')
|
|
|
|
self.assertDictEqual(email.as_dict(), self.get_message_dict(message))
|
|
|
|
@with_transaction()
|
|
def test_email_as_dict_postmark(self):
|
|
"Test postmark email as dict"
|
|
pool = Pool()
|
|
Email = pool.get('inbound.email')
|
|
|
|
data = json.dumps({
|
|
'FromFull': {
|
|
'Name': 'John Doe',
|
|
'Email': 'john.doe@example.com',
|
|
},
|
|
'ToFull': [{
|
|
'Name': 'Michael',
|
|
'Email': 'michael@example.com',
|
|
}, {
|
|
'Email': 'pam@example.com',
|
|
},
|
|
],
|
|
'CcFull': [{
|
|
'Email': 'jim@example.com',
|
|
}],
|
|
'Subject': 'The office',
|
|
'TextBody': 'Hello\n',
|
|
'Attachments': [{
|
|
'Name': 'data.bin',
|
|
'Content': 'Ymlu',
|
|
'ContentType': 'application/octet-stream',
|
|
}],
|
|
'Headers': [{
|
|
'Name': 'Message-ID',
|
|
'Value': '12345@example.com',
|
|
}],
|
|
}).encode('utf-8')
|
|
email = Email(data=data, data_type='postmark')
|
|
|
|
self.assertDictEqual(
|
|
email.as_dict(),
|
|
self.get_message_dict({'Message-ID': '12345@example.com'}))
|
|
|
|
@with_transaction()
|
|
def test_email_as_dict_sendgrid(self):
|
|
"Test sendgrid email as dict"
|
|
pool = Pool()
|
|
Email = pool.get('inbound.email')
|
|
|
|
message = self.get_message()
|
|
data = json.dumps({
|
|
'email': message.as_string(),
|
|
}).encode('utf-8')
|
|
email = Email(data=data, data_type='sendgrid')
|
|
|
|
self.assertDictEqual(email.as_dict(), self.get_message_dict(message))
|
|
|
|
@with_transaction()
|
|
def test_email_process(self):
|
|
"Test email process"
|
|
pool = Pool()
|
|
Inbox = pool.get('inbound.email.inbox')
|
|
Rule = pool.get('inbound.email.rule')
|
|
Email = pool.get('inbound.email')
|
|
|
|
inbox = Inbox(name="Test")
|
|
inbox.rules = [Rule()]
|
|
inbox.save()
|
|
|
|
message = self.get_message()
|
|
email = Email(inbox=inbox, data=message.as_bytes(), data_type='raw')
|
|
email.save()
|
|
|
|
with patch.object(Rule, 'run') as run:
|
|
Email.process([email])
|
|
|
|
run.assert_called_once_with(email)
|
|
|
|
def _get_rule(self, **values):
|
|
pool = Pool()
|
|
Rule = pool.get('inbound.email.rule')
|
|
for name in Rule._fields:
|
|
values.setdefault(name, None)
|
|
return Rule(**values)
|
|
|
|
@with_transaction()
|
|
def test_rule_match_origin(self):
|
|
"Test rule match origin"
|
|
rule = self._get_rule(origin="foo.*@example.com")
|
|
|
|
self.assertTrue(rule.match({'from': "foo@example.com"}))
|
|
self.assertFalse(rule.match({'from': "bar@example.com"}))
|
|
self.assertFalse(rule.match({}))
|
|
|
|
@with_transaction()
|
|
def test_rule_match_destination(self):
|
|
"Test rule match destination"
|
|
rule = self._get_rule(destination="foo.*@example.com")
|
|
|
|
self.assertTrue(rule.match({'to': ["foo@example.com"]}))
|
|
self.assertTrue(rule.match({'to': [
|
|
"bar@example.com", "foo@example.com"]}))
|
|
self.assertTrue(rule.match({'cc': ["foo@example.com"]}))
|
|
self.assertTrue(rule.match({'bcc': ["foo@example.com"]}))
|
|
self.assertFalse(rule.match({'to': ["bar@example.com"]}))
|
|
self.assertFalse(rule.match({}))
|
|
|
|
@with_transaction()
|
|
def test_rule_match_subject(self):
|
|
"Test rule match subject"
|
|
rule = self._get_rule(subject="Test")
|
|
|
|
self.assertTrue(rule.match({'subject': "Email test subject"}))
|
|
self.assertFalse(rule.match({'subject': "Email subject"}))
|
|
self.assertFalse(rule.match({}))
|
|
|
|
@with_transaction()
|
|
def test_rule_match_attachment_name(self):
|
|
"Test rule match attachment name"
|
|
rule = self._get_rule(attachment_name="foo")
|
|
|
|
self.assertTrue(rule.match({'attachments': [
|
|
{'filename': "foo.pdf"}]}))
|
|
self.assertTrue(rule.match({'attachments': [
|
|
{'filename': "bar.pdf"}, {'filename': "foo.pdf"}]}))
|
|
self.assertFalse(rule.match({'attachments': [
|
|
{'filename': "bar.pdf"}]}))
|
|
self.assertFalse(rule.match({}))
|
|
|
|
@with_transaction()
|
|
def test_rule_match_headers(self):
|
|
"Test rule match headers"
|
|
rule = self._get_rule(
|
|
headers=[{'name': 'Message-ID', 'value': 'tryton-'}])
|
|
|
|
self.assertTrue(rule.match({'headers': {'Message-ID': 'tryton-42'}}))
|
|
self.assertTrue(rule.match({'headers': {
|
|
'From': 'foo@example.com',
|
|
'Message-ID': 'tryton-42',
|
|
}}))
|
|
self.assertFalse(rule.match({'headers': {'Message-ID': 'bar-42'}}))
|
|
self.assertFalse(rule.match({'headers': {
|
|
'From': 'tryton-test@example.com',
|
|
'Message-ID': 'bar-42',
|
|
}}))
|
|
self.assertFalse(rule.match({}))
|
|
|
|
@with_transaction()
|
|
def test_rule_run(self):
|
|
"Test run rule"
|
|
pool = Pool()
|
|
Rule = pool.get('inbound.email.rule')
|
|
Email = pool.get('inbound.email')
|
|
|
|
rule = Rule(action='inbound.email|test')
|
|
email = Email()
|
|
|
|
with patch.object(Email, 'test', create=True) as func:
|
|
func.return_value = result = Rule()
|
|
|
|
rule.run(email)
|
|
|
|
func.assert_called_once_with(email, rule)
|
|
self.assertEqual(email.result, result)
|
|
|
|
def test_get_channel_identifier(self):
|
|
"Test parsing the recipients to find the email channel"
|
|
channel_id = uuid.uuid4().hex
|
|
|
|
for value, expected in [
|
|
(f'chat+{channel_id}@example.org', channel_id),
|
|
(['chat@example.org', f'chat+{channel_id}@example.org'],
|
|
channel_id),
|
|
('chat@example.org', None),
|
|
('chat+foo@example.org', None),
|
|
(f'{channel_id}@example.org', None),
|
|
(f'chat+{channel_id}@example.com', None),
|
|
]:
|
|
for recipient_hdr in ('To', 'Cc'):
|
|
with self.subTest(f"{recipient_hdr}: {value}"):
|
|
msg = EmailMessage()
|
|
msg[recipient_hdr] = value
|
|
self.assertEqual(_get_channel_identifier(msg), expected)
|
|
|
|
@with_transaction()
|
|
def test_find_email_content(self):
|
|
"Test finding the content of the reply"
|
|
pool = Pool()
|
|
Channel = pool.get('ir.chat.channel')
|
|
|
|
for (body, expected) in [
|
|
("Foo", "Foo"),
|
|
("Foo\n-----\nBar", "Foo\n-----\nBar"),
|
|
(f"Foo\n{REPLY_LINE}\nLorem Ipsum", "Foo"),
|
|
(f"Foo\n> {REPLY_LINE}\n> Lorem Ipsum", "Foo"),
|
|
]:
|
|
with self.subTest(body):
|
|
self.assertEqual(Channel._email_content(body), expected)
|
|
|
|
|
|
class InboundEmailRouteTestCase(RouteTestCase):
|
|
"Test Inbound Email route"
|
|
module = 'inbound_email'
|
|
|
|
identifier = uuid.uuid4().hex
|
|
|
|
@classmethod
|
|
def setUpDatabase(cls):
|
|
pool = Pool()
|
|
Inbox = pool.get('inbound.email.inbox')
|
|
Inbox(name="Test", identifier=cls.identifier).save()
|
|
|
|
def test_inbound_email_route_data(self):
|
|
"Test inbound email route with data"
|
|
|
|
client = self.client()
|
|
|
|
response = client.post(
|
|
f'/{self.db_name}/inbound_email/inbox/{self.identifier}',
|
|
data=b'data')
|
|
|
|
self.assertEqual(response.status_code, HTTPStatus.NO_CONTENT)
|
|
|
|
@with_transaction()
|
|
def check():
|
|
pool = Pool()
|
|
Email = pool.get('inbound.email')
|
|
email, = Email.search([])
|
|
|
|
try:
|
|
self.assertEqual(email.data, b'data')
|
|
self.assertEqual(email.data_type, 'raw')
|
|
self.assertTrue(email.inbox)
|
|
finally:
|
|
Email.delete([email])
|
|
Transaction().commit()
|
|
check()
|
|
|
|
def test_inbound_email_route_form(self):
|
|
"Test inbound email route with form"
|
|
client = self.client()
|
|
|
|
response = client.post(
|
|
f'/{self.db_name}/inbound_email/inbox/{self.identifier}',
|
|
data={
|
|
'email': 'data',
|
|
})
|
|
|
|
self.assertEqual(response.status_code, HTTPStatus.NO_CONTENT)
|
|
|
|
@with_transaction()
|
|
def check():
|
|
pool = Pool()
|
|
Email = pool.get('inbound.email')
|
|
email, = Email.search([])
|
|
|
|
try:
|
|
self.assertEqual(email.data, b'{"email": "data"}')
|
|
self.assertEqual(email.data_type, 'raw')
|
|
self.assertTrue(email.inbox)
|
|
finally:
|
|
Email.delete([email])
|
|
Transaction().commit()
|
|
check()
|
|
|
|
def test_inbound_email_route_not_found(self):
|
|
"Test inbound email route not found"
|
|
client = self.client()
|
|
|
|
response = client.post(
|
|
f'/{self.db_name}/inbound_email/inbox/unknown',
|
|
data=b'foo')
|
|
|
|
self.assertEqual(response.status_code, HTTPStatus.NOT_FOUND)
|
|
|
|
|
|
del ModuleTestCase, RouteTestCase
|