PublicArchive/cpython
2
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2026-07-07 18:50:03 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
3.15
cpython/Lib/test/test_imaplib.py
T
Miss Islington (bot) f2cd7ef89a [3.15] gh-143921: Narrow the control character check in imaplib commands (GH-153067) (GH-153135) 
Only NUL, CR and LF are rejected now.  Other control characters are
valid in quoted strings and can occur in mailbox names returned by
the server, so they are now accepted and sent quoted.
(cherry picked from commit d0921efb66)

Co-authored-by: Serhiy Storchaka <storchaka@gmail.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-05 15:58:48 +00:00

2247 lines
93 KiB
Python
Raw Permalink Blame History

from test import support
from test.support import socket_helper
from contextlib import contextmanager
import imaplib
import os.path
import socketserver
import time
import calendar
import threading
import re
import select
import socket
from test.support import verbose, run_with_tz, run_with_locale, cpython_only
from test.support import hashlib_helper, threading_helper
import unittest
from unittest import mock
from datetime import datetime, timezone, timedelta
try:
import ssl
except ImportError:
ssl = None
support.requires_working_socket(module=True)
CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "certdata", "keycert3.pem")
CAFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "certdata", "pycacert.pem")
_quoted_string = re.compile(r'"(?:[^"\\]|\\.)*"')
def splitargs(line):
# Split a command line into IMAP arguments: quoted strings, balanced
# (possibly nested) parenthesized lists, and bare atoms.
args = []
i, n = 0, len(line)
while i < n:
if line[i] == ' ':
i += 1
continue
start = i
depth = 0
while i < n and (depth or line[i] != ' '):
c = line[i]
if c == '"':
i = _quoted_string.match(line, i).end()
continue
elif c == '(':
depth += 1
elif c == ')':
depth -= 1
i += 1
args.append(line[start:i])
return args
def parse_seq_number(s, maxmsg):
if s == '*':
return maxmsg
return int(s)
def parse_sequence_set(arg, maxmsg):
for s in arg.split(','):
if ':' in s:
lo, hi = s.split(':')
lo = parse_seq_number(lo, maxmsg)
hi = parse_seq_number(hi, maxmsg)
yield from range(min(lo, hi), max(lo, hi) + 1)
else:
yield parse_seq_number(s, maxmsg)
def make_simple_handler(command, untagged_response=(),
completed=None):
if completed is None:
completed = f'{command} completed'
def cmd(self, tag, args):
for msg in untagged_response:
self._send_textline(msg)
self._send_tagged(tag, 'OK', completed)
cmd.__name__ = 'cmd_' + command
class Handler(SimpleIMAPHandler):
pass
Handler.__name__ = command.title() + 'Handler'
setattr(Handler, cmd.__name__, cmd)
Handler.__qualname__ = Handler.__name__
cmd.__qualname__ = Handler.__qualname__ + '.' + cmd.__name__
return Handler
def _read_literal(handler, marker):
# Read one literal, a raw octet sequence, by its count from the marker
# ('{N}', or '(~{N}' in UTF8 mode).
size = int(re.search(r'\{(\d+)\}', marker).group(1))
# The client must wait for the continuation, so nothing should be readable.
if select.select([handler.connection], [], [], 0)[0]:
raise AssertionError('client sent the literal before the '
'continuation request')
handler._send_textline('+')
return handler.rfile.read(size)
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),
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)
@run_with_tz('STD-1DST,M3.2.0,M11.1.0')
def test_Time2Internaldate_datetime_timetuple(self):
date_time = datetime.fromtimestamp(2000000000).timetuple()
self.assertIsNone(date_time.tm_gmtoff)
self.assertEqual(
imaplib.Time2Internaldate(date_time),
'"18-May-2033 05:33:20 +0200"',
)
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():
imaplib.Time2Internaldate(t)
@socket_helper.skip_if_tcp_blackhole
def test_imap4_host_default_value(self):
# Check whether the IMAP4_PORT is truly unavailable.
with socket.socket() as s:
try:
s.connect(('', imaplib.IMAP4_PORT))
self.skipTest(
"Cannot run the test with local IMAP server running.")
except socket.error:
pass
# This is the exception that should be raised.
expected_errnos = socket_helper.get_socket_conn_refused_errs()
with self.assertRaises(OSError) as cm:
imaplib.IMAP4()
self.assertIn(cm.exception.errno, expected_errnos)
def test_astring(self):
m = imaplib.IMAP4.__new__(imaplib.IMAP4)
m._encoding = 'ascii'
# Plain atoms are left unquoted.
self.assertEqual(m._astring('INBOX'), b'INBOX')
self.assertEqual(m._astring(b'INBOX'), b'INBOX')
# Names with protocol-sensitive characters are quoted.
self.assertEqual(m._astring('New folder'), b'"New folder"')
self.assertEqual(m._astring('a"b'), b'"a\\"b"')
self.assertEqual(m._astring('a\\b'), b'"a\\\\b"')
self.assertEqual(m._astring(''), b'""')
self.assertEqual(m._astring('*'), b'"*"')
# A well-formed quoted string is passed through unchanged.
self.assertEqual(m._astring('"New folder"'), b'"New folder"')
self.assertEqual(m._astring('""'), b'""')
# Including a lenient (non-RFC) backslash escape, which the server
# may accept.
self.assertEqual(m._astring('"a\\b"'), b'"a\\b"')
# A string that only looks quoted but is not a single token is
# quoted as data, closing the argument injection vector.
self.assertEqual(m._astring('"a" SELECT evil "'),
b'"\\"a\\" SELECT evil \\""')
self.assertEqual(m._astring('"'), b'"\\""')
# Non-ASCII names are only allowed in a quoted string or a
# literal, never in an atom (RFC 6855).
m._encoding = 'utf-8'
self.assertEqual(m._astring('Entwürfe'), '"Entwürfe"'.encode())
self.assertEqual(m._astring(b'Entw\xc3\xbcrfe'), b'"Entw\xc3\xbcrfe"')
def test_astring_idempotent(self):
# Quoting an already quoted argument should not change it, so that
# quoting twice gives the same result as quoting once.
m = imaplib.IMAP4.__new__(imaplib.IMAP4)
m._encoding = 'ascii'
for arg in ['INBOX', 'New folder', 'a"b', 'a\\b', '', '*', '%',
'"New folder"', '""', '"a\\b"', '"a" SELECT evil "',
'"', 'a\tb', 'a\rb', '\x7f', '(a)', b'Entw\xc3\xbcrfe']:
with self.subTest(arg=arg):
once = m._astring(arg)
self.assertEqual(m._astring(once), once)
twice = m._list_mailbox(arg)
self.assertEqual(m._list_mailbox(twice), twice)
def test_list_mailbox(self):
m = imaplib.IMAP4.__new__(imaplib.IMAP4)
m._encoding = 'ascii'
# Wildcards are not quoted in a list pattern.
self.assertEqual(m._list_mailbox('*'), b'*')
self.assertEqual(m._list_mailbox('%'), b'%')
self.assertEqual(m._list_mailbox('foo/%'), b'foo/%')
# But spaces still require quoting.
self.assertEqual(m._list_mailbox('New folder'), b'"New folder"')
self.assertEqual(m._list_mailbox('"New folder"'), b'"New folder"')
# As do non-ASCII names; wildcards keep their meaning inside a
# quoted string.
m._encoding = 'utf-8'
self.assertEqual(m._list_mailbox('Entwürfe/%'),
'"Entwürfe/%"'.encode())
if ssl:
class SecureTCPServer(socketserver.TCPServer):
def get_request(self):
newsocket, fromaddr = self.socket.accept()
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(CERTFILE)
connstream = context.wrap_socket(newsocket, server_side=True)
return connstream, fromaddr
IMAP4_SSL = imaplib.IMAP4_SSL
else:
class SecureTCPServer:
pass
IMAP4_SSL = None
class SimpleIMAPHandler(socketserver.StreamRequestHandler):
timeout = support.LOOPBACK_TIMEOUT
continuation = None
capabilities = ''
def setup(self):
super().setup()
self.server.is_selected = None
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)))
welcome = '* OK IMAP4rev1'
def handle(self):
# Send a welcome message.
self._send_textline(self.welcome)
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
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 = splitargs(line.decode().removesuffix('\r\n'))
tag = splitline[0]
cmd = splitline[1]
args = splitline[2:]
self.server.args = args
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
self._send_tagged(tag, 'OK', 'LOGIN completed')
def cmd_SELECT(self, tag, args):
self.server.is_selected = args
self._send_line(b'* 2 EXISTS')
self._send_tagged(tag, 'OK', '[READ-WRITE] SELECT completed.')
def cmd_EXAMINE(self, tag, args):
self.server.is_selected = args
self._send_line(b'* 2 EXISTS')
self._send_tagged(tag, 'OK', '[READ-ONLY] EXAMINE completed.')
def cmd_UNSELECT(self, tag, args):
if self.server.is_selected is not None:
self.server.is_selected = None
self._send_tagged(tag, 'OK', 'Returned to authenticated state. (Success)')
else:
self._send_tagged(tag, 'BAD', 'No mailbox selected')
class IdleCmdDenyHandler(SimpleIMAPHandler):
capabilities = 'IDLE'
def cmd_IDLE(self, tag, args):
self._send_tagged(tag, 'NO', 'IDLE is not allowed at this time')
class IdleCmdHandler(SimpleIMAPHandler):
capabilities = 'IDLE'
def cmd_IDLE(self, tag, args):
# pre-idle-continuation response
self._send_line(b'* 0 EXISTS')
self._send_textline('+ idling')
# simple response
self._send_line(b'* 2 EXISTS')
# complex response: fragmented data due to literal string
self._send_line(b'* 1 FETCH (BODY[HEADER.FIELDS (DATE)] {41}')
self._send(b'Date: Fri, 06 Dec 2024 06:00:00 +0000\r\n\r\n')
self._send_line(b')')
# simple response following a fragmented one
self._send_line(b'* 3 EXISTS')
# response arriving later
time.sleep(1)
self._send_line(b'* 1 RECENT')
r = yield
if r == b'DONE\r\n':
self._send_line(b'* 9 RECENT')
self._send_tagged(tag, 'OK', 'Idle completed')
else:
self._send_tagged(tag, 'BAD', 'Expected DONE')
class IdleCmdDelayedPacketHandler(SimpleIMAPHandler):
capabilities = 'IDLE'
def cmd_IDLE(self, tag, args):
self._send_textline('+ idling')
# response line spanning multiple packets, the last one delayed
self._send(b'* 1 EX')
time.sleep(0.2)
self._send(b'IS')
time.sleep(1)
self._send(b'TS\r\n')
r = yield
if r == b'DONE\r\n':
self._send_tagged(tag, 'OK', 'Idle completed')
else:
self._send_tagged(tag, 'BAD', 'Expected DONE')
class AuthHandler_CRAM_MD5(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')
class NewIMAPTestsMixin:
client = None
def _setup(self, imap_handler, connect=True):
"""
Sets up imap_handler for tests. imap_handler should inherit from either:
- SimpleIMAPHandler - for testing IMAP commands,
- socketserver.StreamRequestHandler - if raw access to stream is needed.
Returns (client, server).
"""
class TestTCPServer(self.server_class):
def handle_error(self, request, client_address):
"""
End request and raise the error if one occurs.
"""
self.close_request(request)
self.server_close()
raise
self.addCleanup(self._cleanup)
self.server = self.server_class((socket_helper.HOST, 0), imap_handler)
self.thread = threading.Thread(
name=self._testMethodName+'-server',
target=self.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})
self.thread.daemon = True # In case this function raises.
self.thread.start()
if connect:
self.client = self.imap_class(*self.server.server_address)
return self.client, self.server
def _cleanup(self):
"""
Cleans up the test server. This method should not be called manually,
it is added to the cleanup queue in the _setup method already.
"""
# if logout was called already we'd raise an exception trying to
# shutdown the client once again
if self.client is not None and self.client.state != 'LOGOUT':
self.client.shutdown()
# cleanup the server
self.server.shutdown()
self.server.server_close()
threading_helper.join_thread(self.thread)
# Explicitly clear the attribute to prevent dangling thread
self.thread = None
def test_EOF_without_complete_welcome_message(self):
# http://bugs.python.org/issue5949
class EOFHandler(socketserver.StreamRequestHandler):
def handle(self):
self.wfile.write(b'* OK')
_, server = self._setup(EOFHandler, connect=False)
self.assertRaises(imaplib.IMAP4.abort, self.imap_class,
*server.server_address)
def test_invalid_greeting(self):
# An invalid greeting, e.g. from a POP3 server on the IMAP port,
# must not fail with "error: None" but report the server's line
# (gh-108280).
class Pop3Handler(socketserver.StreamRequestHandler):
def handle(self):
self.wfile.write(b'+OK POP3 server ready\r\n')
_, server = self._setup(Pop3Handler, connect=False)
with self.assertRaisesRegex(imaplib.IMAP4.error,
r'invalid greeting: \+OK POP3 server ready'):
self.imap_class(*server.server_address)
def test_invalid_greeting_untagged(self):
# An untagged greeting that is neither OK nor PREAUTH (e.g. BYE)
# is reported as is (gh-108280).
class ByeHandler(socketserver.StreamRequestHandler):
def handle(self):
self.wfile.write(b'* BYE Server unavailable\r\n')
_, server = self._setup(ByeHandler, connect=False)
with self.assertRaisesRegex(imaplib.IMAP4.error,
r'invalid greeting: \* BYE Server unavailable'):
self.imap_class(*server.server_address)
def test_invalid_greeting_bare_continuation(self):
# A bare continuation greeting is still reported (gh-108280).
class BareHandler(socketserver.StreamRequestHandler):
def handle(self):
self.wfile.write(b'+\r\n')
_, server = self._setup(BareHandler, connect=False)
with self.assertRaisesRegex(imaplib.IMAP4.error,
r'invalid greeting: \+'):
self.imap_class(*server.server_address)
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')
_, server = self._setup(BadNewlineHandler, connect=False)
self.assertRaises(imaplib.IMAP4.abort, self.imap_class,
*server.server_address)
def test_enable_raises_error_if_not_AUTH(self):
class EnableHandler(SimpleIMAPHandler):
capabilities = 'AUTH ENABLE UTF8=ACCEPT'
client, _ = self._setup(EnableHandler)
self.assertFalse(client.utf8_enabled)
with self.assertRaisesRegex(imaplib.IMAP4.error, 'ENABLE.*NONAUTH'):
client.enable('foo')
self.assertFalse(client.utf8_enabled)
def test_enable_raises_error_if_no_capability(self):
client, _ = self._setup(SimpleIMAPHandler)
with self.assertRaisesRegex(imaplib.IMAP4.error,
'does not support ENABLE'):
client.enable('foo')
def test_enable_UTF8_raises_error_if_not_supported(self):
client, _ = self._setup(SimpleIMAPHandler)
typ, data = client.login('user', 'pass')
self.assertEqual(typ, 'OK')
with self.assertRaisesRegex(imaplib.IMAP4.error,
'does not support ENABLE'):
client.enable('UTF8=ACCEPT')
def test_enable_UTF8_True_append(self):
class UTF8AppendServer(SimpleIMAPHandler):
capabilities = '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')
def cmd_APPEND(self, tag, args):
self.server.response = args
self.server.response.append(_read_literal(self, args[-1]))
literal = yield
self.server.response.append(literal)
self._send_tagged(tag, 'OK', 'okay')
client, server = self._setup(UTF8AppendServer)
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')
self.assertEqual(server.args, ['UTF8=ACCEPT'])
msg_string = 'Subject: üñí©öðé'
typ, data = client.append(
None, None, None, (msg_string + '\n').encode('utf-8'))
self.assertEqual(typ, 'OK')
self.assertEqual(server.response,
['INBOX', 'UTF8',
'(~{25}', ('%s\r\n' % msg_string).encode('utf-8'),
b')\r\n' ])
def test_search_disallows_charset_in_utf8_mode(self):
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')
client, _ = self._setup(UTF8Server)
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)
with self.assertRaisesRegex(imaplib.IMAP4.error, 'charset.*UTF8'):
client.search('foo', 'bar')
def test_utf8_mailbox_name(self):
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')
client, server = self._setup(UTF8Server)
typ, _ = client.authenticate('MYAUTH', lambda x: b'fake')
self.assertEqual(typ, 'OK')
typ, _ = client.enable('UTF8=ACCEPT')
self.assertEqual(typ, 'OK')
# A non-ASCII mailbox name is only allowed in a quoted string
# or a literal, never in an atom (RFC 6855).
typ, _ = client.select('Entwürfe')
self.assertEqual(typ, 'OK')
self.assertEqual(server.is_selected, ['"Entwürfe"'])
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]))
client, _ = self._setup(MyServer)
with self.assertRaisesRegex(imaplib.IMAP4.error,
'unrecognized authentication type METHOD'):
client.authenticate('METHOD', lambda: 1)
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')
client, _ = self._setup(MyServer)
with self.assertRaisesRegex(imaplib.IMAP4.error,
r'\[AUTHENTICATIONFAILED\] invalid'):
client.authenticate('MYAUTH', lambda x: b'fake')
def test_invalid_login(self):
class MyServer(SimpleIMAPHandler):
def cmd_LOGIN(self, tag, args):
self.server.logged = args[0]
self._send_tagged(tag, 'NO', '[LOGIN] failed')
client, _ = self._setup(MyServer)
with self.assertRaisesRegex(imaplib.IMAP4.error,
r'\[LOGIN\] failed'):
client.login('user', 'wrongpass')
def test_valid_authentication_bytes(self):
class MyServer(SimpleIMAPHandler):
def cmd_AUTHENTICATE(self, tag, args):
self._send_textline('+')
self.server.response = yield
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
client, server = self._setup(MyServer)
code, _ = client.authenticate('MYAUTH', lambda x: b'fake')
self.assertEqual(code, 'OK')
self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake'
def test_valid_authentication_plain_text(self):
class MyServer(SimpleIMAPHandler):
def cmd_AUTHENTICATE(self, tag, args):
self._send_textline('+')
self.server.response = yield
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
client, server = self._setup(MyServer)
code, _ = client.authenticate('MYAUTH', lambda x: 'fake')
self.assertEqual(code, 'OK')
self.assertEqual(server.response, b'ZmFrZQ==\r\n') # b64 encoded 'fake'
@hashlib_helper.requires_hashdigest('md5', openssl=True)
def test_login_cram_md5_bytes(self):
client, _ = self._setup(AuthHandler_CRAM_MD5)
self.assertIn('AUTH=CRAM-MD5', client.capabilities)
ret, _ = client.login_cram_md5("tim", b"tanstaaftanstaaf")
self.assertEqual(ret, "OK")
@hashlib_helper.requires_hashdigest('md5', openssl=True)
def test_login_cram_md5_plain_text(self):
client, _ = self._setup(AuthHandler_CRAM_MD5)
self.assertIn('AUTH=CRAM-MD5', client.capabilities)
ret, _ = client.login_cram_md5("tim", "tanstaaftanstaaf")
self.assertEqual(ret, "OK")
@hashlib_helper.block_algorithm("md5")
def test_login_cram_md5_blocked(self):
client, _ = self._setup(AuthHandler_CRAM_MD5)
self.assertIn('AUTH=CRAM-MD5', client.capabilities)
msg = re.escape("CRAM-MD5 authentication is not supported")
with self.assertRaisesRegex(imaplib.IMAP4.error, msg):
client.login_cram_md5("tim", b"tanstaaftanstaaf")
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')
client, _ = self._setup(MyServer)
with self.assertRaisesRegex(imaplib.IMAP4.error,
r'\[AUTHENTICATIONFAILED\] aborted'):
client.authenticate('MYAUTH', lambda x: None)
@mock.patch('imaplib._MAXLINE', 10)
def test_linetoolong(self):
class TooLongHandler(SimpleIMAPHandler):
def handle(self):
# send response line longer than the limit set in the next line
self.wfile.write(b'* OK ' + 11 * b'x' + b'\r\n')
_, server = self._setup(TooLongHandler, connect=False)
with self.assertRaisesRegex(imaplib.IMAP4.error,
'got more than 10 bytes'):
self.imap_class(*server.server_address)
def test_simple_with_statement(self):
_, server = self._setup(SimpleIMAPHandler, connect=False)
with self.imap_class(*server.server_address):
pass
def test_imaplib_timeout_test(self):
_, server = self._setup(SimpleIMAPHandler, connect=False)
with self.imap_class(*server.server_address, timeout=None) as client:
self.assertEqual(client.sock.timeout, None)
with self.imap_class(*server.server_address, timeout=support.LOOPBACK_TIMEOUT) as client:
self.assertEqual(client.sock.timeout, support.LOOPBACK_TIMEOUT)
with self.assertRaises(ValueError):
self.imap_class(*server.server_address, timeout=0)
def test_imaplib_timeout_functionality_test(self):
class TimeoutHandler(SimpleIMAPHandler):
def handle(self):
time.sleep(1)
SimpleIMAPHandler.handle(self)
_, server = self._setup(TimeoutHandler)
addr = server.server_address[1]
with self.assertRaises(TimeoutError):
client = self.imap_class("localhost", addr, timeout=0.001)
def test_with_statement(self):
_, server = self._setup(SimpleIMAPHandler, connect=False)
with self.imap_class(*server.server_address) as imap:
imap.login('user', 'pass')
self.assertEqual(server.logged, ['user', '"pass"'])
self.assertIsNone(server.logged)
def test_with_statement_logout(self):
# It is legal to log out explicitly inside the with block
_, server = self._setup(SimpleIMAPHandler, connect=False)
with self.imap_class(*server.server_address) as imap:
imap.login('user', 'pass')
self.assertEqual(server.logged, ['user', '"pass"'])
imap.logout()
self.assertIsNone(server.logged)
self.assertIsNone(server.logged)
# command tests
def test_idle_capability(self):
client, _ = self._setup(SimpleIMAPHandler)
with self.assertRaisesRegex(imaplib.IMAP4.error,
'does not support IMAP4 IDLE'):
with client.idle():
pass
def test_idle_denied(self):
client, _ = self._setup(IdleCmdDenyHandler)
client.login('user', 'pass')
with self.assertRaises(imaplib.IMAP4.error):
with client.idle() as idler:
pass
def test_idle_iter(self):
client, _ = self._setup(IdleCmdHandler)
client.login('user', 'pass')
with client.idle() as idler:
# iteration should include response between 'IDLE' & '+ idling'
response = next(idler)
self.assertEqual(response, ('EXISTS', [b'0']))
# iteration should produce responses
response = next(idler)
self.assertEqual(response, ('EXISTS', [b'2']))
# fragmented response (with literal string) should arrive whole
expected_fetch_data = [
(b'1 (BODY[HEADER.FIELDS (DATE)] {41}',
b'Date: Fri, 06 Dec 2024 06:00:00 +0000\r\n\r\n'),
b')']
typ, data = next(idler)
self.assertEqual(typ, 'FETCH')
self.assertEqual(data, expected_fetch_data)
# response after a fragmented one should arrive separately
response = next(idler)
self.assertEqual(response, ('EXISTS', [b'3']))
# iteration should have consumed untagged responses
_, data = client.response('EXISTS')
self.assertEqual(data, [None])
# responses not iterated should be available after idle
_, data = client.response('RECENT')
self.assertEqual(data[0], b'1')
# responses received after 'DONE' should be available after idle
self.assertEqual(data[1], b'9')
def test_idle_burst(self):
client, _ = self._setup(IdleCmdHandler)
client.login('user', 'pass')
# burst() should yield immediately available responses
with client.idle() as idler:
batch = list(idler.burst())
self.assertEqual(len(batch), 4)
# burst() should not have consumed later responses
_, data = client.response('RECENT')
self.assertEqual(data, [b'1', b'9'])
def test_idle_delayed_packet(self):
client, _ = self._setup(IdleCmdDelayedPacketHandler)
client.login('user', 'pass')
# If our readline() implementation fails to preserve line fragments
# when idle timeouts trigger, a response spanning delayed packets
# can be corrupted, leaving the protocol stream in a bad state.
try:
with client.idle(0.5) as idler:
self.assertRaises(StopIteration, next, idler)
except client.abort as err:
self.fail('multi-packet response was corrupted by idle timeout')
def test_login(self):
client, server = self._setup(SimpleIMAPHandler)
typ, data = client.login('user', 'pass')
self.assertEqual(typ, 'OK')
self.assertEqual(data[0], b'LOGIN completed')
self.assertEqual(client.state, 'AUTH')
# The user name is quoted only when necessary, but the password
# is always quoted.
self.assertEqual(server.logged, ['user', '"pass"'])
self.assertRaises(imaplib.IMAP4.error, client.login, 'user', 'pass')
def test_login_quoted(self):
client, server = self._setup(SimpleIMAPHandler)
typ, data = client.login('us*r', 'p%ss')
self.assertEqual(typ, 'OK')
self.assertEqual(data[0], b'LOGIN completed')
self.assertEqual(client.state, 'AUTH')
self.assertEqual(server.logged, ['"us*r"', '"p%ss"'])
def test_login_quoted2(self):
# An already quoted user name is passed through unchanged, rather
# than being quoted a second time; the password is always quoted.
client, server = self._setup(SimpleIMAPHandler)
typ, data = client.login('"user"', '"pass"')
self.assertEqual(typ, 'OK')
self.assertEqual(data[0], b'LOGIN completed')
self.assertEqual(client.state, 'AUTH')
self.assertEqual(server.logged, ['"user"', r'"\"pass\""'])
def test_append_line_endings(self):
# append() normalizes bare CR and LF in the message to CRLF.
class AppendHandler(SimpleIMAPHandler):
def cmd_APPEND(self, tag, args):
self.server.response = _read_literal(self, args[-1])
yield # read the trailer line
self._send_tagged(tag, 'OK', 'APPEND completed')
client, server = self._setup(AppendHandler)
client.login('user', 'pass')
message = b'a\rb\nc\r\nd'
client.append('INBOX', None, None, message)
self.assertEqual(server.response, b'a\r\nb\r\nc\r\nd')
# The mailbox is quoted and the flags are wrapped in parentheses
# when necessary.
client.append('New folder', r'\Seen', None, b'data')
self.assertEqual(server.args, ['"New folder"', r'(\Seen)', '{4}'])
def test_login_capabilities(self):
# A server may advertise new capabilities after login (as an
# untagged CAPABILITY response); imaplib must refresh its cached
# capability list (gh-63121, gh-103451).
class CapabilityLoginHandler(SimpleIMAPHandler):
def cmd_LOGIN(self, tag, args):
self.server.logged = args[0]
self._send_textline('* CAPABILITY IMAP4rev1 ENABLE UTF8=ACCEPT')
self._send_tagged(tag, 'OK', 'LOGIN completed')
def cmd_ENABLE(self, tag, args):
self._send_tagged(tag, 'OK', 'ENABLE completed')
client, _ = self._setup(CapabilityLoginHandler)
self.assertNotIn('ENABLE', client.capabilities)
client.login('user', 'pass')
self.assertIn('ENABLE', client.capabilities)
self.assertIn('UTF8=ACCEPT', client.capabilities)
typ, _ = client.enable('UTF8=ACCEPT')
self.assertEqual(typ, 'OK')
def test_authenticate_capabilities(self):
# Capabilities are also refreshed after AUTHENTICATE, here from a
# CAPABILITY response code in the tagged OK response.
class CapabilityAuthHandler(SimpleIMAPHandler):
def cmd_AUTHENTICATE(self, tag, args):
self._send_textline('+')
self.server.response = yield
self._send_tagged(
tag, 'OK',
'[CAPABILITY IMAP4rev1 ENABLE] AUTHENTICATE completed')
client, _ = self._setup(CapabilityAuthHandler)
self.assertNotIn('ENABLE', client.capabilities)
client.authenticate('MYAUTH', lambda x: b'fake')
self.assertIn('ENABLE', client.capabilities)
def test_greeting_capabilities(self):
# Capabilities advertised in the greeting are used directly,
# without sending a separate CAPABILITY command.
class GreetingHandler(SimpleIMAPHandler):
welcome = '* OK [CAPABILITY IMAP4rev1 ENABLE] Server ready'
def cmd_CAPABILITY(self, tag, args):
self.server.capability_queried = True
super().cmd_CAPABILITY(tag, args)
client, server = self._setup(GreetingHandler)
self.assertEqual(client.capabilities, ('IMAP4REV1', 'ENABLE'))
self.assertFalse(getattr(server, 'capability_queried', False))
def test_login_requery_capabilities(self):
# If the server does not advertise capabilities after login,
# imaplib re-queries them (as it does after STARTTLS), so a
# capability that becomes available only after authentication is
# still recognized (gh-63121).
class RequeryHandler(SimpleIMAPHandler):
def cmd_CAPABILITY(self, tag, args):
caps = 'IMAP4rev1 ENABLE' if self.server.logged else 'IMAP4rev1'
self._send_textline('* CAPABILITY ' + caps)
self._send_tagged(tag, 'OK', 'CAPABILITY completed')
client, _ = self._setup(RequeryHandler)
self.assertNotIn('ENABLE', client.capabilities)
client.login('user', 'pass')
self.assertIn('ENABLE', client.capabilities)
def test_logout(self):
client, _ = self._setup(SimpleIMAPHandler)
typ, data = client.login('user', 'pass')
self.assertEqual(typ, 'OK')
self.assertEqual(data[0], b'LOGIN completed')
typ, data = client.logout()
self.assertEqual(typ, 'BYE', (typ, data))
self.assertEqual(data[0], b'IMAP4ref1 Server logging out', (typ, data))
self.assertEqual(client.state, 'LOGOUT')
def test_lsub(self):
client, server = self._setup(make_simple_handler('LSUB',
['* LSUB () "." directoryA', '* LSUB () "." directoryB']))
client.login('user', 'pass')
typ, data = client.lsub()
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'() "." directoryA', b'() "." directoryB'])
self.assertEqual(server.args, ['""', '*'])
typ, data = client.lsub('~/Mail/', '%')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['~/Mail/', '%'])
# The directory is quoted when necessary; wildcards in the pattern
# are preserved.
typ, data = client.lsub('New folder', '%')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"', '%'])
def test_extra_blank_line_after_literal(self):
# Some buggy servers send an extra blank line after the counted
# literal data. imaplib should skip it instead of failing.
class BlankLineHandler(SimpleIMAPHandler):
def cmd_FETCH(self, tag, args):
self._send(b'* 1 FETCH (BODY[HEADER] {13}\r\n')
self._send(b'Subject: test') # 13-byte literal
self._send(b'\r\n)\r\n') # stray blank line, then ')'
self._send_tagged(tag, 'OK', 'FETCH completed')
client, _ = self._setup(BlankLineHandler)
client.login('user', 'pass')
client.select()
typ, data = client.fetch('1', '(BODY[HEADER])')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [(b'1 (BODY[HEADER] {13}', b'Subject: test'),
b')'])
def test_unselect(self):
client, server = self._setup(SimpleIMAPHandler)
client.login('user', 'pass')
typ, data = client.select()
self.assertEqual(typ, 'OK')
self.assertEqual(data[0], b'2')
typ, data = client.unselect()
self.assertEqual(typ, 'OK')
self.assertEqual(data[0], b'Returned to authenticated state. (Success)')
self.assertEqual(client.state, 'AUTH')
self.assertIsNone(server.is_selected)
def test_enable(self):
class EnableHandler(SimpleIMAPHandler):
capabilities = 'IMAP4rev1 ID LITERAL+ ENABLE X-GOOD-IDEA'
def cmd_ENABLE(self, tag, args):
capabilities = self.capabilities.split()
for arg in args:
if arg in capabilities:
self._send_textline('* ENABLED ' + arg)
self._send_tagged(tag, 'OK', 'foo')
client, server = self._setup(EnableHandler)
client.login('user', 'pass')
code, data = client.enable('CONDSTORE X-GOOD-IDEA')
self.assertEqual(code, 'OK')
self.assertEqual(data, [b'foo'])
self.assertEqual(server.args, ['CONDSTORE', 'X-GOOD-IDEA'])
def test_select(self):
client, server = self._setup(SimpleIMAPHandler)
client.login('user', 'pass')
typ, data = client.select()
self.assertEqual(typ, 'OK')
self.assertEqual(data[0], b'2')
self.assertEqual(server.is_selected, ['INBOX'])
typ, data = client.select('Archive')
self.assertEqual(typ, 'OK')
self.assertEqual(server.is_selected, ['Archive'])
# readonly=True issues EXAMINE instead of SELECT
# (there is no separate examine() method).
typ, data = client.select(readonly=True)
self.assertEqual(typ, 'OK')
self.assertEqual(data[0], b'2')
self.assertEqual(server.is_selected, ['INBOX'])
self.assertTrue(client.is_readonly)
typ, data = client.select('New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(server.is_selected, ['"New folder"'])
def test_expunge(self):
client, server = self._setup(make_simple_handler('EXPUNGE',
['* 3 EXPUNGE', '* 3 EXPUNGE', '* 5 EXPUNGE', '* 8 EXPUNGE']))
client.login('user', 'pass')
client.select()
typ, data = client.expunge()
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'3', b'3', b'5', b'8'])
def test_close(self):
client, server = self._setup(make_simple_handler('CLOSE'))
client.login('user', 'pass')
client.select()
typ, data = client.close()
self.assertEqual(typ, 'OK')
self.assertEqual(client.state, 'AUTH')
def test_check(self):
client, server = self._setup(make_simple_handler('CHECK'))
client.login('user', 'pass')
client.select()
typ, data = client.check()
self.assertEqual(typ, 'OK')
def test_noop(self):
client, server = self._setup(make_simple_handler('NOOP',
['* 4 EXISTS']))
client.login('user', 'pass')
typ, data = client.noop()
self.assertEqual(typ, 'OK')
# NOOP is used to pick up server-pushed untagged responses.
self.assertEqual(client.untagged_responses['EXISTS'], [b'4'])
def test_namespace(self):
client, server = self._setup(make_simple_handler('NAMESPACE',
['* NAMESPACE (("" "/")) NIL NIL']))
client.login('user', 'pass')
typ, data = client.namespace()
self.assertEqual(typ, 'OK')
self.assertEqual(data[0], b'(("" "/")) NIL NIL')
def test_capability(self):
client, server = self._setup(make_simple_handler('CAPABILITY',
['* CAPABILITY IMAP4rev1 IDLE NAMESPACE']))
client.login('user', 'pass')
typ, data = client.capability()
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'IMAP4rev1 IDLE NAMESPACE'])
def test_recent(self):
# recent() prods the server with NOOP if no RECENT was seen yet.
client, server = self._setup(make_simple_handler('NOOP',
['* 5 RECENT']))
client.login('user', 'pass')
typ, data = client.recent()
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'5'])
def test_response(self):
client, server = self._setup(SimpleIMAPHandler)
client.login('user', 'pass')
client.select() # the handler answers with '* 2 EXISTS'
typ, data = client.response('EXISTS')
self.assertEqual(typ, 'EXISTS')
self.assertEqual(data, [b'2'])
# The value is cleared once read.
typ, data = client.response('EXISTS')
self.assertEqual(data, [None])
def test_create(self):
client, server = self._setup(make_simple_handler('CREATE'))
client.login('user', 'pass')
typ, data = client.create('owatagusiam/')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'CREATE completed'])
self.assertEqual(server.args, ['owatagusiam/'])
typ, data = client.create('owatagusiam/blurdybloop')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'CREATE completed'])
self.assertEqual(server.args, ['owatagusiam/blurdybloop'])
typ, data = client.create('New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'CREATE completed'])
self.assertEqual(server.args, ['"New folder"'])
def test_copy(self):
client, server = self._setup(make_simple_handler('COPY'))
client.login('user', 'pass')
client.select()
typ, data = client.copy('2:4', 'MEETING')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'COPY completed'])
self.assertEqual(server.args, ['2:4', 'MEETING'])
typ, data = client.copy('2:4', 'New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'COPY completed'])
self.assertEqual(server.args, ['2:4', '"New folder"'])
def test_uid_copy(self):
client, server = self._setup(make_simple_handler('UID',
completed='UID COPY completed'))
client.login('user', 'pass')
client.select()
typ, data = client.uid('copy', '4827313:4828442', 'MEETING')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [None])
self.assertEqual(server.args, ['COPY', '4827313:4828442', 'MEETING'])
typ, data = client.uid('copy', '4827313:4828442', 'New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [None])
self.assertEqual(server.args, ['COPY', '4827313:4828442', '"New folder"'])
def test_store(self):
client, server = self._setup(make_simple_handler('STORE', [
r'* 2 FETCH (FLAGS (\Deleted \Seen))',
r'* 3 FETCH (FLAGS (\Deleted))',
r'* 4 FETCH (FLAGS (\Deleted \Flagged \Seen))',
]))
client.login('user', 'pass')
client.select()
typ, data = client.store('2:4', '+FLAGS', r'(\Deleted)')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [
br'2 (FLAGS (\Deleted \Seen))',
br'3 (FLAGS (\Deleted))',
br'4 (FLAGS (\Deleted \Flagged \Seen))',
])
self.assertEqual(server.args, ['2:4', '+FLAGS', r'(\Deleted)'])
typ, data = client.store('2:4', '+FLAGS', r'\Deleted')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['2:4', '+FLAGS', r'(\Deleted)'])
def test_uid_store(self):
client, server = self._setup(make_simple_handler('UID', [
r'* 23 FETCH (FLAGS (\Deleted \Seen) UID 4827313)',
r'* 24 FETCH (FLAGS (\Deleted) UID 4827943)',
r'* 25 FETCH (FLAGS (\Deleted \Flagged \Seen) UID 4828442)',
], 'UID STORE completed'))
client.login('user', 'pass')
client.select()
typ, data = client.uid('store', '4827313:4828442', '+FLAGS', r'(\Deleted)')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [
br'23 (FLAGS (\Deleted \Seen) UID 4827313)',
br'24 (FLAGS (\Deleted) UID 4827943)',
br'25 (FLAGS (\Deleted \Flagged \Seen) UID 4828442)',
])
self.assertEqual(server.args, ['STORE', '4827313:4828442', '+FLAGS', r'(\Deleted)'])
typ, data = client.uid('store', '4827313:4828442', '+FLAGS', r'\Deleted')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['STORE', '4827313:4828442', '+FLAGS', r'(\Deleted)'])
def test_fetch(self):
# The handler expands the requested sequence set and answers for
# exactly those messages, so the test exercises the round trip of
# the message set, not just a canned reply.
class FetchHandler(SimpleIMAPHandler):
messages = 4
def cmd_SELECT(self, tag, args):
self.server.is_selected = args
self._send_line(b'* %d EXISTS' % self.messages)
self._send_tagged(tag, 'OK', '[READ-WRITE] SELECT completed.')
def cmd_FETCH(self, tag, args):
for n in parse_sequence_set(args[0], self.messages):
self._send_textline(r'* %d FETCH (FLAGS (\Seen))' % n)
self._send_tagged(tag, 'OK', 'FETCH completed')
client, server = self._setup(FetchHandler)
client.login('user', 'pass')
client.select()
typ, data = client.fetch('2:4', '(FLAGS)')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [
br'2 (FLAGS (\Seen))',
br'3 (FLAGS (\Seen))',
br'4 (FLAGS (\Seen))',
])
self.assertEqual(server.args, ['2:4', '(FLAGS)'])
# A comma-separated set with an open range up to '*'.
typ, data = client.fetch('1,3:*', '(FLAGS)')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [
br'1 (FLAGS (\Seen))',
br'3 (FLAGS (\Seen))',
br'4 (FLAGS (\Seen))',
])
self.assertEqual(server.args, ['1,3:*', '(FLAGS)'])
# An item with nested parentheses is sent (and parsed) as a
# single argument.
typ, data = client.fetch('1', '(BODY[HEADER.FIELDS (DATE FROM)])')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [br'1 (FLAGS (\Seen))'])
self.assertEqual(server.args, ['1', '(BODY[HEADER.FIELDS (DATE FROM)])'])
# message_parts is wrapped in parentheses if it is not already.
typ, data = client.fetch('2:4', 'FLAGS')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['2:4', '(FLAGS)'])
# But the macros are not, as they are not data item names.
typ, data = client.fetch('2:4', 'ALL')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['2:4', 'ALL'])
typ, data = client.fetch('2:4', 'fast')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['2:4', 'fast'])
def test_uid_fetch(self):
client, server = self._setup(make_simple_handler('UID', [
r'* 23 FETCH (FLAGS (\Seen) UID 4827313)',
r'* 24 FETCH (FLAGS (\Seen) UID 4827943)',
r'* 25 FETCH (FLAGS (\Seen) UID 4828442)',
], 'UID FETCH completed'))
client.login('user', 'pass')
client.select()
typ, data = client.uid('fetch', '4827313:4828442', '(FLAGS)')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [
br'23 (FLAGS (\Seen) UID 4827313)',
br'24 (FLAGS (\Seen) UID 4827943)',
br'25 (FLAGS (\Seen) UID 4828442)',
])
self.assertEqual(server.args, ['FETCH', '4827313:4828442', '(FLAGS)'])
typ, data = client.uid('fetch', '4827313:4828442', 'FLAGS')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['FETCH', '4827313:4828442', '(FLAGS)'])
typ, data = client.uid('fetch', '4827313:4828442', 'ALL')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['FETCH', '4827313:4828442', 'ALL'])
def test_partial(self):
client, server = self._setup(make_simple_handler('PARTIAL',
['* 1 FETCH (RFC822.TEXT<0.10> "0123456789")']))
client.login('user', 'pass')
client.select()
typ, data = client.partial('1', 'RFC822.TEXT', '0', '10')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'1 (RFC822.TEXT<0.10> "0123456789")'])
self.assertEqual(server.args, ['1', 'RFC822.TEXT', '0', '10'])
def test_search(self):
response = []
client, server = self._setup(make_simple_handler('SEARCH', response))
client.login('user', 'pass')
client.select()
response[:] = ['* SEARCH 2 84 882']
typ, data = client.search(None, 'FLAGGED', 'SINCE', '1-Feb-1994', 'NOT', 'FROM', '"Smith"')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'2 84 882'])
self.assertEqual(server.args, ['FLAGGED', 'SINCE', '1-Feb-1994', 'NOT', 'FROM', '"Smith"'])
response[:] = ['* SEARCH']
typ, data = client.search(None, 'TEXT', '"string not in mailbox"')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b''])
self.assertEqual(server.args, ['TEXT', '"string not in mailbox"'])
response[:] = ['* SEARCH 43']
typ, data = client.search('UTF-8', 'TEXT', 'XXXXXX')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'43'])
self.assertEqual(server.args, ['CHARSET', 'UTF-8', 'TEXT', 'XXXXXX'])
typ, data = client.search('NF_Z_62-010_(1973)', 'TEXT', 'XXXXXX')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['CHARSET', '"NF_Z_62-010_(1973)"', 'TEXT', 'XXXXXX'])
def test_uid_search(self):
response = []
client, server = self._setup(make_simple_handler('UID', response,
'UID SEARCH completed'))
client.login('user', 'pass')
client.select()
response[:] = ['* SEARCH 2 84 882']
typ, data = client.uid('SEARCH', 'FLAGGED', 'SINCE', '1-Feb-1994', 'NOT', 'FROM', '"Smith"')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'2 84 882'])
self.assertEqual(server.args, ['SEARCH', 'FLAGGED', 'SINCE', '1-Feb-1994', 'NOT', 'FROM', '"Smith"'])
response[:] = ['* SEARCH']
typ, data = client.uid('SEARCH', 'TEXT', '"string not in mailbox"')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b''])
self.assertEqual(server.args, ['SEARCH', 'TEXT', '"string not in mailbox"'])
response[:] = ['* SEARCH 43']
typ, data = client.uid('SEARCH', 'CHARSET', 'UTF-8', 'TEXT', 'XXXXXX')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'43'])
self.assertEqual(server.args, ['SEARCH', 'CHARSET', 'UTF-8', 'TEXT', 'XXXXXX'])
typ, data = client.uid('SEARCH', 'CHARSET', '"NF_Z_62-010_(1973)"', 'TEXT', 'XXXXXX')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['SEARCH', 'CHARSET', '"NF_Z_62-010_(1973)"', 'TEXT', 'XXXXXX'])
def test_sort(self):
response = []
client, server = self._setup(make_simple_handler('SORT', response))
client.login('user', 'pass')
client.select()
response[:] = ['* SORT 2 84 882']
typ, data = client.sort('(SUBJECT)', 'UTF-8', 'SINCE', '1-Feb-1994')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [br'2 84 882'])
self.assertEqual(server.args, ['(SUBJECT)', 'UTF-8', 'SINCE', '1-Feb-1994'])
response[:] = ['* SORT 5 3 4 1 2']
typ, data = client.sort('(SUBJECT REVERSE DATE)', 'UTF-8', 'ALL')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [br'5 3 4 1 2'])
self.assertEqual(server.args, ['(SUBJECT REVERSE DATE)', 'UTF-8', 'ALL'])
response[:] = ['* SORT']
typ, data = client.sort('(SUBJECT)', 'US-ASCII', 'TEXT', '"not in mailbox"')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [br''])
self.assertEqual(server.args, ['(SUBJECT)', 'US-ASCII', 'TEXT', '"not in mailbox"'])
typ, data = client.sort('SUBJECT', 'NF_Z_62-010_(1973)', 'TEXT', '"not in mailbox"')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['(SUBJECT)', '"NF_Z_62-010_(1973)"', 'TEXT', '"not in mailbox"'])
def test_uid_sort(self):
response = []
client, server = self._setup(make_simple_handler('UID', response,
'UID SORT completed'))
client.login('user', 'pass')
client.select()
response[:] = ['* SORT 2 84 882']
typ, data = client.uid('sort', '(SUBJECT)', 'UTF-8', 'SINCE', '1-Feb-1994')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [br'2 84 882'])
self.assertEqual(server.args, ['SORT', '(SUBJECT)', 'UTF-8', 'SINCE', '1-Feb-1994'])
response[:] = ['* SORT 5 3 4 1 2']
typ, data = client.uid('sort', '(SUBJECT REVERSE DATE)', 'UTF-8', 'ALL')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [br'5 3 4 1 2'])
self.assertEqual(server.args, ['SORT', '(SUBJECT REVERSE DATE)', 'UTF-8', 'ALL'])
response[:] = ['* SORT']
typ, data = client.uid('sort', '(SUBJECT)', 'US-ASCII', 'TEXT', '"not in mailbox"')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [br''])
self.assertEqual(server.args, ['SORT', '(SUBJECT)', 'US-ASCII', 'TEXT', '"not in mailbox"'])
typ, data = client.uid('sort', 'SUBJECT', 'NF_Z_62-010_(1973)', 'TEXT', '"not in mailbox"')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['SORT', '(SUBJECT)', '"NF_Z_62-010_(1973)"', 'TEXT', '"not in mailbox"'])
def test_thread(self):
response = []
client, server = self._setup(make_simple_handler('THREAD', response))
client.login('user', 'pass')
client.select()
response[:] = [
'* THREAD (166)(167)(168)(169)(172)(170)(171)'
'(173)(174 (175)(176)(178)(181)(180))(179)(177 '
'(183)(182)(188)(184)(185)(186)(187)(189))(190)'
'(191)(192)(193)(194 195)(196 (197)(198))(199)'
'(200 202)(201)(203)(204)(205)(206 207)(208)']
typ, data = client.thread('ORDEREDSUBJECT', 'UTF-8', 'SINCE', '5-MAR-2000')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [
b'(166)(167)(168)(169)(172)(170)(171)'
b'(173)(174 (175)(176)(178)(181)(180))(179)(177 '
b'(183)(182)(188)(184)(185)(186)(187)(189))(190)'
b'(191)(192)(193)(194 195)(196 (197)(198))(199)'
b'(200 202)(201)(203)(204)(205)(206 207)(208)'])
self.assertEqual(server.args, ['ORDEREDSUBJECT', 'UTF-8', 'SINCE', '5-MAR-2000'])
response[:] = [
'* THREAD (166)(167)(168)(169)(172)((170)(179))'
'(171)(173)((174)(175)(176)(178)(181)(180))'
'((177)(183)(182)(188 (184)(189))(185 186)(187))'
'(190)(191)(192)(193)((194)(195 196))(197 198)'
'(199)(200 202)(201)(203)(204)(205 206 207)(208)']
typ, data = client.thread('ORDEREDSUBJECT', 'US-ASCII', 'TEXT', '"gewp"')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [
b'(166)(167)(168)(169)(172)((170)(179))'
b'(171)(173)((174)(175)(176)(178)(181)(180))'
b'((177)(183)(182)(188 (184)(189))(185 186)(187))'
b'(190)(191)(192)(193)((194)(195 196))(197 198)'
b'(199)(200 202)(201)(203)(204)(205 206 207)(208)'])
self.assertEqual(server.args, ['ORDEREDSUBJECT', 'US-ASCII', 'TEXT', '"gewp"'])
typ, data = client.thread('ORDEREDSUBJECT', 'NF_Z_62-010_(1973)', 'TEXT', '"gewp"')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['ORDEREDSUBJECT', '"NF_Z_62-010_(1973)"', 'TEXT', '"gewp"'])
def test_uid_thread(self):
response = []
client, server = self._setup(make_simple_handler('UID', response,
'UID THREAD completed'))
client.login('user', 'pass')
client.select()
response[:] = [
'* THREAD (166)(167)(168)(169)(172)(170)(171)'
'(173)(174 (175)(176)(178)(181)(180))(179)(177 '
'(183)(182)(188)(184)(185)(186)(187)(189))(190)'
'(191)(192)(193)(194 195)(196 (197)(198))(199)'
'(200 202)(201)(203)(204)(205)(206 207)(208)']
typ, data = client.uid('THREAD', 'ORDEREDSUBJECT', 'UTF-8', 'SINCE', '5-MAR-2000')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [
b'(166)(167)(168)(169)(172)(170)(171)'
b'(173)(174 (175)(176)(178)(181)(180))(179)(177 '
b'(183)(182)(188)(184)(185)(186)(187)(189))(190)'
b'(191)(192)(193)(194 195)(196 (197)(198))(199)'
b'(200 202)(201)(203)(204)(205)(206 207)(208)'])
self.assertEqual(server.args, ['THREAD', 'ORDEREDSUBJECT', 'UTF-8', 'SINCE', '5-MAR-2000'])
response[:] = [
'* THREAD (166)(167)(168)(169)(172)((170)(179))'
'(171)(173)((174)(175)(176)(178)(181)(180))'
'((177)(183)(182)(188 (184)(189))(185 186)(187))'
'(190)(191)(192)(193)((194)(195 196))(197 198)'
'(199)(200 202)(201)(203)(204)(205 206 207)(208)']
typ, data = client.uid('THREAD', 'ORDEREDSUBJECT', 'US-ASCII', 'TEXT', '"gewp"')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [
b'(166)(167)(168)(169)(172)((170)(179))'
b'(171)(173)((174)(175)(176)(178)(181)(180))'
b'((177)(183)(182)(188 (184)(189))(185 186)(187))'
b'(190)(191)(192)(193)((194)(195 196))(197 198)'
b'(199)(200 202)(201)(203)(204)(205 206 207)(208)'])
self.assertEqual(server.args, ['THREAD', 'ORDEREDSUBJECT', 'US-ASCII', 'TEXT', '"gewp"'])
typ, data = client.uid('THREAD', 'ORDEREDSUBJECT', 'NF_Z_62-010_(1973)', 'TEXT', '"gewp"')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['THREAD', 'ORDEREDSUBJECT', '"NF_Z_62-010_(1973)"', 'TEXT', '"gewp"'])
def test_delete(self):
client, server = self._setup(make_simple_handler('DELETE'))
client.login('user', 'pass')
typ, data = client.delete('blurdybloop')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'DELETE completed'])
self.assertEqual(server.args, ['blurdybloop'])
typ, data = client.delete('foo/bar')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['foo/bar'])
typ, data = client.delete('New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"'])
def test_rename(self):
client, server = self._setup(make_simple_handler('RENAME'))
client.login('user', 'pass')
typ, data = client.rename('blurdybloop', 'sarasoop')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'RENAME completed'])
self.assertEqual(server.args, ['blurdybloop', 'sarasoop'])
typ, data = client.rename('Old folder', 'New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"Old folder"', '"New folder"'])
def test_subscribe(self):
client, server = self._setup(make_simple_handler('SUBSCRIBE'))
client.login('user', 'pass')
typ, data = client.subscribe('#news.comp.mail.mime')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'SUBSCRIBE completed'])
self.assertEqual(server.args, ['#news.comp.mail.mime'])
typ, data = client.subscribe('New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"'])
def test_unsubscribe(self):
client, server = self._setup(make_simple_handler('UNSUBSCRIBE'))
client.login('user', 'pass')
typ, data = client.unsubscribe('#news.comp.mail.mime')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'UNSUBSCRIBE completed'])
self.assertEqual(server.args, ['#news.comp.mail.mime'])
typ, data = client.unsubscribe('New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"'])
def test_list(self):
client, server = self._setup(make_simple_handler('LIST',
[r'* LIST (\Noselect) "/" ""',
r'* LIST (\Unmarked) "/" "~/Mail/foo"']))
client.login('user', 'pass')
typ, data = client.list()
self.assertEqual(typ, 'OK')
self.assertEqual(data, [br'(\Noselect) "/" ""',
br'(\Unmarked) "/" "~/Mail/foo"'])
self.assertEqual(server.args, ['""', '*'])
typ, data = client.list('~/Mail/', '%')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['~/Mail/', '%'])
typ, data = client.list('New folder', '*')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"', '*'])
# A pattern without wildcards is quoted when necessary.
typ, data = client.list('~/Mail/', 'My Folder')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['~/Mail/', '"My Folder"'])
def test_status(self):
client, server = self._setup(make_simple_handler('STATUS',
['* STATUS blurdybloop (MESSAGES 231 UIDNEXT 44292)']))
client.login('user', 'pass')
typ, data = client.status('blurdybloop', '(UIDNEXT MESSAGES)')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'blurdybloop (MESSAGES 231 UIDNEXT 44292)'])
self.assertEqual(server.args, ['blurdybloop', '(UIDNEXT MESSAGES)'])
# The names argument is wrapped in parentheses if it is not already.
typ, data = client.status('New folder', 'UIDNEXT MESSAGES')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"', '(UIDNEXT MESSAGES)'])
def test_getacl(self):
client, server = self._setup(make_simple_handler('GETACL',
['* ACL INBOX Fred rwipslxetad']))
client.login('user', 'pass')
typ, data = client.getacl('INBOX')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'INBOX Fred rwipslxetad'])
self.assertEqual(server.args, ['INBOX'])
typ, data = client.getacl('New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"'])
def test_setacl(self):
client, server = self._setup(make_simple_handler('SETACL'))
client.login('user', 'pass')
typ, data = client.setacl('INBOX', 'Fred', 'rwipslxetad')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'SETACL completed'])
self.assertEqual(server.args, ['INBOX', 'Fred', 'rwipslxetad'])
typ, data = client.setacl('New folder', 'Fred', '+lr')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"', 'Fred', '+lr'])
# The identifier and the rights are quoted when necessary too.
typ, data = client.setacl('INBOX', 'John Doe', 'a b')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['INBOX', '"John Doe"', '"a b"'])
def test_deleteacl(self):
client, server = self._setup(make_simple_handler('DELETEACL'))
client.login('user', 'pass')
typ, data = client.deleteacl('INBOX', 'Fred')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'DELETEACL completed'])
self.assertEqual(server.args, ['INBOX', 'Fred'])
# The identifier is quoted when necessary too.
typ, data = client.deleteacl('New folder', 'John Doe')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"', '"John Doe"'])
def test_myrights(self):
client, server = self._setup(make_simple_handler('MYRIGHTS',
['* MYRIGHTS INBOX rwiptsldaex']))
client.login('user', 'pass')
typ, data = client.myrights('INBOX')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'INBOX rwiptsldaex'])
self.assertEqual(server.args, ['INBOX'])
typ, data = client.myrights('New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"'])
def test_getquota(self):
client, server = self._setup(make_simple_handler('GETQUOTA',
['* QUOTA "" (STORAGE 10 512)']))
client.login('user', 'pass')
typ, data = client.getquota('#news')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'"" (STORAGE 10 512)'])
self.assertEqual(server.args, ['#news'])
typ, data = client.getquota('')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['""'])
def test_getquotaroot(self):
client, server = self._setup(make_simple_handler('GETQUOTAROOT',
['* QUOTAROOT INBOX ""', '* QUOTA "" (STORAGE 10 512)']))
client.login('user', 'pass')
typ, data = client.getquotaroot('INBOX')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [[b'INBOX ""'], [b'"" (STORAGE 10 512)']])
self.assertEqual(server.args, ['INBOX'])
typ, data = client.getquotaroot('New folder')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"'])
def test_setquota(self):
client, server = self._setup(make_simple_handler('SETQUOTA',
['* QUOTA "" (STORAGE 512)']))
client.login('user', 'pass')
typ, data = client.setquota('#news', '(STORAGE 512)')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'"" (STORAGE 512)'])
self.assertEqual(server.args, ['#news', '(STORAGE 512)'])
typ, data = client.setquota('', '(STORAGE 512)')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['""', '(STORAGE 512)'])
# The limits argument is wrapped in parentheses if it is not already.
typ, data = client.setquota('', 'STORAGE 512')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['""', '(STORAGE 512)'])
def test_getannotation(self):
client, server = self._setup(make_simple_handler('GETANNOTATION',
['* ANNOTATION INBOX "/comment" ("value.shared" "Hello")']))
client.login('user', 'pass')
typ, data = client.getannotation('INBOX', '/comment', 'value.shared')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'INBOX "/comment" ("value.shared" "Hello")'])
self.assertEqual(server.args, ['INBOX', '/comment', 'value.shared'])
typ, data = client.getannotation('New folder', '/comment', 'value.shared')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"New folder"', '/comment', 'value.shared'])
def test_setannotation(self):
client, server = self._setup(make_simple_handler('SETANNOTATION'))
client.login('user', 'pass')
typ, data = client.setannotation('INBOX', '/comment',
'("value.shared" "My comment")')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args,
['INBOX', '/comment', '("value.shared" "My comment")'])
typ, data = client.setannotation('New folder', '/comment',
'("value.shared" "My comment")')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args,
['"New folder"', '/comment', '("value.shared" "My comment")'])
def test_proxyauth(self):
client, server = self._setup(make_simple_handler('PROXYAUTH'))
client.login('user', 'pass')
typ, data = client.proxyauth('user')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'PROXYAUTH completed'])
self.assertEqual(server.args, ['user'])
typ, data = client.proxyauth('us er')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['"us er"'])
def test_xatom(self):
client, server = self._setup(make_simple_handler('MYCOMMAND',
completed='MYCOMMAND completed'))
client.login('user', 'pass')
self.addCleanup(imaplib.Commands.pop, 'MYCOMMAND', None)
typ, data = client.xatom('MYCOMMAND', 'arg1', 'arg2')
self.assertEqual(typ, 'OK')
self.assertEqual(data, [b'MYCOMMAND completed'])
self.assertEqual(server.args, ['arg1', 'arg2'])
def test_uppercase_command_names(self):
client, server = self._setup(SimpleIMAPHandler)
client.login('user', 'pass')
self.assertEqual(client.CAPABILITY, client.capability)
self.assertEqual(client.SELECT, client.select)
typ, data = client.CAPABILITY()
self.assertEqual(typ, 'OK')
with self.assertRaises(AttributeError):
client.NONEXISTENT
def test_control_characters(self):
client, server = self._setup(SimpleIMAPHandler)
client.login('user', 'pass')
for c in '\0\r\n':
with self.assertRaises(ValueError):
client.select(f'a{c}b')
# Other control characters are valid in a quoted string and can
# occur in mailbox names returned by the server, so the client
# must be able to send them back.
for c in support.control_characters_c0():
if c in '\0\r\n':
continue
typ, _ = client.select(f'a{c}b')
self.assertEqual(typ, 'OK')
self.assertEqual(server.is_selected, [f'"a{c}b"'])
# property tests
def test_file_property_getter(self):
client, _ = self._setup(SimpleIMAPHandler)
with self.assertWarns(DeprecationWarning):
self.assertIsInstance(client.file.raw, socket.SocketIO)
def test_file_property_setter(self):
client, _ = self._setup(SimpleIMAPHandler)
with self.assertWarns(DeprecationWarning):
# ensure that the caller closes the existing file
client.file.close()
for new_file in [mock.Mock(), None]:
with self.assertWarns(DeprecationWarning):
client.file = new_file
with self.assertWarns(DeprecationWarning):
self.assertIs(client.file, new_file)
def test_file_property_setter_should_not_close_previous_file(self):
client, _ = self._setup(SimpleIMAPHandler)
with mock.patch.object(client, "_imaplib_file", mock.Mock()) as f:
f.close.assert_not_called()
with self.assertWarns(DeprecationWarning):
self.assertIs(client.file, f)
with self.assertWarns(DeprecationWarning):
client.file = None
with self.assertWarns(DeprecationWarning):
self.assertIsNone(client.file)
f.close.assert_not_called()
class NewIMAPTests(NewIMAPTestsMixin, unittest.TestCase):
imap_class = imaplib.IMAP4
server_class = socketserver.TCPServer
@unittest.skipUnless(ssl, "SSL not available")
class NewIMAPSSLTests(NewIMAPTestsMixin, unittest.TestCase):
imap_class = IMAP4_SSL
server_class = SecureTCPServer
def test_ssl_raises(self):
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
self.assertEqual(ssl_context.verify_mode, ssl.CERT_REQUIRED)
self.assertEqual(ssl_context.check_hostname, True)
ssl_context.load_verify_locations(CAFILE)
# Allow for flexible libssl error messages.
regex = re.compile(r"""(
IP address mismatch, certificate is not valid for '127.0.0.1' # OpenSSL
|
CERTIFICATE_VERIFY_FAILED # AWS-LC
)""", re.X)
with self.assertRaisesRegex(ssl.CertificateError, regex):
_, server = self._setup(SimpleIMAPHandler, connect=False)
client = self.imap_class(*server.server_address,
ssl_context=ssl_context)
client.shutdown()
def test_ssl_verified(self):
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.load_verify_locations(CAFILE)
_, server = self._setup(SimpleIMAPHandler, connect=False)
client = self.imap_class("localhost", server.server_address[1],
ssl_context=ssl_context)
client.shutdown()
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((socket_helper.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()
@threading_helper.reap_threads
def test_connect(self):
with self.reaped_server(SimpleIMAPHandler) as server:
client = self.imap_class(*server.server_address)
client.shutdown()
@threading_helper.reap_threads
def test_bracket_flags(self):
# This violates RFC 3501, which disallows ']' characters in flags,
# but imaplib has allowed producing such flags 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)
@threading_helper.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)
@threading_helper.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')
@threading_helper.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.
@threading_helper.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')
@threading_helper.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')
@threading_helper.reap_threads
def test_enable_UTF8_True_append(self):
class UTF8AppendServer(self.UTF8Server):
def cmd_APPEND(self, tag, args):
self.server.response = args
self.server.response.append(_read_literal(self, args[-1]))
literal = yield
self.server.response.append(literal)
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')
self.assertEqual(server.args, ['UTF8=ACCEPT'])
msg_string = 'Subject: üñí©öðé'
typ, data = client.append(
None, None, None, (msg_string + '\n').encode('utf-8'))
self.assertEqual(typ, 'OK')
self.assertEqual(server.response,
['INBOX', 'UTF8',
'(~{25}', ('%s\r\n' % msg_string).encode('utf-8'),
b')\r\n' ])
# 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.
@threading_helper.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')
@threading_helper.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)
@threading_helper.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')
@threading_helper.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'
@threading_helper.reap_threads
@hashlib_helper.requires_hashdigest('md5', openssl=True)
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")
@threading_helper.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)
def test_truncated_large_literal(self):
size = 0
class BadHandler(SimpleIMAPHandler):
def handle(self):
self._send_textline('* OK {%d}' % size)
self._send_textline('IMAP4rev1')
for exponent in range(15, 64):
size = 1 << exponent
with self.subTest(f"size=2e{size}"):
with self.reaped_server(BadHandler) as server:
with self.assertRaises(imaplib.IMAP4.abort):
self.imap_class(*server.server_address)
@threading_helper.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
@threading_helper.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', '"pass"'])
self.assertIsNone(server.logged)
@threading_helper.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', '"pass"'])
imap.logout()
self.assertIsNone(server.logged)
self.assertIsNone(server.logged)
@threading_helper.reap_threads
@cpython_only
@unittest.skipUnless(__debug__, "Won't work if __debug__ is False")
def test_dump_ur(self):
# See: http://bugs.python.org/issue26543
untagged_resp_dict = {'READ-WRITE': [b'']}
with self.reaped_server(SimpleIMAPHandler) as server:
with self.imap_class(*server.server_address) as imap:
with mock.patch.object(imap, '_mesg') as mock_mesg:
imap._dump_ur(untagged_resp_dict)
mock_mesg.assert_called_with(
"untagged responses dump:READ-WRITE: [b'']"
)
@unittest.skipUnless(ssl, "SSL not available")
class ThreadedNetworkedTestsSSL(ThreadedNetworkedTests):
server_class = SecureTCPServer
imap_class = IMAP4_SSL
@threading_helper.reap_threads
def test_ssl_verified(self):
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.load_verify_locations(CAFILE)
# Allow for flexible libssl error messages.
regex = re.compile(r"""(
IP address mismatch, certificate is not valid for '127.0.0.1' # OpenSSL
|
CERTIFICATE_VERIFY_FAILED # AWS-LC
)""", re.X)
with self.assertRaisesRegex(ssl.CertificateError, regex):
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()
class TestModule(unittest.TestCase):
def test_deprecated__version__(self):
with self.assertWarnsRegex(
DeprecationWarning,
"'__version__' is deprecated and slated for removal in Python 3.20",
) as cm:
getattr(imaplib, "__version__")
self.assertEqual(cm.filename, __file__)
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink