/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: 2010-05-06 23:41:35 UTC
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100506234135-yivbzczw1sejxnxc
Lock methods on ``Tree``, ``Branch`` and ``Repository`` are now
expected to return an object which can be used to unlock them. This reduces
duplicate code when using cleanups. The previous 'tokens's returned by
``Branch.lock_write`` and ``Repository.lock_write`` are now attributes
on the result of the lock_write. ``repository.RepositoryWriteLockResult``
and ``branch.BranchWriteLockResult`` document this. (Robert Collins)

``log._get_info_for_log_files`` now takes an add_cleanup callable.
(Robert Collins)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smtp_connection.py
1
 
# Copyright (C) 2007, 2009, 2010, 2011 Canonical Ltd
 
1
# Copyright (C) 2005, 2007, 2009 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
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
try:
18
 
    from email.message import Message
19
 
except ImportError:  # python < 3
20
 
    from email.Message import Message
 
17
from cStringIO import StringIO
 
18
from email.Message import Message
21
19
import errno
22
20
import smtplib
23
21
import socket
24
22
 
25
 
from breezy import (
 
23
from bzrlib import (
26
24
    config,
27
25
    email_message,
 
26
    errors,
28
27
    smtp_connection,
29
28
    tests,
30
29
    ui,
41
40
 
42
41
class StubSMTPFactory(object):
43
42
    """A fake SMTP connection to test the connection setup."""
44
 
 
45
43
    def __init__(self, fail_on=None, smtp_features=None):
46
44
        self._fail_on = fail_on or []
47
45
        self._calls = []
93
91
class TestSMTPConnection(tests.TestCaseInTempDir):
94
92
 
95
93
    def get_connection(self, text, smtp_factory=None):
96
 
        my_config = config.MemoryStack(text)
97
 
        return smtp_connection.SMTPConnection(
98
 
            my_config, _smtp_factory=smtp_factory)
 
94
        my_config = config.GlobalConfig()
 
95
        config_file = StringIO(text)
 
96
        my_config._get_parser(config_file)
 
97
        return smtp_connection.SMTPConnection(my_config,
 
98
                                              _smtp_factory=smtp_factory)
99
99
 
100
100
    def test_defaults(self):
101
 
        conn = self.get_connection(b'')
 
101
        conn = self.get_connection('')
102
102
        self.assertEqual('localhost', conn._smtp_server)
103
103
        self.assertEqual(None, conn._smtp_username)
104
104
        self.assertEqual(None, conn._smtp_password)
105
105
 
106
106
    def test_smtp_server(self):
107
 
        conn = self.get_connection(b'smtp_server=host:10')
 
107
        conn = self.get_connection('[DEFAULT]\nsmtp_server=host:10\n')
108
108
        self.assertEqual('host:10', conn._smtp_server)
109
109
 
110
110
    def test_missing_server(self):
111
 
        conn = self.get_connection(b'', smtp_factory=connection_refuser)
112
 
        self.assertRaises(smtp_connection.DefaultSMTPConnectionRefused,
113
 
                          conn._connect)
114
 
        conn = self.get_connection(b'smtp_server=smtp.example.com',
 
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',
115
114
                                   smtp_factory=connection_refuser)
116
 
        self.assertRaises(smtp_connection.SMTPConnectionRefused, conn._connect)
 
115
        self.assertRaises(errors.SMTPConnectionRefused, conn._connect)
117
116
 
118
117
    def test_smtp_username(self):
119
 
        conn = self.get_connection(b'')
 
118
        conn = self.get_connection('')
120
119
        self.assertIs(None, conn._smtp_username)
121
120
 
122
 
        conn = self.get_connection(b'smtp_username=joebody')
 
121
        conn = self.get_connection('[DEFAULT]\nsmtp_username=joebody\n')
123
122
        self.assertEqual(u'joebody', conn._smtp_username)
124
123
 
125
124
    def test_smtp_password_from_config(self):
126
 
        conn = self.get_connection(b'')
 
125
        conn = self.get_connection('')
127
126
        self.assertIs(None, conn._smtp_password)
128
127
 
129
 
        conn = self.get_connection(b'smtp_password=mypass')
 
128
        conn = self.get_connection('[DEFAULT]\nsmtp_password=mypass\n')
130
129
        self.assertEqual(u'mypass', conn._smtp_password)
131
130
 
132
131
    def test_smtp_password_from_user(self):
133
132
        user = 'joe'
134
133
        password = 'hispass'
135
134
        factory = WideOpenSMTPFactory()
136
 
        conn = self.get_connection(
137
 
            b'[DEFAULT]\nsmtp_username=%s\n' % user.encode('ascii'),
138
 
            smtp_factory=factory)
 
135
        conn = self.get_connection('[DEFAULT]\nsmtp_username=%s\n' % user,
 
136
                                   smtp_factory=factory)
139
137
        self.assertIs(None, conn._smtp_password)
140
138
 
141
139
        ui.ui_factory = ui.CannedInputUIFactory([password])
146
144
        user = 'joe'
147
145
        password = 'hispass'
148
146
        factory = WideOpenSMTPFactory()
149
 
        conn = self.get_connection(
150
 
            b'[DEFAULT]\nsmtp_username=%s\n' % user.encode('ascii'),
151
 
            smtp_factory=factory)
 
147
        conn = self.get_connection('[DEFAULT]\nsmtp_username=%s\n' % user,
 
148
                                   smtp_factory=factory)
152
149
        self.assertEqual(user, conn._smtp_username)
153
150
        self.assertIs(None, conn._smtp_password)
154
151
        # Create a config file with the right password
155
152
        conf = config.AuthenticationConfig()
156
153
        conf._get_config().update({'smtptest':
157
 
                                   {'scheme': 'smtp', 'user': user,
158
 
                                    'password': password}})
 
154
                                       {'scheme': 'smtp', 'user':user,
 
155
                                        'password': password}})
