/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smtp_connection.py

  • Committer: Robert Collins
  • Date: 2007-07-15 15:40:37 UTC
  • mto: (2592.3.33 repository)
  • mto: This revision was merged to the branch mainline in revision 2624.
  • Revision ID: robertc@robertcollins.net-20070715154037-3ar8g89decddc9su
Make GraphIndex accept nodes as key, value, references, so that the method
signature is closer to what a simple key->value index delivers. Also
change the behaviour when the reference list count is zero to accept
key, value as nodes, and emit key, value to make it identical in that case
to a simple key->value index. This may not be a good idea, but for now it
seems ok.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smtp_connection.py
1
 
# Copyright (C) 2005, 2007, 2009 Canonical Ltd
 
1
# Copyright (C) 2005, 2007 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
from cStringIO import StringIO
18
18
from email.Message import Message
19
 
import errno
20
 
import smtplib
21
 
import socket
22
 
 
23
 
from bzrlib import (
24
 
    config,
25
 
    email_message,
26
 
    errors,
27
 
    smtp_connection,
28
 
    tests,
29
 
    ui,
30
 
    )
31
 
 
32
 
 
33
 
def connection_refuser():
34
 
    def connect(server):
35
 
        raise socket.error(errno.ECONNREFUSED, 'Connection Refused')
36
 
    smtp = smtplib.SMTP()
37
 
    smtp.connect = connect
38
 
    return smtp
39
 
 
40
 
 
41
 
class StubSMTPFactory(object):
42
 
    """A fake SMTP connection to test the connection setup."""
43
 
    def __init__(self, fail_on=None, smtp_features=None):
44
 
        self._fail_on = fail_on or []
45
 
        self._calls = []
46
 
        self._smtp_features = smtp_features or []
47
 
        self._ehlo_called = False
48
 
 
49
 
    def __call__(self):
50
 
        # The factory pretends to be a connection
51
 
        return self
52
 
 
53
 
    def connect(self, server):
54
 
        self._calls.append(('connect', server))
55
 
 
56
 
    def helo(self):
57
 
        self._calls.append(('helo',))
58
 
        if 'helo' in self._fail_on:
59
 
            return 500, 'helo failure'
60
 
        else:
61
 
            return 200, 'helo success'
62
 
 
63
 
    def ehlo(self):
64
 
        self._calls.append(('ehlo',))
65
 
        if 'ehlo' in self._fail_on:
66
 
            return 500, 'ehlo failure'
67
 
        else:
68
 
            self._ehlo_called = True
69
 
            return 200, 'ehlo success'
70
 
 
71
 
    def has_extn(self, extension):
72
 
        self._calls.append(('has_extn', extension))
73
 
        return self._ehlo_called and extension in self._smtp_features
74
 
 
75
 
    def starttls(self):
76
 
        self._calls.append(('starttls',))
77
 
        if 'starttls' in self._fail_on:
78
 
            return 500, 'starttls failure'
79
 
        else:
80
 
            self._ehlo_called = True
81
 
            return 200, 'starttls success'
82
 
 
83
 
 
84
 
class WideOpenSMTPFactory(StubSMTPFactory):
85
 
    """A fake smtp server that implements login by accepting anybody."""
86
 
 
87
 
    def login(self, user, password):
88
 
        self._calls.append(('login', user, password))
89
 
 
90
 
 
91
 
class TestSMTPConnection(tests.TestCaseInTempDir):
92
 
 
93
 
    def get_connection(self, text, smtp_factory=None):
 
19
 
 
20
from bzrlib import config
 
21
from bzrlib.errors import NoDestinationAddress
 
22
from bzrlib.tests import TestCase
 
23
from bzrlib.smtp_connection import SMTPConnection
 
24
 
 
25
 
 
26
class TestSMTPConnection(TestCase):
 
27
 
 
28
    def get_connection(self, text):
94
29
        my_config = config.GlobalConfig()
95
30
        config_file = StringIO(text)
96
31
        my_config._get_parser(config_file)
97
 
        return smtp_connection.SMTPConnection(my_config,
98
 
                                              _smtp_factory=smtp_factory)
 
32
        return SMTPConnection(my_config)
99
33
 
100
34
    def test_defaults(self):
101
35
        conn = self.get_connection('')
107
41
        conn = self.get_connection('[DEFAULT]\nsmtp_server=host:10\n')
108
42
        self.assertEqual('host:10', conn._smtp_server)
109
43
 
110
 
    def test_missing_server(self):
