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
17
from cStringIO import StringIO
18
from email.Message import Message
18
from email.message import Message
19
except ImportError: # python < 3
20
from email.Message import Message
91
92
class TestSMTPConnection(tests.TestCaseInTempDir):
93
94
def get_connection(self, text, smtp_factory=None):
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)
95
my_config = config.MemoryStack(text)
96
return smtp_connection.SMTPConnection(
97
my_config, _smtp_factory=smtp_factory)
100
99
def test_defaults(self):
101
conn = self.get_connection('')
100
conn = self.get_connection(b'')
102
101
self.assertEqual('localhost', conn._smtp_server)
103
102
self.assertEqual(None, conn._smtp_username)
104
103
self.assertEqual(None, conn._smtp_password)
106
105
def test_smtp_server(self):
107
conn = self.get_connection('[DEFAULT]\nsmtp_server=host:10\n')
106
conn = self.get_connection(b'smtp_server=host:10')
108
107
self.assertEqual('host:10', conn._smtp_server)
110
109
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',
110
conn = self.get_connection(b'', smtp_factory=connection_refuser)
111
self.assertRaises(smtp_connection.DefaultSMTPConnectionRefused,
113
conn = self.get_connection(b'smtp_server=smtp.example.com',
114
114
smtp_factory=connection_refuser)
115
self.assertRaises(errors.SMTPConnectionRefused, conn._connect)
115
self.assertRaises(smtp_connection.SMTPConnectionRefused, conn._connect)
117
117
def test_smtp_username(self):
118
conn = self.get_connection('')
118
conn = self.get_connection(b'')
119
119
self.assertIs(None, conn._smtp_username)
121
conn = self.get_connection('[DEFAULT]\nsmtp_username=joebody\n')
121
conn = self.get_connection(b'smtp_username=joebody')
122
122
self.assertEqual(u'joebody', conn._smtp_username)
124
124
def test_smtp_password_from_config(self):
125
conn = self.get_connection('')
125
conn = self.get_connection(b'')
126
126
self.assertIs(None, conn._smtp_password)
128
conn = self.get_connection('[DEFAULT]\nsmtp_password=mypass\n')
128
conn = self.get_connection(b'smtp_password=mypass')
129
129
self.assertEqual(u'mypass', conn._smtp_password)
131
131
def test_smtp_password_from_user(self):
133
133
password = 'hispass'
134
134
factory = WideOpenSMTPFactory()
135
conn = self.get_connection('[DEFAULT]\nsmtp_username=%s\n' % user,
136
smtp_factory=factory)
135
conn = self.get_connection(
136
b'[DEFAULT]\nsmtp_username=%s\n' % user.encode('ascii'),
137
smtp_factory=factory)
137
138
self.assertIs(None, conn._smtp_password)
139
140
ui.ui_factory = ui.CannedInputUIFactory([password])
159
161
self.assertEqual(password, conn._smtp_password)
161
163
def test_authenticate_with_byte_strings(self):
163
password = 'h\xC3\xACspass'
165
unicode_pass = u'h\xECspass'
166
utf8_pass = unicode_pass.encode('utf-8')
164
167
factory = WideOpenSMTPFactory()
165
168
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
b'[DEFAULT]\nsmtp_username=%s\nsmtp_password=%s\n'
170
% (user, utf8_pass), smtp_factory=factory)
171
self.assertEqual(unicode_pass, conn._smtp_password)
170
173
self.assertEqual([('connect', 'localhost'),
172
175
('has_extn', 'starttls'),
173
('login', user, password)], factory._calls)
176
('login', user, utf8_pass)], factory._calls)
174
177
smtp_username, smtp_password = factory._calls[-1][1:]
175
self.assertIsInstance(smtp_username, str)
176
self.assertIsInstance(smtp_password, str)
178
self.assertIsInstance(smtp_username, bytes)
179
self.assertIsInstance(smtp_password, bytes)
178
181
def test_create_connection(self):
179
182
factory = StubSMTPFactory()
180
conn = self.get_connection('', smtp_factory=factory)
183
conn = self.get_connection(b'', smtp_factory=factory)
181
184
conn._create_connection()
182
185
self.assertEqual([('connect', 'localhost'),
196
199
def test_create_connection_ehlo_helo_fails(self):
197
200
# Check that we raise an exception if both EHLO and HELO fail.
198
201
factory = StubSMTPFactory(fail_on=['ehlo', 'helo'])
199
conn = self.get_connection('', smtp_factory=factory)
200
self.assertRaises(errors.SMTPError, conn._create_connection)
202
conn = self.get_connection(b'', smtp_factory=factory)
203
self.assertRaises(smtp_connection.SMTPError, conn._create_connection)
201
204
self.assertEqual([('connect', 'localhost'),
203
206
('helo',)], factory._calls)
206
209
# Check that STARTTLS plus a second EHLO are called if the
207
210
# server says it supports the feature.
208
211
factory = StubSMTPFactory(smtp_features=['starttls'])
209
conn = self.get_connection('', smtp_factory=factory)
212
conn = self.get_connection(b'', smtp_factory=factory)
210
213
conn._create_connection()
211
214
self.assertEqual([('connect', 'localhost'),
219
222
# support STARTTLS, but then fails when we try to activate it.
220
223
factory = StubSMTPFactory(fail_on=['starttls'],
221
224
smtp_features=['starttls'])
222
conn = self.get_connection('', smtp_factory=factory)
223
self.assertRaises(errors.SMTPError, conn._create_connection)
225
conn = self.get_connection(b'', smtp_factory=factory)
226
self.assertRaises(smtp_connection.SMTPError, conn._create_connection)
224
227
self.assertEqual([('connect', 'localhost'),
226
229
('has_extn', 'starttls'),
243
246
self.assertEqual(sorted(['john@doe.com', 'jane@doe.com',
244
247
'pperez@ejemplo.com', 'user@localhost']), sorted(to))
246
# now with bzrlib's EmailMessage
249
# now with breezy's EmailMessage
247
250
msg = email_message.EmailMessage(
248
251
'"J. Random Developer" <jrandom@example.com>',
249
252
['John Doe <john@doe.com>', 'Jane Doe <jane@doe.com>',
256
259
'pperez@ejemplo.com', 'user@localhost']), sorted(to))
258
261
def test_destination_address_required(self):
260
def get_user_option(self, option):
264
263
msg['From'] = '"J. Random Developer" <jrandom@example.com>'
265
264
self.assertRaises(
266
errors.NoDestinationAddress,
267
smtp_connection.SMTPConnection(FakeConfig()).send_email, msg)
265
smtp_connection.NoDestinationAddress,
266
smtp_connection.SMTPConnection(config.MemoryStack(b"")
269
269
msg = email_message.EmailMessage('from@from.com', '', 'subject')
270
270
self.assertRaises(
271
errors.NoDestinationAddress,
272
smtp_connection.SMTPConnection(FakeConfig()).send_email, msg)
271
smtp_connection.NoDestinationAddress,
272
smtp_connection.SMTPConnection(config.MemoryStack(b"")
274
275
msg = email_message.EmailMessage('from@from.com', [], 'subject')
275
276
self.assertRaises(
276
errors.NoDestinationAddress,
277
smtp_connection.SMTPConnection(FakeConfig()).send_email, msg)
277
smtp_connection.NoDestinationAddress,
278
smtp_connection.SMTPConnection(config.MemoryStack(b"")