If UTF8 is enabled, charset MUST be None.
If 'uid' is true, the message numbers in the response are UIDs
(UID SEARCH).
+
+ A 'criteria' passed as str is encoded to 'charset'; pass bytes to
+ send criteria that are already encoded.
"""
name = 'SEARCH'
if charset is not None:
if self.utf8_enabled:
raise IMAP4.error("Non-None charset not valid in UTF8 mode")
+ criteria = self._encode_criteria(charset, criteria)
args = ('CHARSET', self._astring(charset), *criteria)
else:
args = criteria
#if not name in self.capabilities: # Let the server decide!
# raise self.error('unimplemented extension command: %s' % name)
sort_criteria = self._set_quote(sort_criteria)
+ search_criteria = self._encode_criteria(charset, search_criteria)
if charset is not None:
charset = self._astring(charset)
args = (sort_criteria, charset, *search_criteria)
(UID THREAD).
"""
name = 'THREAD'
+ search_criteria = self._encode_criteria(charset, search_criteria)
if charset is not None:
charset = self._astring(charset)
args = (self._atom(threading_algorithm), charset, *search_criteria)
self._set_quote(flags))
elif command == 'SORT':
sort_criteria, charset, *search_criteria = args
+ search_criteria = self._encode_criteria(charset, search_criteria)
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
+ search_criteria = self._encode_criteria(charset, search_criteria)
if charset is not None:
charset = self._astring(charset)
args = (self._atom(threading_algorithm), charset,
return arg
return self._set_quote(arg)
+ def _encode_criteria(self, charset, criteria):
+ # Encode str search criteria to the declared CHARSET so the bytes on
+ # the wire match it. bytes criteria are already encoded and pass
+ # through unchanged. charset is None when no CHARSET is sent.
+ if charset is None:
+ return criteria
+ if isinstance(charset, (bytes, bytearray)):
+ charset = str(charset, 'ascii')
+ return tuple(c.encode(charset) if isinstance(c, str) else c
+ for c in criteria)
+
def _quote(self, arg):
if isinstance(arg, str):
arg = bytes(arg, self._encoding)
self.assertEqual(m._astring('Entwürfe'), '"Entwürfe"'.encode())
self.assertEqual(m._astring(b'Entw\xc3\xbcrfe'), b'"Entw\xc3\xbcrfe"')
+ def test_encode_criteria(self):
+ m = imaplib.IMAP4.__new__(imaplib.IMAP4)
+ enc = m._encode_criteria
+ # No charset: criteria are returned unchanged.
+ self.assertEqual(enc(None, ('TEXT', 'x')), ('TEXT', 'x'))
+ # str criteria are encoded to the charset; ASCII is charset-independent.
+ self.assertEqual(enc('UTF-8', ('TEXT', 'XXXXXX')), (b'TEXT', b'XXXXXX'))
+ # Non-ASCII text is encoded to the declared charset, including charsets
+ # other than ASCII, Latin-1 and UTF-8.
+ self.assertEqual(enc('UTF-8', ('"café"',)), ('"café"'.encode('utf-8'),))
+ self.assertEqual(enc('ISO-8859-1', ('"café"',)),
+ ('"café"'.encode('latin-1'),))
+ self.assertEqual(enc('KOI8-U', ('"Київ"',)),
+ ('"Київ"'.encode('koi8-u'),))
+ self.assertEqual(enc('SHIFT_JIS', ('"日本"',)),
+ ('"日本"'.encode('shift_jis'),))
+ # bytes criteria are already encoded and pass through unchanged.
+ self.assertEqual(enc('SHIFT_JIS', (b'"already"',)), (b'"already"',))
+ # The charset name may itself be bytes.
+ self.assertEqual(enc(b'UTF-8', ('"café"',)), ('"café"'.encode('utf-8'),))
+ # A charset with no codec at all (not even via the iconv codec) cannot
+ # encode str criteria; bytes criteria must be used with such
+ # server-only charsets.
+ self.assertRaises(LookupError, enc, 'no-such-charset', ('TEXT',))
+ self.assertEqual(enc('no-such-charset', (b'TEXT',)), (b'TEXT',))
+
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.
except StopIteration:
self.continuation = None
continue
- splitline = splitargs(line.decode().removesuffix('\r\n'))
+ self.server.line = line
+ # surrogateescape so a criterion encoded in a non-UTF-8 charset
+ # does not crash the handler; tests inspect server.line for bytes.
+ splitline = splitargs(line.decode('utf-8', 'surrogateescape')
+ .removesuffix('\r\n'))
tag = splitline[0]
cmd = splitline[1]
args = splitline[2:]
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')
+ # A non-ASCII str criterion is encoded to the declared charset (KOI8-U
+ # here, which is not UTF-8, so check the encoded bytes on the wire).
+ response[:] = ['* SEARCH 43']
+ typ, data = client.search('KOI8-U', 'SUBJECT', '"Київ"')
+ self.assertEqual(typ, 'OK')
+ self.assertIn(b'CHARSET KOI8-U ', server.line)
+ self.assertIn('"Київ"'.encode('koi8-u'), server.line)
+
+ # bytes criteria keep this focused on charset-name quoting (the
+ # parentheses force the name to be quoted) without criteria encoding.
+ typ, data = client.search('NF_Z_62-010_(1973)', b'TEXT', b'XXXXXX')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['CHARSET', '"NF_Z_62-010_(1973)"', 'TEXT', 'XXXXXX'])
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"')
+ typ, data = client.sort('SUBJECT', 'NF_Z_62-010_(1973)', b'TEXT', b'"not in mailbox"')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['(SUBJECT)', '"NF_Z_62-010_(1973)"', 'TEXT', '"not in mailbox"'])
+ # A non-ASCII str criterion is encoded to the declared charset.
+ response[:] = ['* SORT']
+ typ, data = client.sort('(SUBJECT)', 'KOI8-U', 'TEXT', '"Київ"')
+ self.assertEqual(typ, 'OK')
+ self.assertIn('"Київ"'.encode('koi8-u'), server.line)
+
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"')
+ typ, data = client.uid('sort', 'SUBJECT', 'NF_Z_62-010_(1973)', b'TEXT', b'"not in mailbox"')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['SORT', '(SUBJECT)', '"NF_Z_62-010_(1973)"', 'TEXT', '"not in mailbox"'])
+ # A non-ASCII str criterion is encoded to the declared charset.
+ response[:] = ['* SORT']
+ typ, data = client.uid('sort', '(SUBJECT)', 'KOI8-U', 'TEXT', '"Київ"')
+ self.assertEqual(typ, 'OK')
+ self.assertIn('"Київ"'.encode('koi8-u'), server.line)
+
# The uid=True keyword is a shorthand for uid('SORT', ...).
response[:] = ['* SORT 2 84 882']
typ, data = client.sort('(SUBJECT)', 'UTF-8', 'SINCE', '1-Feb-1994', uid=True)
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"')
+ typ, data = client.thread('ORDEREDSUBJECT', 'NF_Z_62-010_(1973)', b'TEXT', b'"gewp"')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['ORDEREDSUBJECT', '"NF_Z_62-010_(1973)"', 'TEXT', '"gewp"'])
+ # A non-ASCII str criterion is encoded to the declared charset.
+ response[:] = ['* THREAD (1)']
+ typ, data = client.thread('ORDEREDSUBJECT', 'KOI8-U', 'TEXT', '"Київ"')
+ self.assertEqual(typ, 'OK')
+ self.assertIn('"Київ"'.encode('koi8-u'), server.line)
+
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"')
+ typ, data = client.uid('THREAD', 'ORDEREDSUBJECT', 'NF_Z_62-010_(1973)', b'TEXT', b'"gewp"')
self.assertEqual(typ, 'OK')
self.assertEqual(server.args, ['THREAD', 'ORDEREDSUBJECT', '"NF_Z_62-010_(1973)"', 'TEXT', '"gewp"'])
+ # A non-ASCII str criterion is encoded to the declared charset.
+ response[:] = ['* THREAD (1)']
+ typ, data = client.uid('THREAD', 'ORDEREDSUBJECT', 'KOI8-U', 'TEXT', '"Київ"')
+ self.assertEqual(typ, 'OK')
+ self.assertIn('"Київ"'.encode('koi8-u'), server.line)
+
# The uid=True keyword is a shorthand for uid('THREAD', ...).
response[:] = ['* THREAD (166)(167)(168)']
typ, data = client.thread('ORDEREDSUBJECT', 'UTF-8', 'SINCE', '5-MAR-2000',