Skip to content

Commit a7613aa

Browse files
bitdancerbitdancer
committed
Merge: python#25591: improve imap tests.
2 parents da31ba9 + b079c07 commit a7613aa

File tree

2 files changed

+331
-1
lines changed

2 files changed

+331
-1
lines changed

‎Lib/imaplib.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -419,7 +419,7 @@ def authenticate(self, mechanism, authobject):
419419
self.literal = _Authenticator(authobject).process
420420
typ, dat = self._simple_command('AUTHENTICATE', mech)
421421
if typ != 'OK':
422-
raise self.error(dat[-1])
422+
raise self.error(dat[-1].decode('utf-8', 'replace'))
423423
self.state = 'AUTH'
424424
return typ, dat
425425

‎Lib/test/test_imaplib.py‎

Lines changed: 330 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,10 +10,12 @@
1010
import socketserver
1111
import time
1212
import calendar
13+
import inspect
1314

1415
from test.support import (reap_threads, verbose, transient_internet,
1516
run_with_tz, run_with_locale)
1617
import unittest
18+
from unittest import mock
1719
from datetime import datetime, timezone, timedelta
1820
try:
1921
import ssl
@@ -174,6 +176,334 @@ def cmd_LOGIN(self, tag, args):
174176
self._send_tagged(tag, 'OK', 'LOGIN completed')
175177

176178

