summaryrefslogtreecommitdiff
path: root/service/test/unit/adapter/services/test_mail_sender.py
blob: 93703c8ed709a7d6ecc17aa5ebd75663c5cf93c8 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
#
# Copyright (c) 2014 ThoughtWorks, Inc.
#
# Pixelated is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Pixelated is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Pixelated. If not, see <http://www.gnu.org/licenses/>.
from leap.bitmask.mail.outgoing.service import OutgoingMail
from mock import patch
from twisted.mail.smtp import User
from twisted.trial import unittest

from mockito import mock, when, verify, any, unstub
from pixelated.adapter.services.mail_sender import MailSender, MailSenderException
from pixelated.adapter.model.mail import InputMail
from pixelated.bitmask_libraries.smtp import LeapSMTPConfig
from pixelated.support.functional import flatten
from test.support.test_helper import mail_dict
from twisted.internet import defer
from mockito.matchers import Matcher


class TwistedSmtpUserCapture(Matcher):

    def __init__(self, username):
        self._username = username

    def matches(self, arg):
        return isinstance(arg, User) \
            and isinstance(arg.dest.addrstr, str) \
            and self._username == arg.dest.addrstr


class MailToSmtpFormatCapture(Matcher):

    def __init__(self, recipient, bccs):
        self._recipient = recipient
        self._bccs = bccs

    def matches(self, mail):
        if self._recipient in self._bccs:
            return 'Bcc: %s\n' % self._recipient in mail
        else:
            return "Bcc: " not in mail


class MailSenderTest(unittest.TestCase):

    def setUp(self):
        self._cert_path = u'/some/cert/path'
        self._keymanager_mock = mock()
        self._remote_smtp_host = 'some.host.test'
        self._remote_smtp_port = 1234
        self._smtp_config = LeapSMTPConfig('someone@somedomain.tld', self._cert_path, self._remote_smtp_host, self._remote_smtp_port)
        self.sender = MailSender(self._smtp_config, self._keymanager_mock)

    def tearDown(self):
        unstub()

    @defer.inlineCallbacks
    def test_iterates_over_recipients(self):
        input_mail = InputMail.from_dict(mail_dict(), from_address='pixelated@org')

        when(OutgoingMail).send_message(any(), any()).thenReturn(defer.succeed(None))

        yield self.sender.sendmail(input_mail)

        for recipient in flatten([input_mail.to, input_mail.cc, input_mail.bcc]):
            verify(OutgoingMail).send_message(any(), TwistedSmtpUserCapture(recipient))

    @defer.inlineCallbacks
    def test_send_leaves_mail_in_tact(self):
        input_mail_dict = mail_dict()
        input_mail = InputMail.from_dict(input_mail_dict, from_address='pixelated@org')

        when(OutgoingMail).send_message(any(), any()).thenReturn(defer.succeed(None))

        yield self.sender.sendmail(input_mail)

        self.assertEqual(input_mail.to, input_mail_dict["header"]["to"])
        self.assertEqual(input_mail.cc, input_mail_dict["header"]["cc"])
        self.assertEqual(input_mail.bcc, input_mail_dict["header"]["bcc"])
        self.assertEqual(input_mail.subject, input_mail_dict["header"]["subject"])

    @defer.inlineCallbacks
    def test_problem_with_email_raises_exception(self):
        input_mail = InputMail.from_dict(mail_dict(), from_address='pixelated@org')

        when(OutgoingMail).send_message(any(), any()).thenReturn(defer.fail(Exception('pretend something went wrong')))

        try:
            yield self.sender.sendmail(input_mail)
            self.fail('Exception expected!')
        except MailSenderException, e:
            for recipient in flatten([input_mail.to, input_mail.cc, input_mail.bcc]):
                self.assertTrue(recipient in e.email_error_map)

    @defer.inlineCallbacks
    def test_keymanager_encrypt_problem_raises_exception(self):
        input_mail = InputMail.from_dict(mail_dict(), from_address='pixelated@org')

        when(OutgoingMail)._attach_key(any(), any()).thenReturn(
            defer.succeed(None))
        when(OutgoingMail)._fix_headers(any(), any(), any()).thenReturn(
            defer.succeed((None, mock())))
        when(self._keymanager_mock).encrypt(any(), any(), sign=any(),
                                            fetch_remote=any()).thenReturn(defer.fail(Exception('pretend key expired')))

        with patch('leap.bitmask.mail.outgoing.service.emit_async'):
            try:
                yield self.sender.sendmail(input_mail)
                self.fail('Exception expected!')
            except MailSenderException, e:
                for recipient in flatten([input_mail.to, input_mail.cc, input_mail.bcc]):
                    self.assertTrue(recipient in e.email_error_map)

    @defer.inlineCallbacks
    def test_iterates_over_recipients_and_send_whitout_bcc_field(self):
        input_mail = InputMail.from_dict(mail_dict(), from_address='pixelated@org')
        bccs = input_mail.bcc

        when(OutgoingMail).send_message(any(), any()).thenReturn(defer.succeed(None))

        yield self.sender.sendmail(input_mail)

        for recipient in flatten([input_mail.to, input_mail.cc, input_mail.bcc]):
            verify(OutgoingMail).send_message(MailToSmtpFormatCapture(recipient, bccs), TwistedSmtpUserCapture(recipient))