_Literal = br'.*{(?P<size>\d+)}$'
_Untagged_status = br'\* (?P<data>\d+) (?P<type>[A-Z-]+)( (?P<data2>.*))?'
_control_chars = re.compile(b'[\x00-\x1F\x7F]')
+_non_astring_char = re.compile(br'[(){ \x00-\x1f\x7f-\xff%*\\"]')
+_non_list_char = re.compile(br'[(){ \x00-\x1f\x7f-\xff\\"]')
+_quoted = re.compile(br'"(?:[^"\\]|\\.)*+"')
class IMAP4:
if not mailbox:
mailbox = 'INBOX'
if flags:
- if (flags[0],flags[-1]) != ('(',')'):
- flags = '(%s)' % flags
+ flags = self._set_quote(flags)
else:
flags = None
if date_time:
if translate_line_endings:
message = MapCRLF.sub(CRLF, message)
self.literal = message
- return self._simple_command(name, mailbox, flags, date_time)
+ return self._simple_command(name, self._astring(mailbox), flags, date_time)
def authenticate(self, mechanism, authobject):
#if not cap in self.capabilities: # Let the server decide!
# raise self.error("Server doesn't allow %s authentication." % mech)
self.literal = _Authenticator(authobject).process
- typ, dat = self._simple_command('AUTHENTICATE', mech)
+ typ, dat = self._simple_command('AUTHENTICATE', self._atom(mech))
if typ != 'OK':
raise self.error(dat[-1].decode('utf-8', 'replace'))
self.state = 'AUTH'
(typ, [data]) = <instance>.copy(message_set, new_mailbox)
"""
- return self._simple_command('COPY', message_set, new_mailbox)
+ return self._simple_command('COPY', self._sequence_set(message_set),
+ self._astring(new_mailbox))
def create(self, mailbox):
(typ, [data]) = <instance>.create(mailbox)
"""
- return self._simple_command('CREATE', mailbox)
+ return self._simple_command('CREATE', self._astring(mailbox))
def delete(self, mailbox):
(typ, [data]) = <instance>.delete(mailbox)
"""
- return self._simple_command('DELETE', mailbox)
+ return self._simple_command('DELETE', self._astring(mailbox))
def deleteacl(self, mailbox, who):
"""Delete the ACLs (remove any rights) set for who on mailbox.
(typ, [data]) = <instance>.deleteacl(mailbox, who)
"""
- return self._simple_command('DELETEACL', mailbox, who)
+ return self._simple_command('DELETEACL', self._astring(mailbox),
+ self._astring(who))
def enable(self, capability):
"""Send an RFC5161 enable string to the server.
'data' are tuples of message part envelope and data.
"""
name = 'FETCH'
- typ, dat = self._simple_command(name, message_set, message_parts)
+ typ, dat = self._simple_command(name, self._sequence_set(message_set),
+ self._fetch_parts(message_parts))
return self._untagged_response(typ, dat, name)
(typ, [data]) = <instance>.getacl(mailbox)
"""
- typ, dat = self._simple_command('GETACL', mailbox)
+ typ, dat = self._simple_command('GETACL', self._astring(mailbox))
return self._untagged_response(typ, dat, 'ACL')
"""(typ, [data]) = <instance>.getannotation(mailbox, entry, attribute)
Retrieve ANNOTATIONs."""
- typ, dat = self._simple_command('GETANNOTATION', mailbox, entry, attribute)
+ typ, dat = self._simple_command('GETANNOTATION', self._astring(mailbox),
+ entry, attribute)
return self._untagged_response(typ, dat, 'ANNOTATION')
(typ, [data]) = <instance>.getquota(root)
"""
- typ, dat = self._simple_command('GETQUOTA', root)
+ typ, dat = self._simple_command('GETQUOTA', self._astring(root))
return self._untagged_response(typ, dat, 'QUOTA')
(typ, [[QUOTAROOT responses...], [QUOTA responses]]) = <instance>.getquotaroot(mailbox)
"""
- typ, dat = self._simple_command('GETQUOTAROOT', mailbox)
+ typ, dat = self._simple_command('GETQUOTAROOT', self._astring(mailbox))
typ, quota = self._untagged_response(typ, dat, 'QUOTA')
typ, quotaroot = self._untagged_response(typ, dat, 'QUOTAROOT')
return typ, [quotaroot, quota]
return Idler(self, duration)
- def list(self, directory='""', pattern='*'):
+ def list(self, directory='', pattern='*'):
"""List mailbox names in directory matching pattern.
- (typ, [data]) = <instance>.list(directory='""', pattern='*')
+ (typ, [data]) = <instance>.list(directory='', pattern='*')
'data' is list of LIST responses.
"""
name = 'LIST'
- typ, dat = self._simple_command(name, directory, pattern)
+ typ, dat = self._simple_command(name, self._astring(directory),
+ self._list_mailbox(pattern))
return self._untagged_response(typ, dat, name)
NB: 'password' will be quoted.
"""
- typ, dat = self._simple_command('LOGIN', user, self._quote(password))
+ typ, dat = self._simple_command('LOGIN', self._astring(user),
+ self._quote(password))
if typ != 'OK':
raise self.error(dat[-1].decode('UTF-8', 'replace'))
self.state = 'AUTH'
return typ, dat
- def lsub(self, directory='""', pattern='*'):
+ def lsub(self, directory='', pattern='*'):
"""List 'subscribed' mailbox names in directory matching pattern.
- (typ, [data, ...]) = <instance>.lsub(directory='""', pattern='*')
+ (typ, [data, ...]) = <instance>.lsub(directory='', pattern='*')
'data' are tuples of message part envelope and data.
"""
name = 'LSUB'
- typ, dat = self._simple_command(name, directory, pattern)
+ typ, dat = self._simple_command(name, self._astring(directory),
+ self._list_mailbox(pattern))
return self._untagged_response(typ, dat, name)
def myrights(self, mailbox):
(typ, [data]) = <instance>.myrights(mailbox)
"""
- typ,dat = self._simple_command('MYRIGHTS', mailbox)
+ typ,dat = self._simple_command('MYRIGHTS', self._astring(mailbox))
return self._untagged_response(typ, dat, 'MYRIGHTS')
def namespace(self):
"""
name = 'PROXYAUTH'
- return self._simple_command(name, user)
+ return self._simple_command(name, self._astring(user))
def rename(self, oldmailbox, newmailbox):
(typ, [data]) = <instance>.rename(oldmailbox, newmailbox)
"""
- return self._simple_command('RENAME', oldmailbox, newmailbox)
+ return self._simple_command('RENAME', self._astring(oldmailbox),
+ self._astring(newmailbox))
def search(self, charset, *criteria):
If UTF8 is enabled, charset MUST be None.
"""
name = 'SEARCH'
- if charset:
+ if charset is not None:
if self.utf8_enabled:
raise IMAP4.error("Non-None charset not valid in UTF8 mode")
- typ, dat = self._simple_command(name, 'CHARSET', charset, *criteria)
+ typ, dat = self._simple_command(name,
+ 'CHARSET', self._astring(charset), *criteria)
else:
typ, dat = self._simple_command(name, *criteria)
return self._untagged_response(typ, dat, name)
name = 'EXAMINE'
else:
name = 'SELECT'
- typ, dat = self._simple_command(name, mailbox)
+ typ, dat = self._simple_command(name, self._astring(mailbox))
if typ != 'OK':
self.state = 'AUTH' # Might have been 'SELECTED'
return typ, dat
(typ, [data]) = <instance>.setacl(mailbox, who, what)
"""
- return self._simple_command('SETACL', mailbox, who, what)
+ return self._simple_command('SETACL', self._astring(mailbox),
+ self._astring(who), self._astring(what))
- def setannotation(self, *args):
+ def setannotation(self, mailbox, *args):
"""(typ, [data]) = <instance>.setannotation(mailbox[, entry, attribute]+)
Set ANNOTATIONs."""
- typ, dat = self._simple_command('SETANNOTATION', *args)
+ typ, dat = self._simple_command('SETANNOTATION', self._astring(mailbox), *args)
return self._untagged_response(typ, dat, 'ANNOTATION')
(typ, [data]) = <instance>.setquota(root, limits)
"""
- typ, dat = self._simple_command('SETQUOTA', root, limits)
+ typ, dat = self._simple_command('SETQUOTA', self._astring(root),
+ self._set_quote(limits))
return self._untagged_response(typ, dat, 'QUOTA')
name = 'SORT'
#if not name in self.capabilities: # Let the server decide!
# raise self.error('unimplemented extension command: %s' % name)
- if (sort_criteria[0],sort_criteria[-1]) != ('(',')'):
- sort_criteria = '(%s)' % sort_criteria
+ sort_criteria = self._set_quote(sort_criteria)
+ if charset is not None:
+ charset = self._astring(charset)
typ, dat = self._simple_command(name, sort_criteria, charset, *search_criteria)
return self._untagged_response(typ, dat, name)
name = 'STATUS'
#if self.PROTOCOL_VERSION == 'IMAP4': # Let the server decide!
# raise self.error('%s unimplemented in IMAP4 (obtain IMAP4rev1 server, or re-code)' % name)
- typ, dat = self._simple_command(name, mailbox, names)
+ typ, dat = self._simple_command(name, self._astring(mailbox),
+ self._set_quote(names))
return self._untagged_response(typ, dat, name)
(typ, [data]) = <instance>.store(message_set, command, flags)
"""
- if (flags[0],flags[-1]) != ('(',')'):
- flags = '(%s)' % flags # Avoid quoting the flags
- typ, dat = self._simple_command('STORE', message_set, command, flags)
+ flags = self._set_quote(flags)
+ typ, dat = self._simple_command('STORE', self._sequence_set(message_set),
+ command, flags)
return self._untagged_response(typ, dat, 'FETCH')
(typ, [data]) = <instance>.subscribe(mailbox)
"""
- return self._simple_command('SUBSCRIBE', mailbox)
+ return self._simple_command('SUBSCRIBE', self._astring(mailbox))
def thread(self, threading_algorithm, charset, *search_criteria):
(type, [data]) = <instance>.thread(threading_algorithm, charset, search_criteria, ...)
"""
name = 'THREAD'
- typ, dat = self._simple_command(name, threading_algorithm, charset, *search_criteria)
+ if charset is not None:
+ charset = self._astring(charset)
+ typ, dat = self._simple_command(name, self._atom(threading_algorithm),
+ charset, *search_criteria)
return self._untagged_response(typ, dat, name)
(command, self.state,
', '.join(Commands[command])))
name = 'UID'
- typ, dat = self._simple_command(name, command, *args)
+ if command == 'COPY':
+ message_set, new_mailbox = args
+ args = (self._sequence_set(message_set),
+ self._astring(new_mailbox))
+ elif command == 'FETCH':
+ message_set, message_parts = args
+ args = (self._sequence_set(message_set),
+ self._fetch_parts(message_parts))
+ elif command == 'STORE':
+ message_set, op, flags = args
+ args = (self._sequence_set(message_set), op,
+ self._set_quote(flags))
+ elif command == 'SORT':
+ sort_criteria, charset, *search_criteria = args
+ if charset is not None:
+ charset = self._astring(charset)
+ args = (self._set_quote(sort_criteria), charset,
+ *search_criteria)
+ elif command == 'THREAD':
+ threading_algorithm, charset, *search_criteria = args
+ if charset is not None:
+ charset = self._astring(charset)
+ args = (self._atom(threading_algorithm), charset,
+ *search_criteria)
+ typ, dat = self._simple_command(name, self._atom(command), *args)
if command in ('SEARCH', 'SORT', 'THREAD'):
name = command
else:
(typ, [data]) = <instance>.unsubscribe(mailbox)
"""
- return self._simple_command('UNSUBSCRIBE', mailbox)
+ return self._simple_command('UNSUBSCRIBE', self._astring(mailbox))
def unselect(self):
return tag
- def _quote(self, arg):
+ def _atom(self, arg):
+ return arg
+
+ def _sequence_set(self, arg):
+ return arg
- arg = arg.replace('\\', '\\\\')
- arg = arg.replace('"', '\\"')
+ def _set_quote(self, arg):
+ if arg and arg[0] == '(' and arg[-1] == ')':
+ return arg
+ return '(' + arg + ')'
- return '"' + arg + '"'
+ def _fetch_parts(self, arg):
+ # "ALL", "FULL" and "FAST" are macros, not data item names;
+ # they cannot be enclosed in parentheses.
+ if arg.upper() in ('ALL', 'FULL', 'FAST'):
+ return arg
+ return self._set_quote(arg)
+ def _quote(self, arg):
+ if isinstance(arg, str):
+ arg = bytes(arg, self._encoding)
+ arg = arg.replace(b'\\', br'\\')
+ arg = arg.replace(b'"', br'\"')
+ return b'"' + arg + b'"'
+
+ # For backward compatibility, an argument already enclosed in double
+ # quotes is left unquoted, so that code which quotes arguments itself
+ # keeps working. New code should pass arguments unquoted and let the
+ # module quote them as needed.
+
+ def _astring(self, arg):
+ if isinstance(arg, str):
+ arg = bytes(arg, self._encoding)
+ if _quoted.fullmatch(arg):
+ return arg
+ if arg and _non_astring_char.search(arg) is None:
+ return arg
+ return self._quote(arg)
+
+ def _list_mailbox(self, arg):
+ if isinstance(arg, str):
+ arg = bytes(arg, self._encoding)
+ if _quoted.fullmatch(arg):
+ return arg
+ if arg and _non_list_char.search(arg) is None:
+ return arg
+ return self._quote(arg)
def _simple_command(self, name, *args):
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):
self._send_tagged(tag, 'OK', 'LOGOUT completed')
def cmd_LOGIN(self, tag, args):
- self.server.logged = args[0]
+ self.server.logged = args
self._send_tagged(tag, 'OK', 'LOGIN completed')
def cmd_SELECT(self, tag, args):
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'))
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):
_, server = self._setup(SimpleIMAPHandler, connect=False)
with self.imap_class(*server.server_address) as imap:
imap.login('user', 'pass')
- self.assertEqual(server.logged, 'user')
+ self.assertEqual(server.logged, ['user', '"pass"'])
self.assertIsNone(server.logged)
def test_with_statement_logout(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')
+ self.assertEqual(server.logged, ['user', '"pass"'])
imap.logout()
self.assertIsNone(server.logged)
self.assertIsNone(server.logged)
self.fail('multi-packet response was corrupted by idle timeout')
def test_login(self):
- client, _ = self._setup(SimpleIMAPHandler)
+ 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_translate_line_endings(self):
# By default line endings are normalized to CRLF; False sends the
translate_line_endings=False)
self.assertEqual(server.response, message)
+ # 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
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.
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']))
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')
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'))
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))',
])
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
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)',
])
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")']))
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,
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,
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))
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,
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')
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')
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')
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')
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) "/" ""',
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)']))
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']))
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')
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')
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']))
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)']))
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)']))
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)']))
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")']))
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')
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')
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'))
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'))
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.assertEqual(server.logged, ['user', '"pass"'])
self.assertIsNone(server.logged)
@threading_helper.reap_threads
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.assertEqual(server.logged, ['user', '"pass"'])
imap.logout()
self.assertIsNone(server.logged)
self.assertIsNone(server.logged)
projects / thirdparty / Python / cpython.git / commitdiff