111
 
        conn = self.get_connection('', smtp_factory=connection_refuser)
112
 
        self.assertRaises(errors.DefaultSMTPConnectionRefused, conn._connect)
113
 
        conn = self.get_connection('[DEFAULT]\nsmtp_server=smtp.example.com\n',
114
 
                                   smtp_factory=connection_refuser)
115
 
        self.assertRaises(errors.SMTPConnectionRefused, conn._connect)
116
 
 
117
44
    def test_smtp_username(self):
118
45
        conn = self.get_connection('')
119
46
        self.assertIs(None, conn._smtp_username)
121
48
        conn = self.get_connection('[DEFAULT]\nsmtp_username=joebody\n')
122
49
        self.assertEqual(u'joebody', conn._smtp_username)
123
50
 
124
 
    def test_smtp_password_from_config(self):
 
51
    def test_smtp_password(self):
125
52
        conn = self.get_connection('')
126
53
        self.assertIs(None, conn._smtp_password)
127
54
 
128
55
        conn = self.get_connection('[DEFAULT]\nsmtp_password=mypass\n')
129
56
        self.assertEqual(u'mypass', conn._smtp_password)
130
57
 
131
 
    def test_smtp_password_from_user(self):
132
 
        user = 'joe'
133
 
        password = 'hispass'
134
 
        factory = WideOpenSMTPFactory()
135
 
        conn = self.get_connection('[DEFAULT]\nsmtp_username=%s\n' % user,
136
 
                                   smtp_factory=factory)
137
 
        self.assertIs(None, conn._smtp_password)
138
 
 
139
 
        ui.ui_factory = ui.CannedInputUIFactory([password])
140
 
        conn._connect()
141
 
        self.assertEqual(password, conn._smtp_password)
142
 
 
143
 
    def test_smtp_password_from_auth_config(self):
144
 
        user = 'joe'
145
 
        password = 'hispass'
146
 
        factory = WideOpenSMTPFactory()
147
 
        conn = self.get_connection('[DEFAULT]\nsmtp_username=%s\n' % user,
148
 
                                   smtp_factory=factory)
149
 
        self.assertEqual(user, conn._smtp_username)
150
 
        self.assertIs(None, conn._smtp_password)
151
 
        # Create a config file with the right password
152
 
        conf = config.AuthenticationConfig()
153
 
        conf._get_config().update({'smtptest':
154
 
                                       {'scheme': 'smtp', 'user':user,
155
 
                                        'password': password}})
156
 
        conf._save()
157
 
 
158
 
        conn._connect()
159
 
        self.assertEqual(password, conn._smtp_password)
160
 
 
161
 
    def test_authenticate_with_byte_strings(self):
162
 
        user = 'joe'
163
 
        password = 'h\xC3\xACspass'
164
 
        factory = WideOpenSMTPFactory()
165
 
        conn = self.get_connection(
166
 
            '[DEFAULT]\nsmtp_username=%s\nsmtp_password=%s\n'
167
 
            % (user, password), smtp_factory=factory)
168
 
        self.assertEqual(u'h\xECspass', conn._smtp_password)
169
 
        conn._connect()
170
 
        self.assertEqual([('connect', 'localhost'),
171
 
                          ('ehlo',),
172
 
                          ('has_extn', 'starttls'),
173
 
                          ('login', user, password)], factory._calls)
174
 
        smtp_username, smtp_password = factory._calls[-1][1:]
175
 
        self.assertIsInstance(smtp_username, str)
176
 
        self.assertIsInstance(smtp_password, str)
177
 
 
178
 
    def test_create_connection(self):
179
 
        factory = StubSMTPFactory()
180
 
        conn = self.get_connection('', smtp_factory=factory)
181
 
        conn._create_connection()
182
 
        self.assertEqual([('connect', 'localhost'),
183
 
                          ('ehlo',),
184
 
                          ('has_extn', 'starttls')], factory._calls)
185
 
 
186
 
    def test_create_connection_ehlo_fails(self):
187
 
        # Check that we call HELO if EHLO failed.
188
 
        factory = StubSMTPFactory(fail_on=['ehlo'])
189
 
        conn = self.get_connection('', smtp_factory=factory)
190
 
        conn._create_connection()
191
 
        self.assertEqual([('connect', 'localhost'),
192
 
                          ('ehlo',),
193
 
                          ('helo',),
194
 
                          ('has_extn', 'starttls')], factory._calls)
195
 
 
196
 
    def test_create_connection_ehlo_helo_fails(self):
