cpython/Lib/test/test_imaplib.py

671 lines
24 KiB
Python
Raw Normal View History

from test import support
# If we end up with a significant number of tests that don't require
# threading, this test module should be split. Right now we skip
# them all if we don't have threading.
threading = support.import_module('threading')
from contextlib import contextmanager
import imaplib
import os.path
import socketserver
import time
import calendar
from test.support import (reap_threads, verbose, transient_internet,
run_with_tz, run_with_locale)
Merged revisions 61913,61915-61916,61918-61919,61922-61926,61928-61929,61931,61935,61938,61943 via svnmerge from svn+ssh://pythondev@svn.python.org/python/trunk ........ r61913 | benjamin.peterson | 2008-03-25 22:14:42 +0100 (Tue, 25 Mar 2008) | 2 lines Merged the ACKS from py3k ........ r61915 | thomas.heller | 2008-03-25 22:18:39 +0100 (Tue, 25 Mar 2008) | 1 line Make _ctypes.c PY_SSIZE_T_CLEAN. ........ r61916 | benjamin.peterson | 2008-03-25 22:55:50 +0100 (Tue, 25 Mar 2008) | 3 lines Opps! I merged the revisions, but forgot to add the header to ACKS ........ r61918 | andrew.kuchling | 2008-03-26 01:16:50 +0100 (Wed, 26 Mar 2008) | 1 line Minor docstring typos ........ r61919 | andrew.kuchling | 2008-03-26 01:30:02 +0100 (Wed, 26 Mar 2008) | 1 line Add various items ........ r61922 | neal.norwitz | 2008-03-26 05:55:51 +0100 (Wed, 26 Mar 2008) | 6 lines Try to get this test to be less flaky. It was failing sometimes because the connect would succeed before the timeout occurred. Try using an address and port that hopefully doesn't exist to ensure we get no response. If this doesn't work, we can use a public address close to python.org and hopefully that address never gets taken. ........ r61923 | jerry.seutter | 2008-03-26 06:03:03 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61924 | neal.norwitz | 2008-03-26 06:19:41 +0100 (Wed, 26 Mar 2008) | 5 lines Ensure that the mailbox is closed to prevent problems on Windows with removing an open file. This doesn't seem to be a problem in 2.6, but that appears to be somewhat accidental (specific to reference counting). When this gets merged to 3.0, it will make the 3.0 code simpler. ........ r61925 | jerry.seutter | 2008-03-26 06:32:51 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61926 | jerry.seutter | 2008-03-26 06:58:14 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61928 | georg.brandl | 2008-03-26 10:04:36 +0100 (Wed, 26 Mar 2008) | 2 lines Add Josiah. ........ r61929 | georg.brandl | 2008-03-26 10:32:46 +0100 (Wed, 26 Mar 2008) | 2 lines Add an example for an RFC 822 continuation. ........ r61931 | benjamin.peterson | 2008-03-26 12:57:47 +0100 (Wed, 26 Mar 2008) | 2 lines Added help options to PDB ........ r61935 | christian.heimes | 2008-03-26 13:32:49 +0100 (Wed, 26 Mar 2008) | 1 line Prepare integration of bytearray backport branch ........ r61938 | christian.heimes | 2008-03-26 13:50:43 +0100 (Wed, 26 Mar 2008) | 3 lines Removed merge tracking for "svnmerge" for svn+ssh://pythondev@svn.python.org/python/branches/trunk-bytearray ........ r61943 | georg.brandl | 2008-03-26 13:57:47 +0100 (Wed, 26 Mar 2008) | 2 lines Fix and simplify error handling, silencing a compiler warning. ........
2008-03-26 10:45:42 -03:00
import unittest
from datetime import datetime, timezone, timedelta
try:
import ssl
except ImportError:
ssl = None
CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "keycert3.pem")
CAFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "pycacert.pem")
Merged revisions 61913,61915-61916,61918-61919,61922-61926,61928-61929,61931,61935,61938,61943 via svnmerge from svn+ssh://pythondev@svn.python.org/python/trunk ........ r61913 | benjamin.peterson | 2008-03-25 22:14:42 +0100 (Tue, 25 Mar 2008) | 2 lines Merged the ACKS from py3k ........ r61915 | thomas.heller | 2008-03-25 22:18:39 +0100 (Tue, 25 Mar 2008) | 1 line Make _ctypes.c PY_SSIZE_T_CLEAN. ........ r61916 | benjamin.peterson | 2008-03-25 22:55:50 +0100 (Tue, 25 Mar 2008) | 3 lines Opps! I merged the revisions, but forgot to add the header to ACKS ........ r61918 | andrew.kuchling | 2008-03-26 01:16:50 +0100 (Wed, 26 Mar 2008) | 1 line Minor docstring typos ........ r61919 | andrew.kuchling | 2008-03-26 01:30:02 +0100 (Wed, 26 Mar 2008) | 1 line Add various items ........ r61922 | neal.norwitz | 2008-03-26 05:55:51 +0100 (Wed, 26 Mar 2008) | 6 lines Try to get this test to be less flaky. It was failing sometimes because the connect would succeed before the timeout occurred. Try using an address and port that hopefully doesn't exist to ensure we get no response. If this doesn't work, we can use a public address close to python.org and hopefully that address never gets taken. ........ r61923 | jerry.seutter | 2008-03-26 06:03:03 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61924 | neal.norwitz | 2008-03-26 06:19:41 +0100 (Wed, 26 Mar 2008) | 5 lines Ensure that the mailbox is closed to prevent problems on Windows with removing an open file. This doesn't seem to be a problem in 2.6, but that appears to be somewhat accidental (specific to reference counting). When this gets merged to 3.0, it will make the 3.0 code simpler. ........ r61925 | jerry.seutter | 2008-03-26 06:32:51 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61926 | jerry.seutter | 2008-03-26 06:58:14 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61928 | georg.brandl | 2008-03-26 10:04:36 +0100 (Wed, 26 Mar 2008) | 2 lines Add Josiah. ........ r61929 | georg.brandl | 2008-03-26 10:32:46 +0100 (Wed, 26 Mar 2008) | 2 lines Add an example for an RFC 822 continuation. ........ r61931 | benjamin.peterson | 2008-03-26 12:57:47 +0100 (Wed, 26 Mar 2008) | 2 lines Added help options to PDB ........ r61935 | christian.heimes | 2008-03-26 13:32:49 +0100 (Wed, 26 Mar 2008) | 1 line Prepare integration of bytearray backport branch ........ r61938 | christian.heimes | 2008-03-26 13:50:43 +0100 (Wed, 26 Mar 2008) | 3 lines Removed merge tracking for "svnmerge" for svn+ssh://pythondev@svn.python.org/python/branches/trunk-bytearray ........ r61943 | georg.brandl | 2008-03-26 13:57:47 +0100 (Wed, 26 Mar 2008) | 2 lines Fix and simplify error handling, silencing a compiler warning. ........
2008-03-26 10:45:42 -03:00
class TestImaplib(unittest.TestCase):
def test_Internaldate2tuple(self):
t0 = calendar.timegm((2000, 1, 1, 0, 0, 0, -1, -1, -1))
tt = imaplib.Internaldate2tuple(
b'25 (INTERNALDATE "01-Jan-2000 00:00:00 +0000")')
self.assertEqual(time.mktime(tt), t0)
tt = imaplib.Internaldate2tuple(
b'25 (INTERNALDATE "01-Jan-2000 11:30:00 +1130")')
self.assertEqual(time.mktime(tt), t0)
tt = imaplib.Internaldate2tuple(
b'25 (INTERNALDATE "31-Dec-1999 12:30:00 -1130")')
self.assertEqual(time.mktime(tt), t0)
@run_with_tz('MST+07MDT,M4.1.0,M10.5.0')
def test_Internaldate2tuple_issue10941(self):
self.assertNotEqual(imaplib.Internaldate2tuple(
b'25 (INTERNALDATE "02-Apr-2000 02:30:00 +0000")'),
imaplib.Internaldate2tuple(
b'25 (INTERNALDATE "02-Apr-2000 03:30:00 +0000")'))
def timevalues(self):
return [2000000000, 2000000000.0, time.localtime(2000000000),
(2033, 5, 18, 5, 33, 20, -1, -1, -1),
(2033, 5, 18, 5, 33, 20, -1, -1, 1),
2012-06-22 22:10:50 -03:00
datetime.fromtimestamp(2000000000,
timezone(timedelta(0, 2 * 60 * 60))),
'"18-May-2033 05:33:20 +0200"']
@run_with_locale('LC_ALL', 'de_DE', 'fr_FR')
# DST rules included to work around quirk where the Gnu C library may not
# otherwise restore the previous time zone
@run_with_tz('STD-1DST,M3.2.0,M11.1.0')
def test_Time2Internaldate(self):
expected = '"18-May-2033 05:33:20 +0200"'
for t in self.timevalues():
internal = imaplib.Time2Internaldate(t)
self.assertEqual(internal, expected)
def test_that_Time2Internaldate_returns_a_result(self):
# Without tzset, we can check only that it successfully
# produces a result, not the correctness of the result itself,
# since the result depends on the timezone the machine is in.
for t in self.timevalues():
Merged revisions 61913,61915-61916,61918-61919,61922-61926,61928-61929,61931,61935,61938,61943 via svnmerge from svn+ssh://pythondev@svn.python.org/python/trunk ........ r61913 | benjamin.peterson | 2008-03-25 22:14:42 +0100 (Tue, 25 Mar 2008) | 2 lines Merged the ACKS from py3k ........ r61915 | thomas.heller | 2008-03-25 22:18:39 +0100 (Tue, 25 Mar 2008) | 1 line Make _ctypes.c PY_SSIZE_T_CLEAN. ........ r61916 | benjamin.peterson | 2008-03-25 22:55:50 +0100 (Tue, 25 Mar 2008) | 3 lines Opps! I merged the revisions, but forgot to add the header to ACKS ........ r61918 | andrew.kuchling | 2008-03-26 01:16:50 +0100 (Wed, 26 Mar 2008) | 1 line Minor docstring typos ........ r61919 | andrew.kuchling | 2008-03-26 01:30:02 +0100 (Wed, 26 Mar 2008) | 1 line Add various items ........ r61922 | neal.norwitz | 2008-03-26 05:55:51 +0100 (Wed, 26 Mar 2008) | 6 lines Try to get this test to be less flaky. It was failing sometimes because the connect would succeed before the timeout occurred. Try using an address and port that hopefully doesn't exist to ensure we get no response. If this doesn't work, we can use a public address close to python.org and hopefully that address never gets taken. ........ r61923 | jerry.seutter | 2008-03-26 06:03:03 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61924 | neal.norwitz | 2008-03-26 06:19:41 +0100 (Wed, 26 Mar 2008) | 5 lines Ensure that the mailbox is closed to prevent problems on Windows with removing an open file. This doesn't seem to be a problem in 2.6, but that appears to be somewhat accidental (specific to reference counting). When this gets merged to 3.0, it will make the 3.0 code simpler. ........ r61925 | jerry.seutter | 2008-03-26 06:32:51 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61926 | jerry.seutter | 2008-03-26 06:58:14 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61928 | georg.brandl | 2008-03-26 10:04:36 +0100 (Wed, 26 Mar 2008) | 2 lines Add Josiah. ........ r61929 | georg.brandl | 2008-03-26 10:32:46 +0100 (Wed, 26 Mar 2008) | 2 lines Add an example for an RFC 822 continuation. ........ r61931 | benjamin.peterson | 2008-03-26 12:57:47 +0100 (Wed, 26 Mar 2008) | 2 lines Added help options to PDB ........ r61935 | christian.heimes | 2008-03-26 13:32:49 +0100 (Wed, 26 Mar 2008) | 1 line Prepare integration of bytearray backport branch ........ r61938 | christian.heimes | 2008-03-26 13:50:43 +0100 (Wed, 26 Mar 2008) | 3 lines Removed merge tracking for "svnmerge" for svn+ssh://pythondev@svn.python.org/python/branches/trunk-bytearray ........ r61943 | georg.brandl | 2008-03-26 13:57:47 +0100 (Wed, 26 Mar 2008) | 2 lines Fix and simplify error handling, silencing a compiler warning. ........
2008-03-26 10:45:42 -03:00
imaplib.Time2Internaldate(t)
if ssl:
class SecureTCPServer(socketserver.TCPServer):
def get_request(self):
newsocket, fromaddr = self.socket.accept()
connstream = ssl.wrap_socket(newsocket,
server_side=True,
certfile=CERTFILE)
return connstream, fromaddr
IMAP4_SSL = imaplib.IMAP4_SSL
else:
class SecureTCPServer:
pass
IMAP4_SSL = None
class SimpleIMAPHandler(socketserver.StreamRequestHandler):
timeout = 1
continuation = None
capabilities = ''
def setup(self):
super().setup()
self.server.logged = None
def _send(self, message):
if verbose:
print("SENT: %r" % message.strip())
self.wfile.write(message)
def _send_line(self, message):
self._send(message + b'\r\n')
def _send_textline(self, message):
self._send_line(message.encode('ASCII'))
def _send_tagged(self, tag, code, message):
self._send_textline(' '.join((tag, code, message)))
def handle(self):
# Send a welcome message.
self._send_textline('* OK IMAP4rev1')
while 1:
# Gather up input until we receive a line terminator or we timeout.
# Accumulate read(1) because it's simpler to handle the differences
# between naked sockets and SSL sockets.
line = b''
while 1:
try:
part = self.rfile.read(1)
if part == b'':
# Naked sockets return empty strings..
return
line += part
2012-12-25 10:47:37 -04:00
except OSError:
# ..but SSLSockets raise exceptions.
return
if line.endswith(b'\r\n'):
break
if verbose:
print('GOT: %r' % line.strip())
if self.continuation:
try:
self.continuation.send(line)
except StopIteration:
self.continuation = None
continue
splitline = line.decode('ASCII').split()
tag = splitline[0]
cmd = splitline[1]
args = splitline[2:]
if hasattr(self, 'cmd_' + cmd):
continuation = getattr(self, 'cmd_' + cmd)(tag, args)
if continuation:
self.continuation = continuation
next(continuation)
else:
self._send_tagged(tag, 'BAD', cmd + ' unknown')
def cmd_CAPABILITY(self, tag, args):
caps = ('IMAP4rev1 ' + self.capabilities
if self.capabilities
else 'IMAP4rev1')
self._send_textline('* CAPABILITY ' + caps)
self._send_tagged(tag, 'OK', 'CAPABILITY completed')
def cmd_LOGOUT(self, tag, args):
self.server.logged = None
self._send_textline('* BYE IMAP4ref1 Server logging out')
self._send_tagged(tag, 'OK', 'LOGOUT completed')
def cmd_LOGIN(self, tag, args):
self.server.logged = args[0]
self._send_tagged(tag, 'OK', 'LOGIN completed')
class ThreadedNetworkedTests(unittest.TestCase):
server_class = socketserver.TCPServer
imap_class = imaplib.IMAP4
def make_server(self, addr, hdlr):
class MyServer(self.server_class):
def handle_error(self, request, client_address):
self.close_request(request)
self.server_close()
raise
if verbose:
print("creating server")
server = MyServer(addr, hdlr)
self.assertEqual(server.server_address, server.socket.getsockname())
if verbose:
print("server created")
print("ADDR =", addr)
print("CLASS =", self.server_class)
print("HDLR =", server.RequestHandlerClass)
t = threading.Thread(
name='%s serving' % self.server_class,
target=server.serve_forever,
# Short poll interval to make the test finish quickly.
# Time between requests is short enough that we won't wake
# up spuriously too many times.
kwargs={'poll_interval': 0.01})
t.daemon = True # In case this function raises.
t.start()
if verbose:
print("server running")
return server, t
def reap_server(self, server, thread):
if verbose:
print("waiting for server")
server.shutdown()
server.server_close()
thread.join()
if verbose:
print("done")
@contextmanager
def reaped_server(self, hdlr):
server, thread = self.make_server((support.HOST, 0), hdlr)
try:
yield server
finally:
self.reap_server(server, thread)
@contextmanager
def reaped_pair(self, hdlr):
with self.reaped_server(hdlr) as server:
client = self.imap_class(*server.server_address)
try:
yield server, client
finally:
client.logout()
@reap_threads
def test_connect(self):
with self.reaped_server(SimpleIMAPHandler) as server:
client = self.imap_class(*server.server_address)
client.shutdown()
@reap_threads
def test_bracket_flags(self):
# This violates RFC 3501, which disallows ']' characters in tag names,
# but imaplib has allowed producing such tags forever, other programs
# also produce them (eg: OtherInbox's Organizer app as of 20140716),
# and Gmail, for example, accepts them and produces them. So we
# support them. See issue #21815.
class BracketFlagHandler(SimpleIMAPHandler):
def handle(self):
self.flags = ['Answered', 'Flagged', 'Deleted', 'Seen', 'Draft']
super().handle()
def cmd_AUTHENTICATE(self, tag, args):
self._send_textline('+')
self.server.response = yield
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
def cmd_SELECT(self, tag, args):
flag_msg = ' \\'.join(self.flags)
self._send_line(('* FLAGS (%s)' % flag_msg).encode('ascii'))
self._send_line(b'* 2 EXISTS')
self._send_line(b'* 0 RECENT')
msg = ('* OK [PERMANENTFLAGS %s \\*)] Flags permitted.'
% flag_msg)
self._send_line(msg.encode('ascii'))
self._send_tagged(tag, 'OK', '[READ-WRITE] SELECT completed.')
def cmd_STORE(self, tag, args):
new_flags = args[2].strip('(').strip(')').split()
self.flags.extend(new_flags)
flags_msg = '(FLAGS (%s))' % ' \\'.join(self.flags)
msg = '* %s FETCH %s' % (args[0], flags_msg)
self._send_line(msg.encode('ascii'))
self._send_tagged(tag, 'OK', 'STORE completed.')
with self.reaped_pair(BracketFlagHandler) as (server, client):
code, data = client.authenticate('MYAUTH', lambda x: b'fake')
self.assertEqual(code, 'OK')
self.assertEqual(server.response, b'ZmFrZQ==\r\n')
client.select('test')
typ, [data] = client.store(b'1', "+FLAGS", "[test]")
self.assertIn(b'[test]', data)
client.select('test')
typ, [data] = client.response('PERMANENTFLAGS')
self.assertIn(b'[test]', data)
@reap_threads
def test_issue5949(self):
class EOFHandler(socketserver.StreamRequestHandler):
def handle(self):
# EOF without sending a complete welcome message.
self.wfile.write(b'* OK')
with self.reaped_server(EOFHandler) as server:
self.assertRaises(imaplib.IMAP4.abort,
self.imap_class, *server.server_address)
@reap_threads
def test_line_termination(self):
class BadNewlineHandler(SimpleIMAPHandler):
def cmd_CAPABILITY(self, tag, args):
self._send(b'* CAPABILITY IMAP4rev1 AUTH\n')
self._send_tagged(tag, 'OK', 'CAPABILITY completed')
with self.reaped_server(BadNewlineHandler) as server:
self.assertRaises(imaplib.IMAP4.abort,
self.imap_class, *server.server_address)
class UTF8Server(SimpleIMAPHandler):
capabilities = 'AUTH ENABLE UTF8=ACCEPT'
def cmd_ENABLE(self, tag, args):
self._send_tagged(tag, 'OK', 'ENABLE successful')
def cmd_AUTHENTICATE(self, tag, args):
self._send_textline('+')
self.server.response = yield
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
@reap_threads
def test_enable_raises_error_if_not_AUTH(self):
with self.reaped_pair(self.UTF8Server) as (server, client):
self.assertFalse(client.utf8_enabled)
self.assertRaises(imaplib.IMAP4.error, client.enable, 'foo')
self.assertFalse(client.utf8_enabled)
# XXX Also need a test that enable after SELECT raises an error.
@reap_threads
def test_enable_raises_error_if_no_capability(self):
class NoEnableServer(self.UTF8Server):
capabilities = 'AUTH'
with self.reaped_pair(NoEnableServer) as (server, client):
self.assertRaises(imaplib.IMAP4.error, client.enable, 'foo')
@reap_threads
def test_enable_UTF8_raises_error_if_not_supported(self):
class NonUTF8Server(SimpleIMAPHandler):
pass
with self.assertRaises(imaplib.IMAP4.error):
with self.reaped_pair(NonUTF8Server) as (server, client):
typ, data = client.login('user', 'pass')
self.assertEqual(typ, 'OK')
client.enable('UTF8=ACCEPT')
pass
@reap_threads
def test_enable_UTF8_True_append(self):
class UTF8AppendServer(self.UTF8Server):
def cmd_APPEND(self, tag, args):
self._send_textline('+')
self.server.response = yield
self._send_tagged(tag, 'OK', 'okay')
with self.reaped_pair(UTF8AppendServer) as (server, client):
self.assertEqual(client._encoding, 'ascii')
code, _ = client.authenticate('MYAUTH', lambda x: b'fake')
self.assertEqual(code, 'OK')
self.assertEqual(server.response,
b'ZmFrZQ==\r\n') # b64 encoded 'fake'
code, _ = client.enable('UTF8=ACCEPT')
self.assertEqual(code, 'OK')
self.assertEqual(client._encoding, 'utf-8')
msg_string = 'Subject: üñí©öðé'
typ, data = client.append(
None, None, None, msg_string.encode('utf-8'))
self.assertEqual(typ, 'OK')
self.assertEqual(
server.response,
('UTF8 (%s)\r\n' % msg_string).encode('utf-8')
)
# XXX also need a test that makes sure that the Literal and Untagged_status
# regexes uses unicode in UTF8 mode instead of the default ASCII.
@reap_threads
def test_search_disallows_charset_in_utf8_mode(self):
with self.reaped_pair(self.UTF8Server) as (server, client):
typ, _ = client.authenticate('MYAUTH', lambda x: b'fake')
self.assertEqual(typ, 'OK')
typ, _ = client.enable('UTF8=ACCEPT')
self.assertEqual(typ, 'OK')
self.assertTrue(client.utf8_enabled)
self.assertRaises(imaplib.IMAP4.error, client.search, 'foo', 'bar')
@reap_threads
def test_bad_auth_name(self):
class MyServer(SimpleIMAPHandler):
def cmd_AUTHENTICATE(self, tag, args):
self._send_tagged(tag, 'NO', 'unrecognized authentication '
'type {}'.format(args[0]))
with self.reaped_pair(MyServer) as (server, client):
with self.assertRaises(imaplib.IMAP4.error):
client.authenticate('METHOD', lambda: 1)
@reap_threads
def test_invalid_authentication(self):
class MyServer(SimpleIMAPHandler):
def cmd_AUTHENTICATE(self, tag, args):
self._send_textline('+')
self.response = yield
self._send_tagged(tag, 'NO', '[AUTHENTICATIONFAILED] invalid')
with self.reaped_pair(MyServer) as (server, client):
with self.assertRaises(imaplib.IMAP4.error):
code, data = client.authenticate('MYAUTH', lambda x: b'fake')
@reap_threads
def test_valid_authentication(self):
class MyServer(SimpleIMAPHandler):
def cmd_AUTHENTICATE(self, tag, args):
self._send_textline('+')
self.server.response = yield
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
with self.reaped_pair(MyServer) as (server, client):
code, data = client.authenticate('MYAUTH', lambda x: b'fake')
self.assertEqual(code, 'OK')
self.assertEqual(server.response,
b'ZmFrZQ==\r\n') # b64 encoded 'fake'
with self.reaped_pair(MyServer) as (server, client):
code, data = client.authenticate('MYAUTH', lambda x: 'fake')
self.assertEqual(code, 'OK')
self.assertEqual(server.response,
b'ZmFrZQ==\r\n') # b64 encoded 'fake'
@reap_threads
def test_login_cram_md5(self):
class AuthHandler(SimpleIMAPHandler):
capabilities = 'LOGINDISABLED AUTH=CRAM-MD5'
def cmd_AUTHENTICATE(self, tag, args):
self._send_textline('+ PDE4OTYuNjk3MTcwOTUyQHBvc3RvZmZpY2Uucm'
'VzdG9uLm1jaS5uZXQ=')
r = yield
if (r == b'dGltIGYxY2E2YmU0NjRiOWVmYT'
b'FjY2E2ZmZkNmNmMmQ5ZjMy\r\n'):
self._send_tagged(tag, 'OK', 'CRAM-MD5 successful')
else:
self._send_tagged(tag, 'NO', 'No access')
with self.reaped_pair(AuthHandler) as (server, client):
self.assertTrue('AUTH=CRAM-MD5' in client.capabilities)
ret, data = client.login_cram_md5("tim", "tanstaaftanstaaf")
self.assertEqual(ret, "OK")
with self.reaped_pair(AuthHandler) as (server, client):
self.assertTrue('AUTH=CRAM-MD5' in client.capabilities)
ret, data = client.login_cram_md5("tim", b"tanstaaftanstaaf")
self.assertEqual(ret, "OK")
@reap_threads
def test_aborted_authentication(self):
class MyServer(SimpleIMAPHandler):
def cmd_AUTHENTICATE(self, tag, args):
self._send_textline('+')
self.response = yield
if self.response == b'*\r\n':
self._send_tagged(tag, 'NO', '[AUTHENTICATIONFAILED] aborted')
else:
self._send_tagged(tag, 'OK', 'MYAUTH successful')
with self.reaped_pair(MyServer) as (server, client):
with self.assertRaises(imaplib.IMAP4.error):
code, data = client.authenticate('MYAUTH', lambda x: None)
def test_linetoolong(self):
class TooLongHandler(SimpleIMAPHandler):
def handle(self):
# Send a very long response line
self.wfile.write(b'* OK ' + imaplib._MAXLINE * b'x' + b'\r\n')
with self.reaped_server(TooLongHandler) as server:
self.assertRaises(imaplib.IMAP4.error,
self.imap_class, *server.server_address)
@reap_threads
def test_simple_with_statement(self):
# simplest call
with self.reaped_server(SimpleIMAPHandler) as server:
with self.imap_class(*server.server_address):
pass
@reap_threads
def test_with_statement(self):
with self.reaped_server(SimpleIMAPHandler) as server:
with self.imap_class(*server.server_address) as imap:
imap.login('user', 'pass')
self.assertEqual(server.logged, 'user')
self.assertIsNone(server.logged)
@reap_threads
def test_with_statement_logout(self):
# what happens if already logout in the block?
with self.reaped_server(SimpleIMAPHandler) as server:
with self.imap_class(*server.server_address) as imap:
imap.login('user', 'pass')
self.assertEqual(server.logged, 'user')
imap.logout()
self.assertIsNone(server.logged)
self.assertIsNone(server.logged)
@unittest.skipUnless(ssl, "SSL not available")
class ThreadedNetworkedTestsSSL(ThreadedNetworkedTests):
server_class = SecureTCPServer
imap_class = IMAP4_SSL
@reap_threads
def test_ssl_verified(self):
ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
ssl_context.verify_mode = ssl.CERT_REQUIRED
ssl_context.check_hostname = True
ssl_context.load_verify_locations(CAFILE)
with self.assertRaisesRegex(
ssl.CertificateError,
"hostname '127.0.0.1' doesn't match 'localhost'"):
with self.reaped_server(SimpleIMAPHandler) as server:
client = self.imap_class(*server.server_address,
ssl_context=ssl_context)
client.shutdown()
with self.reaped_server(SimpleIMAPHandler) as server:
client = self.imap_class("localhost", server.server_address[1],
ssl_context=ssl_context)
client.shutdown()
@unittest.skipUnless(
support.is_resource_enabled('network'), 'network resource disabled')
class RemoteIMAPTest(unittest.TestCase):
host = 'cyrus.andrew.cmu.edu'
port = 143
username = 'anonymous'
password = 'pass'
imap_class = imaplib.IMAP4
def setUp(self):
with transient_internet(self.host):
self.server = self.imap_class(self.host, self.port)
def tearDown(self):
if self.server is not None:
with transient_internet(self.host):
self.server.logout()
def test_logincapa(self):
with transient_internet(self.host):
for cap in self.server.capabilities:
self.assertIsInstance(cap, str)
self.assertIn('LOGINDISABLED', self.server.capabilities)
self.assertIn('AUTH=ANONYMOUS', self.server.capabilities)
rs = self.server.login(self.username, self.password)
self.assertEqual(rs[0], 'OK')
def test_logout(self):
with transient_internet(self.host):
rs = self.server.logout()
self.server = None
self.assertEqual(rs[0], 'BYE')
@unittest.skipUnless(ssl, "SSL not available")
@unittest.skipUnless(
support.is_resource_enabled('network'), 'network resource disabled')
class RemoteIMAP_STARTTLSTest(RemoteIMAPTest):
def setUp(self):
super().setUp()
with transient_internet(self.host):
rs = self.server.starttls()
self.assertEqual(rs[0], 'OK')
def test_logincapa(self):
for cap in self.server.capabilities:
self.assertIsInstance(cap, str)
self.assertNotIn('LOGINDISABLED', self.server.capabilities)
@unittest.skipUnless(ssl, "SSL not available")
class RemoteIMAP_SSLTest(RemoteIMAPTest):
port = 993
imap_class = IMAP4_SSL
def setUp(self):
pass
def tearDown(self):
pass
def create_ssl_context(self):
ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
ssl_context.load_cert_chain(CERTFILE)
return ssl_context
def check_logincapa(self, server):
try:
for cap in server.capabilities:
self.assertIsInstance(cap, str)
self.assertNotIn('LOGINDISABLED', server.capabilities)
self.assertIn('AUTH=PLAIN', server.capabilities)
rs = server.login(self.username, self.password)
self.assertEqual(rs[0], 'OK')
finally:
server.logout()
def test_logincapa(self):
with transient_internet(self.host):
_server = self.imap_class(self.host, self.port)
self.check_logincapa(_server)
def test_logincapa_with_client_certfile(self):
with transient_internet(self.host):
_server = self.imap_class(self.host, self.port, certfile=CERTFILE)
self.check_logincapa(_server)
def test_logincapa_with_client_ssl_context(self):
with transient_internet(self.host):
_server = self.imap_class(
self.host, self.port, ssl_context=self.create_ssl_context())
self.check_logincapa(_server)
def test_logout(self):
with transient_internet(self.host):
_server = self.imap_class(self.host, self.port)
rs = _server.logout()
self.assertEqual(rs[0], 'BYE')
def test_ssl_context_certfile_exclusive(self):
with transient_internet(self.host):
self.assertRaises(
ValueError, self.imap_class, self.host, self.port,
certfile=CERTFILE, ssl_context=self.create_ssl_context())
def test_ssl_context_keyfile_exclusive(self):
with transient_internet(self.host):
self.assertRaises(
ValueError, self.imap_class, self.host, self.port,
keyfile=CERTFILE, ssl_context=self.create_ssl_context())
Merged revisions 61913,61915-61916,61918-61919,61922-61926,61928-61929,61931,61935,61938,61943 via svnmerge from svn+ssh://pythondev@svn.python.org/python/trunk ........ r61913 | benjamin.peterson | 2008-03-25 22:14:42 +0100 (Tue, 25 Mar 2008) | 2 lines Merged the ACKS from py3k ........ r61915 | thomas.heller | 2008-03-25 22:18:39 +0100 (Tue, 25 Mar 2008) | 1 line Make _ctypes.c PY_SSIZE_T_CLEAN. ........ r61916 | benjamin.peterson | 2008-03-25 22:55:50 +0100 (Tue, 25 Mar 2008) | 3 lines Opps! I merged the revisions, but forgot to add the header to ACKS ........ r61918 | andrew.kuchling | 2008-03-26 01:16:50 +0100 (Wed, 26 Mar 2008) | 1 line Minor docstring typos ........ r61919 | andrew.kuchling | 2008-03-26 01:30:02 +0100 (Wed, 26 Mar 2008) | 1 line Add various items ........ r61922 | neal.norwitz | 2008-03-26 05:55:51 +0100 (Wed, 26 Mar 2008) | 6 lines Try to get this test to be less flaky. It was failing sometimes because the connect would succeed before the timeout occurred. Try using an address and port that hopefully doesn't exist to ensure we get no response. If this doesn't work, we can use a public address close to python.org and hopefully that address never gets taken. ........ r61923 | jerry.seutter | 2008-03-26 06:03:03 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61924 | neal.norwitz | 2008-03-26 06:19:41 +0100 (Wed, 26 Mar 2008) | 5 lines Ensure that the mailbox is closed to prevent problems on Windows with removing an open file. This doesn't seem to be a problem in 2.6, but that appears to be somewhat accidental (specific to reference counting). When this gets merged to 3.0, it will make the 3.0 code simpler. ........ r61925 | jerry.seutter | 2008-03-26 06:32:51 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61926 | jerry.seutter | 2008-03-26 06:58:14 +0100 (Wed, 26 Mar 2008) | 1 line Changed test so it no longer runs as a side effect of importing. ........ r61928 | georg.brandl | 2008-03-26 10:04:36 +0100 (Wed, 26 Mar 2008) | 2 lines Add Josiah. ........ r61929 | georg.brandl | 2008-03-26 10:32:46 +0100 (Wed, 26 Mar 2008) | 2 lines Add an example for an RFC 822 continuation. ........ r61931 | benjamin.peterson | 2008-03-26 12:57:47 +0100 (Wed, 26 Mar 2008) | 2 lines Added help options to PDB ........ r61935 | christian.heimes | 2008-03-26 13:32:49 +0100 (Wed, 26 Mar 2008) | 1 line Prepare integration of bytearray backport branch ........ r61938 | christian.heimes | 2008-03-26 13:50:43 +0100 (Wed, 26 Mar 2008) | 3 lines Removed merge tracking for "svnmerge" for svn+ssh://pythondev@svn.python.org/python/branches/trunk-bytearray ........ r61943 | georg.brandl | 2008-03-26 13:57:47 +0100 (Wed, 26 Mar 2008) | 2 lines Fix and simplify error handling, silencing a compiler warning. ........
2008-03-26 10:45:42 -03:00
if __name__ == "__main__":
unittest.main()