1009 lines
37 KiB
Python
1009 lines
37 KiB
Python
#!/usr/bin/python
|
|
|
|
# Copyright (c) 2014, Menno Smits
|
|
# Released subject to the New BSD License
|
|
# Please see http://en.wikipedia.org/wiki/BSD_licenses
|
|
|
|
from __future__ import print_function, unicode_literals
|
|
|
|
import imp
|
|
import os
|
|
import random
|
|
import re
|
|
import string
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from email.utils import make_msgid
|
|
|
|
from six import binary_type, text_type, PY3, iteritems
|
|
|
|
from .config import parse_config_file, create_client_from_config
|
|
from .fixed_offset import FixedOffset
|
|
from .imapclient import IMAPClient, DELETED, RECENT, _dict_bytes_normaliser
|
|
from .response_types import Envelope, Address
|
|
from .test.util import unittest
|
|
from .util import to_bytes, to_unicode
|
|
|
|
# TODO cleaner verbose output: avoid "__main__" and separator between classes
|
|
|
|
|
|
SIMPLE_MESSAGE = 'Subject: something\r\n\r\nFoo\r\n'
|
|
|
|
# Simple address in To header triggers interesting Fastmail.fm
|
|
# behaviour with ENVELOPE responses.
|
|
MULTIPART_MESSAGE = """\
|
|
From: Bob Smith <bob@smith.com>
|
|
To: Some One <some@one.com>, foo@foo.com
|
|
Date: Tue, 16 Mar 2010 16:45:32 +0000
|
|
MIME-Version: 1.0
|
|
Subject: A multipart message
|
|
Content-Type: multipart/mixed; boundary="===============1534046211=="
|
|
|
|
--===============1534046211==
|
|
Content-Type: text/html; charset="us-ascii"
|
|
Content-Transfer-Encoding: quoted-printable
|
|
|
|
<html><body>
|
|
Here is the first part.
|
|
</body></html>
|
|
|
|
--===============1534046211==
|
|
Content-Type: text/plain; charset="us-ascii"
|
|
Content-Transfer-Encoding: 7bit
|
|
|
|
Here is the second part.
|
|
|
|
--===============1534046211==--
|
|
""".replace('\n', '\r\n')
|
|
|
|
|
|
SMILE = '\u263a'
|
|
MICRO = '\u00b5'
|
|
|
|
SMILE_MESSAGE = b"""\
|
|
Subject: stuff
|
|
Content-Type: text/plain; charset="UTF-8"
|
|
|
|
\xe2\x98\xba
|
|
""".replace(b'\n', b'\r\n')
|
|
|
|
|
|
class _TestBase(unittest.TestCase):
|
|
|
|
conf = None
|
|
use_uid = True
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.client = create_client_from_config(cls.conf)
|
|
cls.client.use_uid = cls.use_uid
|
|
cls.base_folder = cls.conf.namespace[0] + '__imapclient'
|
|
cls.folder_delimiter = cls.conf.namespace[1]
|
|
|
|
def setUp(self):
|
|
self.clear_test_folders()
|
|
self.unsub_all_test_folders()
|
|
self.client.create_folder(self.base_folder)
|
|
self.client.select_folder(self.base_folder)
|
|
|
|
def tearDown(self):
|
|
self.clear_test_folders()
|
|
self.unsub_all_test_folders()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.client.logout()
|
|
|
|
def skip_unless_capable(self, capability, name=None):
|
|
if not self.client.has_capability(capability):
|
|
if not name:
|
|
name = capability
|
|
self.skipTest("Server doesn't support %s" % name)
|
|
|
|
def just_folder_names(self, dat):
|
|
if self.client.folder_encode:
|
|
gmail_special_prefix = '['
|
|
else:
|
|
gmail_special_prefix = b'['
|
|
ret = []
|
|
for _, _, folder_name in dat:
|
|
# gmail's "special" folders start with '['
|
|
if not folder_name.startswith(gmail_special_prefix):
|
|
ret.append(folder_name)
|
|
return ret
|
|
|
|
def all_test_folder_names(self):
|
|
return self.just_folder_names(self.client.list_folders(self.base_folder))
|
|
|
|
def all_sub_test_folder_names(self):
|
|
return self.just_folder_names(self.client.list_sub_folders(self.base_folder))
|
|
|
|
def clear_test_folders(self):
|
|
# Some servers (e.g. newer Dovecot) don't like it when you
|
|
# delete the currently selected folder.
|
|
try:
|
|
self.client.close_folder()
|
|
except IMAPClient.Error:
|
|
pass
|
|
|
|
self.client.folder_encode = False
|
|
folder_names = sorted(self.all_test_folder_names(),
|
|
key=self.get_folder_depth,
|
|
reverse=True)
|
|
for folder in folder_names:
|
|
try:
|
|
self.client.delete_folder(folder)
|
|
except IMAPClient.Error:
|
|
if not self.is_fastmail():
|
|
raise
|
|
self.client.folder_encode = True
|
|
|
|
def get_folder_depth(self, folder):
|
|
# Sort folders depth first because some implementations
|
|
# (e.g. MS Exchange) will delete child folders when a
|
|
# parent is deleted.
|
|
return folder.count(self.folder_delimiter.encode('ascii'))
|
|
|
|
def clear_folder(self, folder):
|
|
self.client.select_folder(folder)
|
|
self.client.delete_messages(self.client.search())
|
|
self.client.expunge()
|
|
|
|
def add_prefix_to_folder(self, folder):
|
|
if isinstance(folder, binary_type):
|
|
return self.base_folder.encode('ascii') + \
|
|
self.folder_delimiter.encode('ascii') + folder
|
|
else:
|
|
return self.base_folder + self.folder_delimiter + folder
|
|
|
|
def add_prefix_to_folders(self, folders):
|
|
return [self.add_prefix_to_folder(folder) for folder in folders]
|
|
|
|
def unsub_all_test_folders(self):
|
|
for folder in self.all_sub_test_folder_names():
|
|
self.client.unsubscribe_folder(folder)
|
|
|
|
def is_gmail(self):
|
|
return self.client._imap.host == 'imap.gmail.com'
|
|
|
|
def is_fastmail(self):
|
|
return self.client._imap.host == 'mail.messagingengine.com'
|
|
|
|
def is_exchange(self):
|
|
# Assume that these capabilities mean we're talking to MS
|
|
# Exchange. A bit of a guess really.
|
|
return (self.client.has_capability('IMAP4') and
|
|
self.client.has_capability('AUTH=NTLM') and
|
|
self.client.has_capability('AUTH=GSSAPI'))
|
|
|
|
def append_msg(self, msg, folder=None):
|
|
if not folder:
|
|
folder = self.base_folder
|
|
self.client.append(folder, msg)
|
|
if self.is_gmail():
|
|
self.client.noop()
|
|
|
|
|
|
class TestGeneral(_TestBase):
|
|
"""
|
|
Tests that don't involve message number/UID functionality.
|
|
"""
|
|
|
|
def test_alternates(self):
|
|
# Check alternate connection/login configurations.
|
|
for name, conf in iteritems(self.conf.alternates):
|
|
if PY3 and conf.oauth:
|
|
print("Skipping OAUTH test %r on Python 3 (not compatible)" % name)
|
|
continue
|
|
try:
|
|
client = create_client_from_config(conf)
|
|
client.logout()
|
|
except Exception as err:
|
|
if conf.expect_failure:
|
|
if not re.search(conf.expect_failure, str(err)):
|
|
self.fail(
|
|
"connection test %r failed with %r, expected %r" %
|
|
(name, err, conf.expect_failure))
|
|
else:
|
|
self.fail("connection test %r failed unexpectedly with %r" % (name, err))
|
|
else:
|
|
if conf.expect_failure:
|
|
self.fail(
|
|
"connection test %r didn't fail, expected %r" %
|
|
(name, conf.expect_failure))
|
|
|
|
def test_capabilities(self):
|
|
caps = self.client.capabilities()
|
|
self.assertIsInstance(caps, tuple)
|
|
self.assertGreater(len(caps), 1)
|
|
for cap in caps:
|
|
self.assertTrue(self.client.has_capability(cap))
|
|
self.assertFalse(self.client.has_capability('WONT EXIST'))
|
|
|
|
def test_namespace(self):
|
|
self.skip_unless_capable('NAMESPACE')
|
|
|
|
def assertNoneOrTuple(val):
|
|
assert val is None or isinstance(val, tuple), \
|
|
"unexpected namespace value %r" % val
|
|
|
|
ns = self.client.namespace()
|
|
self.assertEqual(len(ns), 3)
|
|
assertNoneOrTuple(ns.personal)
|
|
assertNoneOrTuple(ns.other)
|
|
assertNoneOrTuple(ns.shared)
|
|
self.assertEqual(ns.personal, ns[0])
|
|
self.assertEqual(ns.other, ns[1])
|
|
self.assertEqual(ns.shared, ns[2])
|
|
|
|
def test_select_and_close(self):
|
|
resp = self.client.select_folder(self.base_folder)
|
|
self.assertEqual(resp[b'EXISTS'], 0)
|
|
self.assertIsInstance(resp[b'RECENT'], int)
|
|
self.assertIsInstance(resp[b'FLAGS'], tuple)
|
|
self.assertGreater(len(resp[b'FLAGS']), 1)
|
|
self.client.close_folder()
|
|
|
|
def test_select_read_only(self):
|
|
self.append_msg(SIMPLE_MESSAGE)
|
|
untagged = _dict_bytes_normaliser(self.client._imap.untagged_responses)
|
|
self.assertNotIn(b'READ-ONLY', untagged)
|
|
|
|
resp = self.client.select_folder(self.base_folder, readonly=True)
|
|
|
|
untagged = _dict_bytes_normaliser(self.client._imap.untagged_responses)
|
|
self.assertIn(b'READ-ONLY', untagged)
|
|
self.assertEqual(resp[b'EXISTS'], 1)
|
|
self.assertIsInstance(resp[b'RECENT'], int)
|
|
self.assertIsInstance(resp[b'FLAGS'], tuple)
|
|
self.assertGreater(len(resp[b'FLAGS']), 1)
|
|
|
|
def test_list_folders(self):
|
|
some_folders = ['simple', b'simple2', 'L\xffR']
|
|
if not self.is_fastmail():
|
|
some_folders.extend([r'test"folder"', br'foo\bar'])
|
|
some_folders = self.add_prefix_to_folders(some_folders)
|
|
for name in some_folders:
|
|
self.client.create_folder(name)
|
|
|
|
folders = self.all_test_folder_names()
|
|
self.assertGreater(len(folders), 1, 'No folders visible on server')
|
|
self.assertIn(self.base_folder, folders)
|
|
for name in some_folders:
|
|
self.assertIn(to_unicode(name), folders)
|
|
|
|
# TODO: test LIST with wildcards
|
|
|
|
def test_gmail_xlist(self):
|
|
caps = self.client.capabilities()
|
|
if self.is_gmail():
|
|
self.assertIn(b"XLIST", caps, "expected XLIST in Gmail's capabilities")
|
|
|
|
def test_xlist(self):
|
|
self.skip_unless_capable('XLIST')
|
|
|
|
result = self.client.xlist_folders()
|
|
self.assertGreater(len(result), 0, 'No folders returned by XLIST')
|
|
|
|
foundInbox = False
|
|
for flags, _, _ in result:
|
|
if br'\INBOX' in [flag.upper() for flag in flags]:
|
|
foundInbox = True
|
|
break
|
|
if not foundInbox:
|
|
self.fail('INBOX not returned in XLIST output')
|
|
|
|
def test_subscriptions(self):
|
|
folders = self.add_prefix_to_folders([
|
|
'foobar',
|
|
b'foobar2',
|
|
'stuff & things',
|
|
b'stuff & things2',
|
|
'test & \u2622',
|
|
])
|
|
for folder in folders:
|
|
self.client.create_folder(folder)
|
|
self.client.subscribe_folder(folder)
|
|
|
|
server_folders = self.all_test_folder_names()
|
|
server_folders.remove(self.base_folder)
|
|
server_folders.sort()
|
|
self.assertListEqual(server_folders, sorted(self.all_sub_test_folder_names()))
|
|
|
|
for folder in folders:
|
|
self.client.unsubscribe_folder(folder)
|
|
self.assertListEqual(self.all_sub_test_folder_names(), [])
|
|
|
|
# Exchange doesn't return an error when subscribing to a
|
|
# non-existent folder
|
|
if not self.is_exchange():
|
|
self.assertRaises(IMAPClient.Error,
|
|
self.client.subscribe_folder,
|
|
'this folder is not likely to exist')
|
|
|
|
def test_folders(self):
|
|
self.assertTrue(self.client.folder_exists(self.base_folder))
|
|
self.assertFalse(self.client.folder_exists('this is very unlikely to exist'))
|
|
|
|
folders = [
|
|
'foobar',
|
|
'123',
|
|
b'foobar',
|
|
b'123',
|
|
]
|
|
if not self.is_fastmail():
|
|
# Fastmail doesn't appear to like double quotes in folder names
|
|
folders.extend([
|
|
'"foobar"',
|
|
'foo "bar"',
|
|
b'"foobar"',
|
|
b'foo "bar"',
|
|
])
|
|
|
|
# Run folder tests with folder_encode off
|
|
self.run_folder_tests(folders, False)
|
|
|
|
# Now with folder_encode on, adding in names that only work
|
|
# when this is enabled.
|
|
folders.extend([
|
|
'test & \u2622',
|
|
'stuff & things',
|
|
b'stuff & things',
|
|
])
|
|
self.run_folder_tests(folders, True)
|
|
|
|
def run_folder_tests(self, folder_names, folder_encode):
|
|
self.client.folder_encode = folder_encode
|
|
try:
|
|
folder_names = self.add_prefix_to_folders(folder_names)
|
|
|
|
for folder in folder_names:
|
|
self.assertFalse(self.client.folder_exists(folder))
|
|
|
|
self.client.create_folder(folder)
|
|
|
|
self.assertTrue(self.client.folder_exists(folder))
|
|
|
|
self.assertIn(
|
|
to_unicode(folder) if folder_encode else to_bytes(folder),
|
|
self.all_test_folder_names()
|
|
)
|
|
|
|
self.client.select_folder(folder)
|
|
self.client.close_folder()
|
|
|
|
self.client.delete_folder(folder)
|
|
self.assertFalse(self.client.folder_exists(folder))
|
|
finally:
|
|
self.client.folder_encode = True
|
|
|
|
def test_rename_folder(self):
|
|
folders = self.add_prefix_to_folders([
|
|
'foobar',
|
|
b'foobar2',
|
|
'stuff & things',
|
|
b'stuff & things2',
|
|
'123',
|
|
b'1232',
|
|
'test & \u2622',
|
|
])
|
|
for folder in folders:
|
|
self.client.create_folder(folder)
|
|
|
|
if isinstance(folder, binary_type):
|
|
new_folder = folder + b'x'
|
|
else:
|
|
new_folder = folder + 'x'
|
|
|
|
resp = self.client.rename_folder(folder, new_folder)
|
|
self.assertIsInstance(resp, binary_type)
|
|
self.assertTrue(len(resp) > 0)
|
|
|
|
self.assertFalse(self.client.folder_exists(folder))
|
|
self.assertTrue(self.client.folder_exists(new_folder))
|
|
|
|
def test_status(self):
|
|
# Default behaviour should return 5 keys
|
|
self.assertEqual(len(self.client.folder_status(self.base_folder)), 5)
|
|
|
|
new_folder = self.add_prefix_to_folder('test \u2622')
|
|
self.client.create_folder(new_folder)
|
|
try:
|
|
status = self.client.folder_status(new_folder)
|
|
self.assertEqual(status[b'MESSAGES'], 0)
|
|
self.assertEqual(status[b'RECENT'], 0)
|
|
self.assertEqual(status[b'UNSEEN'], 0)
|
|
|
|
# Add a message to the folder, it should show up now.
|
|
self.append_msg(SIMPLE_MESSAGE, new_folder)
|
|
|
|
status = self.client.folder_status(new_folder)
|
|
self.assertEqual(status[b'MESSAGES'], 1)
|
|
if not self.is_gmail():
|
|
self.assertEqual(status[b'RECENT'], 1)
|
|
self.assertEqual(status[b'UNSEEN'], 1)
|
|
finally:
|
|
self.client.delete_folder(new_folder)
|
|
|
|
def test_idle(self):
|
|
if not self.client.has_capability('IDLE'):
|
|
return self.skipTest("Server doesn't support IDLE")
|
|
|
|
# Start main connection idling
|
|
self.client.select_folder(self.base_folder)
|
|
self.client.idle()
|
|
|
|
try:
|
|
# Start a new connection and upload a new message
|
|
client2 = create_client_from_config(self.conf)
|
|
self.addCleanup(quiet_logout, client2)
|
|
client2.select_folder(self.base_folder)
|
|
client2.append(self.base_folder, SIMPLE_MESSAGE)
|
|
|
|
# Check for the idle data
|
|
responses = self.client.idle_check(timeout=5)
|
|
finally:
|
|
text, more_responses = self.client.idle_done()
|
|
self.assertIn((1, b'EXISTS'), responses)
|
|
self.assertTrue(isinstance(text, binary_type))
|
|
self.assertGreater(len(text), 0)
|
|
self.assertTrue(isinstance(more_responses, list))
|
|
|
|
# Check for IDLE data returned by idle_done()
|
|
|
|
# Gmail now delays updates following APPEND making this
|
|
# part of the test impractical.
|
|
if self.is_gmail():
|
|
return
|
|
|
|
self.client.idle()
|
|
try:
|
|
client2.select_folder(self.base_folder)
|
|
client2.append(self.base_folder, SIMPLE_MESSAGE)
|
|
time.sleep(2) # Allow some time for the IDLE response to be sent
|
|
finally:
|
|
text, responses = self.client.idle_done()
|
|
self.assertIn((2, b'EXISTS'), responses)
|
|
self.assertTrue(isinstance(text, binary_type))
|
|
self.assertGreater(len(text), 0)
|
|
|
|
def test_noop(self):
|
|
self.client.select_folder(self.base_folder)
|
|
|
|
# Initially there should be no responses
|
|
text, resps = self.client.noop()
|
|
self.assertTrue(isinstance(text, binary_type))
|
|
self.assertGreater(len(text), 0)
|
|
self.assertEqual(resps, [])
|
|
|
|
# Start a new connection and upload a new message
|
|
client2 = create_client_from_config(self.conf)
|
|
self.addCleanup(quiet_logout, client2)
|
|
client2.select_folder(self.base_folder)
|
|
client2.append(self.base_folder, SIMPLE_MESSAGE)
|
|
|
|
# Check for this addition in the NOOP data
|
|
msg, resps = self.client.noop()
|
|
self.assertTrue(isinstance(text, binary_type))
|
|
self.assertGreater(len(text), 0)
|
|
self.assertTrue(isinstance(resps, list))
|
|
self.assertIn((1, b'EXISTS'), resps)
|
|
|
|
|
|
def createUidTestClass(conf, use_uid):
|
|
|
|
class LiveTest(_TestBase):
|
|
"""
|
|
Tests could possibily involve message number/UID functionality
|
|
or change behaviour based on the use_uid attribute should go
|
|
here.
|
|
|
|
They are tested twice: once with use_uid on and once with it
|
|
off.
|
|
"""
|
|
|
|
def test_append_unicode(self):
|
|
self.check_append(SIMPLE_MESSAGE, SIMPLE_MESSAGE)
|
|
|
|
def test_append_bytes(self):
|
|
self.check_append(SIMPLE_MESSAGE.encode('ascii'), SIMPLE_MESSAGE)
|
|
|
|
def check_append(self, in_message, out_message):
|
|
# Message time microseconds are set to 0 because the server will return
|
|
# time with only seconds precision.
|
|
msg_time = datetime.now().replace(microsecond=0)
|
|
|
|
# Append message
|
|
resp = self.client.append(self.base_folder, in_message, ('abc', 'def'), msg_time)
|
|
self.assertIsInstance(resp, binary_type)
|
|
|
|
# Retrieve the just added message and check that all looks well
|
|
self.assertEqual(self.client.select_folder(self.base_folder)[b'EXISTS'], 1)
|
|
|
|
resp = self.client.fetch(self.client.search()[0], ('RFC822', 'FLAGS', 'INTERNALDATE'))
|
|
|
|
self.assertEqual(len(resp), 1)
|
|
msginfo = tuple(resp.values())[0]
|
|
|
|
# Time should match the time we specified
|
|
returned_msg_time = msginfo[b'INTERNALDATE']
|
|
self.assertIsNone(returned_msg_time.tzinfo)
|
|
self.assertEqual(returned_msg_time, msg_time)
|
|
|
|
# Flags should be the same
|
|
self.assertIn(b'abc', msginfo[b'FLAGS'])
|
|
self.assertIn(b'def', msginfo[b'FLAGS'])
|
|
|
|
# Message body should match
|
|
self.assertEqual(msginfo[b'RFC822'], to_bytes(out_message))
|
|
|
|
def test_flags(self):
|
|
self.append_msg(SIMPLE_MESSAGE)
|
|
msg_id = self.client.search()[0]
|
|
|
|
def _flagtest(func, args, expected_flags):
|
|
answer = func(msg_id, *args)
|
|
self.assertTrue(msg_id in answer)
|
|
answer_flags = set(answer[msg_id])
|
|
answer_flags.discard(RECENT) # Might be present but don't care
|
|
self.assertSetEqual(answer_flags, set(to_bytes(f) for f in expected_flags))
|
|
|
|
base_flags = ['abc', 'def']
|
|
_flagtest(self.client.set_flags, [base_flags], base_flags)
|
|
_flagtest(self.client.get_flags, [], base_flags)
|
|
_flagtest(self.client.add_flags, ['boo'], base_flags + ['boo'])
|
|
_flagtest(self.client.remove_flags, ['boo'], base_flags)
|
|
|
|
def test_gmail_labels(self):
|
|
self.skip_unless_capable('X-GM-EXT-1', 'labels')
|
|
|
|
self.append_msg(SIMPLE_MESSAGE)
|
|
msg_id = self.client.search()[0]
|
|
|
|
def _labeltest(func, args, expected_labels):
|
|
answer = func(msg_id, *args)
|
|
self.assertEqual(list(answer.keys()), [msg_id])
|
|
actual_labels = set(answer[msg_id])
|
|
self.assertSetEqual(actual_labels, set(expected_labels))
|
|
|
|
FOO = b'_imapclient_foo'
|
|
BAR = b'_imapclient_bar'
|
|
BAZ = b'_imapclient_baz'
|
|
all_labels = [FOO, BAR, BAZ]
|
|
base_labels = [FOO, BAR]
|
|
try:
|
|
_labeltest(self.client.set_gmail_labels, [base_labels], base_labels)
|
|
_labeltest(self.client.get_gmail_labels, [], base_labels)
|
|
_labeltest(self.client.add_gmail_labels, [BAZ], all_labels)
|
|
_labeltest(self.client.remove_gmail_labels, [BAZ], base_labels)
|
|
finally:
|
|
# Clean up folders created by assigning labels.
|
|
for label in all_labels:
|
|
if self.client.folder_exists(label):
|
|
self.client.delete_folder(label)
|
|
|
|
def test_search(self):
|
|
# Add some test messages
|
|
msg_tmpl = 'Subject: %s\r\n\r\nBody'
|
|
subjects = ('a', 'b', 'c')
|
|
for subject in subjects:
|
|
msg = msg_tmpl % subject
|
|
if subject == 'c':
|
|
flags = (DELETED,)
|
|
else:
|
|
flags = ()
|
|
self.client.append(self.base_folder, msg, flags)
|
|
self.client.noop() # For Gmail
|
|
|
|
# Check we see all messages
|
|
messages_all = self.client.search('ALL')
|
|
if self.is_gmail():
|
|
# Gmail seems to never return deleted items.
|
|
self.assertEqual(len(messages_all), len(subjects) - 1)
|
|
else:
|
|
self.assertEqual(len(messages_all), len(subjects))
|
|
self.assertListEqual(self.client.search(), messages_all) # Check default
|
|
|
|
if not self.is_gmail():
|
|
# Delete behaviour is dependent on a setting with Gmail.
|
|
self.assertEqual(len(self.client.search('DELETED')), 1)
|
|
|
|
self.assertEqual(len(self.client.search(['NOT', 'DELETED'])), len(subjects) - 1)
|
|
self.assertEqual(len(self.client.search(
|
|
['NOT', 'DELETED', 'SMALLER', 500])), len(subjects) - 1)
|
|
self.assertEqual(len(self.client.search(['NOT', 'DELETED', 'SMALLER', 5])), 0)
|
|
self.assertEqual(len(self.client.search(['NOT', 'DELETED', 'SUBJECT', 'a'])), 1)
|
|
self.assertEqual(len(self.client.search(['NOT', 'DELETED', 'SUBJECT', 'c'])), 0)
|
|
|
|
# Exercise "raw" strings where all criteria are provided as a single string.
|
|
self.assertEqual(len(self.client.search('SUBJECT "a" NOT DELETED')), 1)
|
|
self.assertEqual(len(self.client.search('NOT DELETED SUBJECT SMALLER 5')), 0)
|
|
|
|
def test_search_with_modseq(self):
|
|
# CONDSTORE (RFC 4551) means that the server supports the
|
|
# MODSEQ search criteria and response.
|
|
if not self.client.has_capability('CONDSTORE'):
|
|
return self.skipTest("Server doesn't support CONDSTORE")
|
|
|
|
if self.is_gmail():
|
|
return self.skipTest(
|
|
"Gmail doesn't seem to return MODSEQ parts in SEARCH responses")
|
|
|
|
# A little dance to ensure MODSEQ tracking is turned on.
|
|
# TODO: use ENABLE for this instead
|
|
self.client.select_folder(self.base_folder)
|
|
self.append_msg(SIMPLE_MESSAGE)
|
|
msg_id = self.client.search()[0]
|
|
self.client.fetch(msg_id, ["MODSEQ"])
|
|
self.client.close_folder()
|
|
self.clear_folder(self.base_folder)
|
|
|
|
# Remember the initial MODSEQ
|
|
initial_modseq = self.client.select_folder(self.base_folder)[b'HIGHESTMODSEQ']
|
|
|
|
# Add a message so that the MODSEQ increases
|
|
self.append_msg(SIMPLE_MESSAGE)
|
|
|
|
# Ensure the message is seen and the new MODSEQ value is returned
|
|
ids = self.client.search(['MODSEQ', str(initial_modseq)])
|
|
self.assertEqual(len(ids), 1)
|
|
self.assertGreater(ids.modseq, initial_modseq)
|
|
|
|
def test_search_with_unicode(self):
|
|
self.client.append(self.base_folder, SMILE_MESSAGE)
|
|
|
|
self.assertEqual(len(self.client.search(['BODY', SMILE], charset='UTF-8')), 1)
|
|
self.assertEqual(len(self.client.search(['BODY', MICRO], charset='UTF-8')), 0)
|
|
|
|
# Try multiple criteria too
|
|
self.assertEqual(len(self.client.search(
|
|
['TEXT', SMILE, 'NOT', 'DELETED'], charset='UTF-8')), 1)
|
|
|
|
def test_gmail_search(self):
|
|
self.skip_unless_capable('X-GM-EXT-1', 'Gmail search')
|
|
|
|
random_string = ''.join(random.sample(string.ascii_letters * 20, 64))
|
|
msg = 'Subject: something\r\n\r\nFoo\r\n%s\r\n' % random_string
|
|
self.append_msg(msg)
|
|
|
|
self.append_msg(SMILE_MESSAGE)
|
|
|
|
ids = self.client.gmail_search(random_string)
|
|
self.assertEqual(len(ids), 1)
|
|
|
|
ids = self.client.gmail_search('s0mewh4t unl1kely')
|
|
self.assertEqual(len(ids), 0)
|
|
|
|
# Test encoded queries
|
|
ids = self.client.gmail_search(MICRO)
|
|
self.assertEqual(len(ids), 0)
|
|
ids = self.client.gmail_search(SMILE)
|
|
self.assertGreater(len(ids), 0)
|
|
|
|
def test_sort(self):
|
|
self.skip_unless_capable('SORT')
|
|
|
|
# Add some test messages
|
|
msg_tmpl = 'Subject: Test\r\n\r\nBody'
|
|
num_lines = (10, 20, 30)
|
|
line = '\n' + ('x' * 72)
|
|
for line_cnt in num_lines:
|
|
msg = msg_tmpl + (line * line_cnt)
|
|
self.client.append(self.base_folder, msg)
|
|
|
|
messages = self.client.sort('REVERSE SIZE')
|
|
self.assertEqual(len(messages), 3)
|
|
first_id = messages[0]
|
|
expected = [first_id, first_id - 1, first_id - 2]
|
|
self.assertListEqual(messages, expected)
|
|
|
|
messages = self.client.sort('REVERSE SIZE', ['NOT', 'DELETED'])
|
|
self.assertListEqual(messages, expected)
|
|
|
|
messages = self.client.sort('REVERSE SIZE', 'NOT DELETED')
|
|
self.assertListEqual(messages, expected)
|
|
|
|
def test_sort_with_unicode(self):
|
|
self.skip_unless_capable('SORT')
|
|
self.append_msg(SMILE_MESSAGE)
|
|
|
|
messages = self.client.sort('ARRIVAL', ['TEXT', SMILE])
|
|
self.assertEqual(len(messages), 1)
|
|
|
|
messages = self.client.sort('ARRIVAL', ['TEXT', MICRO])
|
|
self.assertEqual(len(messages), 0)
|
|
|
|
def test_thread(self):
|
|
self.skip_unless_capable('THREAD=REFERENCES')
|
|
|
|
msg_tmpl = 'Subject: %s\r\n\r\nBody'
|
|
subjects = ('a', 'b', 'c')
|
|
for subject in subjects:
|
|
self.append_msg(msg_tmpl % subject)
|
|
|
|
threads = self.client.thread()
|
|
|
|
self.assertEqual(len(threads), 3)
|
|
self.assertIsInstance(threads[0], tuple)
|
|
first_id = threads[0][0]
|
|
expected = ((first_id,), (first_id + 1,), (first_id + 2,))
|
|
self.assertTupleEqual(threads, expected)
|
|
|
|
threads = self.client.thread(criteria=['NOT', 'DELETED'])
|
|
self.assertTupleEqual(threads, expected)
|
|
|
|
threads = self.client.thread(criteria='NOT DELETED')
|
|
self.assertTupleEqual(threads, expected)
|
|
|
|
def test_thread_with_unicode(self):
|
|
self.skip_unless_capable('THREAD=REFERENCES')
|
|
|
|
self.append_msg(SMILE_MESSAGE)
|
|
|
|
threads = self.client.thread(criteria=['TEXT', SMILE])
|
|
self.assertEqual(len(threads), 1)
|
|
self.assertEqual(len(threads[0]), 1)
|
|
|
|
threads = self.client.thread(criteria=['TEXT', MICRO])
|
|
self.assertEqual(len(threads), 0)
|
|
|
|
def test_copy(self):
|
|
self.append_msg(SIMPLE_MESSAGE)
|
|
target_folder = self.add_prefix_to_folder('target')
|
|
self.client.create_folder(target_folder)
|
|
msg_id = self.client.search()[0]
|
|
|
|
self.client.copy(msg_id, target_folder)
|
|
|
|
self.client.select_folder(target_folder)
|
|
msgs = self.client.search()
|
|
self.assertEqual(len(msgs), 1)
|
|
msg_id = msgs[0]
|
|
self.assertIn(b'something', self.client.fetch(msg_id, ['RFC822'])[msg_id][b'RFC822'])
|
|
|
|
def test_fetch(self):
|
|
# Generate a fresh message-id each time because Gmail is
|
|
# clever and will treat appends of messages with
|
|
# previously seen message-ids as the same message. This
|
|
# breaks our tests when the test message is updated.
|
|
msg_id_header = make_msgid()
|
|
msg = ('Message-ID: %s\r\n' % msg_id_header) + MULTIPART_MESSAGE
|
|
|
|
self.client.select_folder(self.base_folder)
|
|
self.append_msg(msg)
|
|
self.client.normalise_times = False
|
|
|
|
fields = ['RFC822', b'FLAGS', 'INTERNALDATE', 'ENVELOPE']
|
|
msg_id = self.client.search()[0]
|
|
resp = self.client.fetch(msg_id, fields)
|
|
|
|
self.assertEqual(len(resp), 1)
|
|
msginfo = resp[msg_id]
|
|
|
|
self.assertSetEqual(
|
|
set(msginfo.keys()),
|
|
set([to_bytes(f) for f in fields] + [b'SEQ'])
|
|
)
|
|
self.assertEqual(msginfo[b'SEQ'], 1)
|
|
self.assertEqual(msginfo[b'RFC822'], to_bytes(msg))
|
|
self.assertIsInstance(msginfo[b'INTERNALDATE'], datetime)
|
|
self.assertIsInstance(msginfo[b'FLAGS'], tuple)
|
|
self.assertSequenceEqual(msginfo[b'ENVELOPE'],
|
|
Envelope(
|
|
datetime(2010, 3, 16, 16, 45, 32, tzinfo=FixedOffset(0)),
|
|
b'A multipart message',
|
|
(Address(b'Bob Smith', None, b'bob', b'smith.com'),),
|
|
(Address(b'Bob Smith', None, b'bob', b'smith.com'),),
|
|
(Address(b'Bob Smith', None, b'bob', b'smith.com'),),
|
|
(Address(b'Some One', None, b'some', b'one.com'),
|
|
Address(None, None, b'foo', b'foo.com')),
|
|
None, None, None, to_bytes(msg_id_header)))
|
|
|
|
def test_partial_fetch(self):
|
|
self.client.append(self.base_folder, MULTIPART_MESSAGE)
|
|
self.client.select_folder(self.base_folder)
|
|
msg_id = self.client.search()[0]
|
|
|
|
resp = self.client.fetch(msg_id, ['BODY[]<0.20>'])
|
|
body = resp[msg_id][b'BODY[]<0>']
|
|
self.assertEqual(len(body), 20)
|
|
self.assertTrue(body.startswith(b'From: Bob Smith'))
|
|
|
|
resp = self.client.fetch(msg_id, ['BODY[]<2.25>'])
|
|
body = resp[msg_id][b'BODY[]<2>']
|
|
self.assertEqual(len(body), 25)
|
|
self.assertTrue(body.startswith(b'om: Bob Smith'))
|
|
|
|
def test_fetch_modifiers(self):
|
|
# CONDSTORE (RFC 4551) provides a good way to use FETCH
|
|
# modifiers but it isn't commonly available.
|
|
if not self.client.has_capability('CONDSTORE'):
|
|
return self.skipTest("Server doesn't support CONDSTORE")
|
|
|
|
# A little dance to ensure MODSEQ tracking is turned on.
|
|
self.client.select_folder(self.base_folder)
|
|
self.append_msg(SIMPLE_MESSAGE)
|
|
msg_id = self.client.search()[0]
|
|
self.client.fetch(msg_id, ["MODSEQ"])
|
|
self.client.close_folder()
|
|
self.clear_folder(self.base_folder)
|
|
|
|
#
|
|
# Actual testing starts here
|
|
#
|
|
|
|
# Get the starting MODSEQ
|
|
modseq = self.client.select_folder(self.base_folder)[b'HIGHESTMODSEQ']
|
|
|
|
# Add a message so that the MODSEQ gets bumped
|
|
self.append_msg(SIMPLE_MESSAGE)
|
|
msg_id = self.client.search()[0]
|
|
|
|
# Request changes since the starting MODSEQ: this should
|
|
# return the new message.
|
|
resp = self.client.fetch(msg_id, ['FLAGS'], ['CHANGEDSINCE %d' % modseq])
|
|
new_modseq = resp[msg_id][b"MODSEQ"][0]
|
|
self.assertGreater(new_modseq, modseq)
|
|
|
|
# Now asked for changes since the MODSEQ on the added
|
|
# message. These shouldn't be any.
|
|
resp = self.client.fetch(msg_id, ['FLAGS'], ['CHANGEDSINCE %d' % new_modseq])
|
|
self.assertEqual(resp, {})
|
|
|
|
def test_BODYSTRUCTURE(self):
|
|
self.client.select_folder(self.base_folder)
|
|
self.append_msg(SIMPLE_MESSAGE)
|
|
self.append_msg(MULTIPART_MESSAGE)
|
|
msgs = self.client.search()
|
|
|
|
fetched = self.client.fetch(msgs, ['BODY', 'BODYSTRUCTURE'])
|
|
|
|
# The expected test data is the same for BODY and BODYSTRUCTURE
|
|
# since we can't predicate what the server we're testing against
|
|
# will return.
|
|
|
|
expected = (b'text', b'plain', (b'charset', b'us-ascii'), None, None, b'7bit', 5, 1)
|
|
self.check_BODYSTRUCTURE(expected, fetched[msgs[0]][b'BODY'], multipart=False)
|
|
self.check_BODYSTRUCTURE(expected, fetched[msgs[0]][b'BODYSTRUCTURE'], multipart=False)
|
|
|
|
expected = ([(b'text', b'html', (b'charset', b'us-ascii'), None, None, b'quoted-printable', 55, 3),
|
|
(b'text', b'plain', (b'charset', b'us-ascii'), None, None, b'7bit', 26, 1),
|
|
],
|
|
b'mixed',
|
|
(b'boundary', b'===============1534046211=='))
|
|
self.check_BODYSTRUCTURE(expected, fetched[msgs[1]][b'BODY'], multipart=True)
|
|
self.check_BODYSTRUCTURE(expected, fetched[msgs[1]][b'BODYSTRUCTURE'], multipart=True)
|
|
|
|
def check_BODYSTRUCTURE(self, expected, actual, multipart=None):
|
|
if multipart is not None:
|
|
self.assertEqual(actual.is_multipart, multipart)
|
|
|
|
if have_matching_types(expected, actual, (list, tuple)):
|
|
# BODYSTRUCTURE lengths can various according to the
|
|
# server so compare up until what is returned
|
|
for pair in zip(expected, actual):
|
|
self.check_BODYSTRUCTURE(*pair)
|
|
elif expected == (b'charset', b'us-ascii') and actual is None:
|
|
pass # Some servers don't return a charset when it's us-ascii
|
|
else:
|
|
self.assertEqual(maybe_lower(expected), maybe_lower(actual))
|
|
|
|
def test_expunge(self):
|
|
self.client.select_folder(self.base_folder)
|
|
|
|
# Test empty mailbox
|
|
text, resps = self.client.expunge()
|
|
self.assertTrue(isinstance(text, binary_type))
|
|
self.assertGreater(len(text), 0)
|
|
# Some servers return nothing while others (e.g. Exchange) return (0, 'EXISTS')
|
|
self.assertIn(resps, ([], [(0, b'EXISTS')]))
|
|
|
|
# Now try with a message to expunge
|
|
self.client.append(self.base_folder, SIMPLE_MESSAGE, flags=[DELETED])
|
|
|
|
msg, resps = self.client.expunge()
|
|
|
|
self.assertTrue(isinstance(text, binary_type))
|
|
self.assertGreater(len(text), 0)
|
|
self.assertTrue(isinstance(resps, list))
|
|
if not self.is_gmail():
|
|
# GMail has an auto-expunge feature which might be
|
|
# on. EXPUNGE won't return anything in this case
|
|
self.assertIn((1, b'EXPUNGE'), resps)
|
|
|
|
def test_getacl(self):
|
|
self.skip_unless_capable('ACL')
|
|
|
|
folder = self.add_prefix_to_folder('test_acl')
|
|
self.client.create_folder(folder)
|
|
|
|
who = to_bytes(conf['username'])
|
|
rights = self.client.getacl(folder)
|
|
self.assertIn(who, [u for u, r in rights])
|
|
|
|
LiveTest.conf = conf
|
|
LiveTest.use_uid = use_uid
|
|
|
|
return LiveTest
|
|
|
|
|
|
def quiet_logout(client):
|
|
"""Log out a connection, ignoring errors (say because the connection is down)
|
|
"""
|
|
try:
|
|
client.logout()
|
|
except IMAPClient.Error:
|
|
pass
|
|
|
|
|
|
def maybe_lower(val):
|
|
if isinstance(val, (text_type, binary_type)):
|
|
return val.lower()
|
|
return val
|
|
|
|
|
|
def have_matching_types(a, b, type_or_types):
|
|
"""True if a and b are instances of the same type and that type is
|
|
one of type_or_types.
|
|
"""
|
|
if not isinstance(a, type_or_types):
|
|
return False
|
|
return isinstance(b, type(a))
|
|
|
|
|
|
def argv_error(msg):
|
|
print(msg, file=sys.stderr)
|
|
print(file=sys.stderr)
|
|
print("usage: %s <livetest.ini> [ optional unittest arguments ]" % sys.argv[0], file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def parse_argv():
|
|
args = sys.argv[1:]
|
|
if not args:
|
|
argv_error('Please specify a host configuration file. See livetest-sample.ini for an example.')
|
|
ini_path = sys.argv.pop(1) # 2nd arg should be the INI file
|
|
if not os.path.isfile(ini_path):
|
|
argv_error('%r is not a livetest INI file' % ini_path)
|
|
host_config = parse_config_file(ini_path)
|
|
return host_config
|
|
|
|
|
|
def probe_host(config):
|
|
client = create_client_from_config(config)
|
|
ns = client.namespace()
|
|
client.logout()
|
|
if not ns.personal:
|
|
raise RuntimeError('Can\'t run tests: IMAP account has no personal namespace')
|
|
return ns.personal[0] # Use first personal namespace
|
|
|
|
|
|
def main():
|
|
host_config = parse_argv()
|
|
|
|
namespace = probe_host(host_config)
|
|
host_config.namespace = namespace
|
|
|
|
live_test_mod = imp.new_module('livetests')
|
|
sys.modules['livetests'] = live_test_mod
|
|
|
|
def add_test_class(klass, name=None):
|
|
if name is None:
|
|
name = klass.__name__
|
|
else:
|
|
if not PY3:
|
|
name = name.encode('ascii')
|
|
klass.__name__ = name
|
|
setattr(live_test_mod, name, klass)
|
|
|
|
TestGeneral.conf = host_config
|
|
add_test_class(TestGeneral)
|
|
add_test_class(createUidTestClass(host_config, use_uid=True), 'TestWithUIDs')
|
|
add_test_class(createUidTestClass(host_config, use_uid=False), 'TestWithoutUIDs')
|
|
|
|
unittest.main(module='livetests')
|
|
|
|
if __name__ == '__main__':
|
|
main()
|