179+
class NewIMAPTestsMixin():
180+
client = None
181+
182+
def _setup(self, imap_handler, connect=True):
183+
"""
184+
Sets up imap_handler for tests. imap_handler should inherit from either:
185+
- SimpleIMAPHandler - for testing IMAP commands,
186+
- socketserver.StreamRequestHandler - if raw access to stream is needed.
187+
Returns (client, server).
188+
"""
189+
class TestTCPServer(self.server_class):
190+
def handle_error(self, request, client_address):
191+
"""
192+
End request and raise the error if one occurs.
193+
"""
194+
self.close_request(request)
195+
self.server_close()
196+
raise
197+
198+
self.addCleanup(self._cleanup)
199+
self.server = self.server_class((support.HOST, 0), imap_handler)
200+
self.thread = threading.Thread(
201+
name=self._testMethodName+'-server',
202+
target=self.server.serve_forever,
203+
# Short poll interval to make the test finish quickly.
204+
# Time between requests is short enough that we won't wake
205+
# up spuriously too many times.
206+
kwargs={'poll_interval': 0.01})
207+
self.thread.daemon = True # In case this function raises.
208+
self.thread.start()
209+
210+
if connect:
211+
self.client = self.imap_class(*self.server.server_address)
212+
213+
return self.client, self.server
214+
215+
def _cleanup(self):
216+
"""
217+
Cleans up the test server. This method should not be called manually,
218+
it is added to the cleanup queue in the _setup method already.
219+
"""
220+
# if logout was called already we'd raise an exception trying to
221+
# shutdown the client once again
222+
if self.client is not None and self.client.state != 'LOGOUT':
223+
self.client.shutdown()
224+
# cleanup the server
225+
self.server.shutdown()
226+
self.server.server_close()
227+
self.thread.join(3.0)
228+
229+
def test_EOF_without_complete_welcome_message(self):
230+
# http://bugs.python.org/issue5949
231+
class EOFHandler(socketserver.StreamRequestHandler):
232+
def handle(self):
233+
self.wfile.write(b'* OK')
234+
_, server = self._setup(EOFHandler, connect=False)
235+
self.assertRaises(imaplib.IMAP4.abort, self.imap_class,
236+
*server.server_address)
237+
238+
def test_line_termination(self):
239+
class BadNewlineHandler(SimpleIMAPHandler):
240+
def cmd_CAPABILITY(self, tag, args):
241+
self._send(b'* CAPABILITY IMAP4rev1 AUTH\n')
242+
self._send_tagged(tag, 'OK', 'CAPABILITY completed')
243+
_, server = self._setup(BadNewlineHandler, connect=False)
244+
self.assertRaises(imaplib.IMAP4.abort, self.imap_class,
245+
*server.server_address)
246+
247+
def test_enable_raises_error_if_not_AUTH(self):
248+
class EnableHandler(SimpleIMAPHandler):
249+
capabilities = 'AUTH ENABLE UTF8=ACCEPT'
250+
client, _ = self._setup(EnableHandler)
251+
self.assertFalse(client.utf8_enabled)
252+
with self.assertRaisesRegex(imaplib.IMAP4.error, 'ENABLE.*NONAUTH'):
253+
client.enable('foo')
254+
self.assertFalse(client.utf8_enabled)
255+
256+
def test_enable_raises_error_if_no_capability(self):
257+
client, _ = self._setup(SimpleIMAPHandler)
258+
with self.assertRaisesRegex(imaplib.IMAP4.error,
259+
'does not support ENABLE'):
260+
client.enable('foo')
261+
262+
def test_enable_UTF8_raises_error_if_not_supported(self):
263+
client, _ = self._setup(SimpleIMAPHandler)
264+
typ, data = client.login('user', 'pass')
265+
self.assertEqual(typ, 'OK')
266+
with self.assertRaisesRegex(imaplib.IMAP4.error,
267+
'does not support ENABLE'):
268+
client.enable('UTF8=ACCEPT')
269+
270+
def test_enable_UTF8_True_append(self):
271+
class UTF8AppendServer(SimpleIMAPHandler):
272+
capabilities = 'ENABLE UTF8=ACCEPT'
273+
def cmd_ENABLE(self, tag, args):
274+
self._send_tagged(tag, 'OK', 'ENABLE successful')
275+
def cmd_AUTHENTICATE(self, tag, args):
276+
self._send_textline('+')
277+
self.server.response = yield
278+
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
279+
def cmd_APPEND(self, tag, args):
280+
self._send_textline('+')
281+
self.server.response = yield
282+
self._send_tagged(tag, 'OK', 'okay')
283+
client, server = self._setup(UTF8AppendServer)
284+
self.assertEqual(client._encoding, 'ascii')
285+
code, _ = client.authenticate('MYAUTH', lambda x: b'fake')
286+
self.assertEqual(code, 'OK')
287+
self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake'
288+
code, _ = client.enable('UTF8=ACCEPT')
289+
self.assertEqual(code, 'OK')
290+
self.assertEqual(client._encoding, 'utf-8')
291+
msg_string = 'Subject: üñí©öðé'
292+
typ, data = client.append(None, None, None, msg_string.encode('utf-8'))
293+
self.assertEqual(typ, 'OK')
294+
self.assertEqual(server.response,
295+
('UTF8 (%s)\r\n' % msg_string).encode('utf-8'))
296+
297+
def test_search_disallows_charset_in_utf8_mode(self):
298+
class UTF8Server(SimpleIMAPHandler):
299+
capabilities = 'AUTH ENABLE UTF8=ACCEPT'
300+
def cmd_ENABLE(self, tag, args):
301+
self._send_tagged(tag, 'OK', 'ENABLE successful')
302+
def cmd_AUTHENTICATE(self, tag, args):
303+
self._send_textline('+')
304+
self.server.response = yield
305+
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
306+
client, _ = self._setup(UTF8Server)
307+
typ, _ = client.authenticate('MYAUTH', lambda x: b'fake')
308+
self.assertEqual(typ, 'OK')
309+
typ, _ = client.enable('UTF8=ACCEPT')
310+
self.assertEqual(typ, 'OK')
311+
self.assertTrue(client.utf8_enabled)
312+
with self.assertRaisesRegex(imaplib.IMAP4.error, 'charset.*UTF8'):
313+
client.search('foo', 'bar')
314+
315+
def test_bad_auth_name(self):
316+
class MyServer(SimpleIMAPHandler):
317+
def cmd_AUTHENTICATE(self, tag, args):
318+
self._send_tagged(tag, 'NO',
319+
'unrecognized authentication type {}'.format(args[0]))
320+
client, _ = self._setup(MyServer)
321+
with self.assertRaisesRegex(imaplib.IMAP4.error,
322+
'unrecognized authentication type METHOD'):
323+
client.authenticate('METHOD', lambda: 1)
324+
325+
def test_invalid_authentication(self):
326+
class MyServer(SimpleIMAPHandler):
327+
def cmd_AUTHENTICATE(self, tag, args):
328+
self._send_textline('+')
329+
self.response = yield
330+
self._send_tagged(tag, 'NO', '[AUTHENTICATIONFAILED] invalid')
331+
client, _ = self._setup(MyServer)
332+
with self.assertRaisesRegex(imaplib.IMAP4.error,
333+
r'\[AUTHENTICATIONFAILED\] invalid'):
334+
client.authenticate('MYAUTH', lambda x: b'fake')
335+
336+
def test_valid_authentication_bytes(self):
337+
class MyServer(SimpleIMAPHandler):
338+
def cmd_AUTHENTICATE(self, tag, args):
339+
self._send_textline('+')
340+
self.server.response = yield
341+
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
342+
client, server = self._setup(MyServer)
343+
code, _ = client.authenticate('MYAUTH', lambda x: b'fake')
344+
self.assertEqual(code, 'OK')
345+
self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake'
346+
347+
def test_valid_authentication_plain_text(self):
348+
class MyServer(SimpleIMAPHandler):
349+
def cmd_AUTHENTICATE(self, tag, args):
350+
self._send_textline('+')
351+
self.server.response = yield
352+
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
353+
client, server = self._setup(MyServer)
354+
code, _ = client.authenticate('MYAUTH', lambda x: 'fake')
355+
self.assertEqual(code, 'OK')
356+
self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake'
357+
358+
def test_login_cram_md5_bytes(self):
359+
class AuthHandler(SimpleIMAPHandler):
360+
capabilities = 'LOGINDISABLED AUTH=CRAM-MD5'
361+
def cmd_AUTHENTICATE(self, tag, args):
362+
self._send_textline('+ PDE4OTYuNjk3MTcwOTUyQHBvc3RvZmZpY2Uucm'
363+
'VzdG9uLm1jaS5uZXQ=')
364+
r = yield
365+
if (r == b'dGltIGYxY2E2YmU0NjRiOWVmYT'
366+
b'FjY2E2ZmZkNmNmMmQ5ZjMy\r\n'):
367+
self._send_tagged(tag, 'OK', 'CRAM-MD5 successful')
368+
else:
369+
self._send_tagged(tag, 'NO', 'No access')
370+
client, _ = self._setup(AuthHandler)
371+
self.assertTrue('AUTH=CRAM-MD5' in client.capabilities)
372+
ret, _ = client.login_cram_md5("tim", b"tanstaaftanstaaf")
373+
self.assertEqual(ret, "OK")
374+
375+
def test_login_cram_md5_plain_text(self):
376+
class AuthHandler(SimpleIMAPHandler):
377+
capabilities = 'LOGINDISABLED AUTH=CRAM-MD5'
378+
def cmd_AUTHENTICATE(self, tag, args):
379+
self._send_textline('+ PDE4OTYuNjk3MTcwOTUyQHBvc3RvZmZpY2Uucm'
380+
'VzdG9uLm1jaS5uZXQ=')
381+
r = yield
382+
if (r == b'dGltIGYxY2E2YmU0NjRiOWVmYT'
383+
b'FjY2E2ZmZkNmNmMmQ5ZjMy\r\n'):
384+
self._send_tagged(tag, 'OK', 'CRAM-MD5 successful')
385+
else:
386+
self._send_tagged(tag, 'NO', 'No access')
387+
client, _ = self._setup(AuthHandler)
388+
self.assertTrue('AUTH=CRAM-MD5' in client.capabilities)
389+
ret, _ = client.login_cram_md5("tim", "tanstaaftanstaaf")
390+
self.assertEqual(ret, "OK")
391+
392+
def test_aborted_authentication(self):
393+
class MyServer(SimpleIMAPHandler):
394+
def cmd_AUTHENTICATE(self, tag, args):
395+
self._send_textline('+')
396+
self.response = yield
397+
if self.response == b'*\r\n':
398+
self._send_tagged(
399+
tag,
400+
'NO',
401+
'[AUTHENTICATIONFAILED] aborted')
402+
else:
403+
self._send_tagged(tag, 'OK', 'MYAUTH successful')
404+
client, _ = self._setup(MyServer)
405+
with self.assertRaisesRegex(imaplib.IMAP4.error,
406+
r'\[AUTHENTICATIONFAILED\] aborted'):
407+
client.authenticate('MYAUTH', lambda x: None)
408+
409+
@mock.patch('imaplib._MAXLINE', 10)
410+
def test_linetoolong(self):
411+
class TooLongHandler(SimpleIMAPHandler):
412+
def handle(self):
413+
# send response line longer than the limit set in the next line
414+
self.wfile.write(b'* OK ' + 11 * b'x' + b'\r\n')
415+
_, server = self._setup(TooLongHandler, connect=False)
416+
with self.assertRaisesRegex(imaplib.IMAP4.error,
417+
'got more than 10 bytes'):
418+
self.imap_class(*server.server_address)
419+
420+
def test_simple_with_statement(self):
421+
_, server = self._setup(SimpleIMAPHandler, connect=False)
422+
with self.imap_class(*server.server_address):
423+
pass
424+
425+
def test_with_statement(self):
426+
_, server = self._setup(SimpleIMAPHandler, connect=False)
427+
with self.imap_class(*server.server_address) as imap:
428+
imap.login('user', 'pass')
429+
self.assertEqual(server.logged, 'user')
430+
self.assertIsNone(server.logged)
431+
432+
def test_with_statement_logout(self):
433+
# It is legal to log out explicitly inside the with block
434+
_, server = self._setup(SimpleIMAPHandler, connect=False)
435+
with self.imap_class(*server.server_address) as imap:
436+
imap.login('user', 'pass')
437+
self.assertEqual(server.logged, 'user')
438+
imap.logout()
439+
self.assertIsNone(server.logged)
440+
self.assertIsNone(server.logged)
441+
442+
# command tests
443+
444+
def test_login(self):
445+
client, _ = self._setup(SimpleIMAPHandler)
446+
typ, data = client.login('user', 'pass')
447+
self.assertEqual(typ, 'OK')
448+
self.assertEqual(data[0], b'LOGIN completed')
449+
self.assertEqual(client.state, 'AUTH')
450+
451+
def test_logout(self):
452+
client, _ = self._setup(SimpleIMAPHandler)
453+
typ, data = client.login('user', 'pass')
454+
self.assertEqual(typ, 'OK')
455+
self.assertEqual(data[0], b'LOGIN completed')
456+
typ, data = client.logout()
457+
self.assertEqual(typ, 'BYE')
458+
self.assertEqual(data[0], b'IMAP4ref1 Server logging out')
459+
self.assertEqual(client.state, 'LOGOUT')
460+
461+
def test_lsub(self):
462+
class LsubCmd(SimpleIMAPHandler):
463+
def cmd_LSUB(self, tag, args):
464+
self._send_textline('* LSUB () "." directoryA')
465+
return self._send_tagged(tag, 'OK', 'LSUB completed')
466+
client, _ = self._setup(LsubCmd)
467+
client.login('user', 'pass')
468+
typ, data = client.lsub()
469+
self.assertEqual(typ, 'OK')
470+
self.assertEqual(data[0], b'() "." directoryA')
471+
472+
473+
class NewIMAPTests(NewIMAPTestsMixin, unittest.TestCase):
474+
imap_class = imaplib.IMAP4
475+
server_class = socketserver.TCPServer
476+
477+
478+
@unittest.skipUnless(ssl, "SSL not available")
479+
class NewIMAPSSLTests(NewIMAPTestsMixin, unittest.TestCase):
480+
imap_class = imaplib.IMAP4_SSL
481+
server_class = SecureTCPServer
482+
483+
def test_ssl_raises(self):
484+
ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
485+
ssl_context.verify_mode = ssl.CERT_REQUIRED
486+
ssl_context.check_hostname = True
487+
ssl_context.load_verify_locations(CAFILE)
488+
489+
with self.assertRaisesRegex(ssl.CertificateError,
490+
"hostname '127.0.0.1' doesn't match 'localhost'"):
491+
_, server = self._setup(SimpleIMAPHandler)
492+
client = self.imap_class(*server.server_address,
493+
ssl_context=ssl_context)
494+
client.shutdown()
495+
496+
def test_ssl_verified(self):
497+
ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
498+
ssl_context.verify_mode = ssl.CERT_REQUIRED
499+
ssl_context.check_hostname = True
500+
ssl_context.load_verify_locations(CAFILE)
501+
502+
_, server = self._setup(SimpleIMAPHandler)
503+
client = self.imap_class("localhost", server.server_address[1],
504+
ssl_context=ssl_context)
505+
client.shutdown()
506+
177507
class ThreadedNetworkedTests(unittest.TestCase):
178508
server_class = socketserver.TCPServer
179509
imap_class = imaplib.IMAP4

0 commit comments

Comments
 (0)