159
156
        conf._save()
160
157
 
161
158
        conn._connect()
162
159
        self.assertEqual(password, conn._smtp_password)
163
160
 
164
161
    def test_authenticate_with_byte_strings(self):
165
 
        user = b'joe'
166
 
        unicode_pass = u'h\xECspass'
167
 
        utf8_pass = unicode_pass.encode('utf-8')
 
162
        user = 'joe'
 
163
        password = 'h\xC3\xACspass'
168
164
        factory = WideOpenSMTPFactory()
169
165
        conn = self.get_connection(
170
 
            b'[DEFAULT]\nsmtp_username=%s\nsmtp_password=%s\n'
171
 
            % (user, utf8_pass), smtp_factory=factory)
172
 
        self.assertEqual(unicode_pass, conn._smtp_password)
 
166
            '[DEFAULT]\nsmtp_username=%s\nsmtp_password=%s\n'
 
167
            % (user, password), smtp_factory=factory)
 
168
        self.assertEqual(u'h\xECspass', conn._smtp_password)
173
169
        conn._connect()
174
170
        self.assertEqual([('connect', 'localhost'),
175
171
                          ('ehlo',),
176
172
                          ('has_extn', 'starttls'),
177
 
                          ('login', user, utf8_pass)], factory._calls)
 
173
                          ('login', user, password)], factory._calls)
178
174
        smtp_username, smtp_password = factory._calls[-1][1:]
179
 
        self.assertIsInstance(smtp_username, bytes)
180
 
        self.assertIsInstance(smtp_password, bytes)
 
175
        self.assertIsInstance(smtp_username, str)
 
176
        self.assertIsInstance(smtp_password, str)
181
177
 
182
178
    def test_create_connection(self):
183
179
        factory = StubSMTPFactory()
184
 
        conn = self.get_connection(b'', smtp_factory=factory)
 
180
        conn = self.get_connection('', smtp_factory=factory)
185
181
        conn._create_connection()