197
 
        # Check that we raise an exception if both EHLO and HELO fail.
198
 
        factory = StubSMTPFactory(fail_on=['ehlo', 'helo'])
199
 
        conn = self.get_connection('', smtp_factory=factory)
200
 
        self.assertRaises(errors.SMTPError, conn._create_connection)
201
 
        self.assertEqual([('connect', 'localhost'),
202
 
                          ('ehlo',),
203
 
                          ('helo',)], factory._calls)
204
 
 
205
 
    def test_create_connection_starttls(self):
206
 
        # Check that STARTTLS plus a second EHLO are called if the
207
 
        # server says it supports the feature.
208
 
        factory = StubSMTPFactory(smtp_features=['starttls'])
209
 
        conn = self.get_connection('', smtp_factory=factory)
210
 
        conn._create_connection()
211
 
        self.assertEqual([('connect', 'localhost'),
212
 
                          ('ehlo',),
213
 
                          ('has_extn', 'starttls'),
214
 
                          ('starttls',),
215
 
                          ('ehlo',)], factory._calls)
216
 
 
217
 
    def test_create_connection_starttls_fails(self):
218
 
        # Check that we raise an exception if the server claims to
219
 
        # support STARTTLS, but then fails when we try to activate it.
220
 
        factory = StubSMTPFactory(fail_on=['starttls'],
221
 
                                  smtp_features=['starttls'])
222
 
        conn = self.get_connection('', smtp_factory=factory)
223
 
        self.assertRaises(errors.SMTPError, conn._create_connection)
224
 
        self.assertEqual([('connect', 'localhost'),
225
 
                          ('ehlo',),
226
 
                          ('has_extn', 'starttls'),
227
 
                          ('starttls',)], factory._calls)
228
 
 
229
58
    def test_get_message_addresses(self):
230
59
        msg = Message()
231
60
 
232
 
        from_, to = smtp_connection.SMTPConnection.get_message_addresses(msg)
 
61
        from_, to = SMTPConnection.get_message_addresses(msg)
233
62
        self.assertEqual('', from_)
234
63
        self.assertEqual([], to)
235
64
 
238
67
        msg['CC'] = u'Pepe P\xe9rez <pperez@ejemplo.com>'
239
68
        msg['Bcc'] = 'user@localhost'
240
69
 
241
 
        from_, to = smtp_connection.SMTPConnection.get_message_addresses(msg)
242
 
        self.assertEqual('jrandom@example.com', from_)
243
 
        self.assertEqual(sorted(['john@doe.com', 'jane@doe.com',
244
 
            'pperez@ejemplo.com', 'user@localhost']), sorted(to))
245
 
 
246
 
        # now with bzrlib's EmailMessage
247
 
        msg = email_message.EmailMessage(
248
 
            '"J. Random Developer" <jrandom@example.com>',
249
 
            ['John Doe <john@doe.com>', 'Jane Doe <jane@doe.com>',
250
 
             u'Pepe P\xe9rez <pperez@ejemplo.com>', 'user@localhost' ],
251
 
            'subject')
252
 
 
253
 
        from_, to = smtp_connection.SMTPConnection.get_message_addresses(msg)
 
70
        from_, to = SMTPConnection.get_message_addresses(msg)
254
71
        self.assertEqual('jrandom@example.com', from_)
255
72
        self.assertEqual(sorted(['john@doe.com', 'jane@doe.com',
256
73
            'pperez@ejemplo.com', 'user@localhost']), sorted(to))
262
79
 
263
80
        msg = Message()
264
81
        msg['From'] = '"J. Random Developer" <jrandom@example.com>'
265
 
        self.assertRaises(
266
 
            errors.NoDestinationAddress,
267
 
            smtp_connection.SMTPConnection(FakeConfig()).send_email, msg)
268
 
 
269
 
        msg = email_message.EmailMessage('from@from.com', '', 'subject')
270
 
        self.assertRaises(
271
 
            errors.NoDestinationAddress,
272
 
            smtp_connection.SMTPConnection(FakeConfig()).send_email, msg)
273
 
 
274
 
        msg = email_message.EmailMessage('from@from.com', [], 'subject')
275
 
        self.assertRaises(
276
 
            errors.NoDestinationAddress,
277
 
            smtp_connection.SMTPConnection(FakeConfig()).send_email, msg)
 
82
        self.assertRaises(NoDestinationAddress,
 
83
                SMTPConnection(FakeConfig()).send_email, msg)