186
182
        self.assertEqual([('connect', 'localhost'),
187
183
                          ('ehlo',),
190
186
    def test_create_connection_ehlo_fails(self):
191
187
        # Check that we call HELO if EHLO failed.
192
188
        factory = StubSMTPFactory(fail_on=['ehlo'])
193
 
        conn = self.get_connection(b'', smtp_factory=factory)
 
189
        conn = self.get_connection('', smtp_factory=factory)
194
190
        conn._create_connection()
195
191
        self.assertEqual([('connect', 'localhost'),
196
192
                          ('ehlo',),
200
196
    def test_create_connection_ehlo_helo_fails(self):
201
197
        # Check that we raise an exception if both EHLO and HELO fail.
202
198
        factory = StubSMTPFactory(fail_on=['ehlo', 'helo'])
203
 
        conn = self.get_connection(b'', smtp_factory=factory)
204
 
        self.assertRaises(smtp_connection.SMTPError, conn._create_connection)
 
199
        conn = self.get_connection('', smtp_factory=factory)
 
200
        self.assertRaises(errors.SMTPError, conn._create_connection)
205
201
        self.assertEqual([('connect', 'localhost'),
206
202
                          ('ehlo',),
207
203
                          ('helo',)], factory._calls)
210
206
        # Check that STARTTLS plus a second EHLO are called if the
211
207
        # server says it supports the feature.
212
208
        factory = StubSMTPFactory(smtp_features=['starttls'])
213
 
        conn = self.get_connection(b'', smtp_factory=factory)
 
209
        conn = self.get_connection('', smtp_factory=factory)
214
210
        conn._create_connection()
215
211
        self.assertEqual([('connect', 'localhost'),
216
212
                          ('ehlo',),
223
219
        # support STARTTLS, but then fails when we try to activate it.
224
220
        factory = StubSMTPFactory(fail_on=['starttls'],
225
221
                                  smtp_features=['starttls'])
226
 
        conn = self.get_connection(b'', smtp_factory=factory)
227
 
        self.assertRaises(smtp_connection.SMTPError, conn._create_connection)
 
222
        conn = self.get_connection('', smtp_factory=factory)
 
223
        self.assertRaises(errors.SMTPError, conn._create_connection)
228
224
        self.assertEqual([('connect', 'localhost'),
229
225
                          ('ehlo',),
230
226
                          ('has_extn', 'starttls'),
245
241
        from_, to = smtp_connection.SMTPConnection.get_message_addresses(msg)
246
242
        self.assertEqual('jrandom@example.com', from_)
247
243
        self.assertEqual(sorted(['john@doe.com', 'jane@doe.com',
248
 
                                 'pperez@ejemplo.com', 'user@localhost']), sorted(to))
 
244
            'pperez@ejemplo.com', 'user@localhost']), sorted(to))
249
245
 
250
 
        # now with breezy's EmailMessage
 
246
        # now with bzrlib's EmailMessage
251
247
        msg = email_message.EmailMessage(
252
248
            '"J. Random Developer" <jrandom@example.com>',
253
249
            ['John Doe <john@doe.com>', 'Jane Doe <jane@doe.com>',
254
 
             u'Pepe P\xe9rez <pperez@ejemplo.com>', 'user@localhost'],
 
250
             u'Pepe P\xe9rez <pperez@ejemplo.com>', 'user@localhost' ],
255
251
            'subject')
256
252
 
257
253
        from_, to = smtp_connection.SMTPConnection.get_message_addresses(msg)
258
254
        self.assertEqual('jrandom@example.com', from_)
259
255
        self.assertEqual(sorted(['john@doe.com', 'jane@doe.com',
260
 
                                 'pperez@ejemplo.com', 'user@localhost']), sorted(to))
 
256
            'pperez@ejemplo.com', 'user@localhost']), sorted(to))
261
257
 
262
258
    def test_destination_address_required(self):
 
259
        class FakeConfig:
 
260
            def get_user_option(self, option):
 
261
                return None
 
262
 
263
263
        msg = Message()
264
264
        msg['From'] = '"J. Random Developer" <jrandom@example.com>'
265
265
        self.assertRaises(
266
 
            smtp_connection.NoDestinationAddress,
267
 
            smtp_connection.SMTPConnection(config.MemoryStack(b"")
268
 
                                           ).send_email, msg)
 
266
            errors.NoDestinationAddress,
 
267
            smtp_connection.SMTPConnection(FakeConfig()).send_email, msg)
269
268
 
270
269
        msg = email_message.EmailMessage('from@from.com', '', 'subject')
271
270
        self.assertRaises(
272
 
            smtp_connection.NoDestinationAddress,
273
 
            smtp_connection.SMTPConnection(config.MemoryStack(b"")
274
 
                                           ).send_email, msg)
 
271
            errors.NoDestinationAddress,
 
272
            smtp_connection.SMTPConnection(FakeConfig()).send_email, msg)
275
273
 
276
274
        msg = email_message.EmailMessage('from@from.com', [], 'subject')
277
275
        self.assertRaises(
278
 
            smtp_connection.NoDestinationAddress,
279
 
            smtp_connection.SMTPConnection(config.MemoryStack(b"")
280
 
                                           ).send_email, msg)
 
276
            errors.NoDestinationAddress,
 
277
            smtp_connection.SMTPConnection(FakeConfig()).send_email, msg)