diff options
Diffstat (limited to 'src/leap/mail/imap/tests/test_imap.py')
| -rw-r--r-- | src/leap/mail/imap/tests/test_imap.py | 555 |
1 files changed, 310 insertions, 245 deletions
diff --git a/src/leap/mail/imap/tests/test_imap.py b/src/leap/mail/imap/tests/test_imap.py index 7837aaa..67a24cd 100644 --- a/src/leap/mail/imap/tests/test_imap.py +++ b/src/leap/mail/imap/tests/test_imap.py @@ -17,7 +17,7 @@ """ Test case for leap.email.imap.server TestCases taken from twisted tests and modified to make them work -against SoledadBackedAccount. +against our implementation of the IMAPAccount. @authors: Kali Kaneko, <kali@leap.se> XXX add authors from the original twisted tests. @@ -32,19 +32,13 @@ import types from twisted.mail import imap4 from twisted.internet import defer -from twisted.trial import unittest from twisted.python import util from twisted.python import failure from twisted import cred - -# import u1db - -from leap.mail.imap.mailbox import SoledadMailbox -from leap.mail.imap.memorystore import MemoryStore -from leap.mail.imap.messages import MessageCollection -from leap.mail.imap.server import LeapIMAPServer +from leap.mail.imap.mailbox import IMAPMailbox +from leap.mail.imap.messages import IMAPMessageCollection from leap.mail.imap.tests.utils import IMAP4HelperMixin @@ -68,7 +62,6 @@ def sortNest(l): class TestRealm: - """ A minimal auth realm for testing purposes only """ @@ -82,8 +75,8 @@ class TestRealm: # TestCases # -class MessageCollectionTestCase(IMAP4HelperMixin, unittest.TestCase): - +# TODO rename to IMAPMessageCollection +class MessageCollectionTestCase(IMAP4HelperMixin): """ Tests for the MessageCollection class """ @@ -95,10 +88,12 @@ class MessageCollectionTestCase(IMAP4HelperMixin, unittest.TestCase): We override mixin method since we are only testing MessageCollection interface in this particular TestCase """ + # FIXME -- return deferred super(MessageCollectionTestCase, self).setUp() - memstore = MemoryStore() - self.messages = MessageCollection("testmbox%s" % (self.count,), - self._soledad, memstore=memstore) + + # FIXME --- update initialization + self.messages = IMAPMessageCollection( + "testmbox%s" % (self.count,), self._soledad) MessageCollectionTestCase.count += 1 def tearDown(self): @@ -207,23 +202,18 @@ class MessageCollectionTestCase(IMAP4HelperMixin, unittest.TestCase): add_1().addCallback(lambda ignored: self.assertEqual( mc.count(), 3)) - # XXX this has to be redone to fit memstore ------------# - #newmsg = mc._get_empty_doc() - #newmsg['mailbox'] = "mailbox/foo" - #mc._soledad.create_doc(newmsg) - #self.assertEqual(mc.count(), 3) - #self.assertEqual( - #len(mc._soledad.get_from_index(mc.TYPE_IDX, "flags")), 4) + +# DEBUG --- +#from twisted.internet.base import DelayedCall +#DelayedCall.debug = True -class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): - # TODO this currently will use a memory-only store. - # create a different one for testing soledad sync. +class LEAPIMAP4ServerTestCase(IMAP4HelperMixin): """ - Tests for the generic behavior of the LeapIMAP4Server + Tests for the generic behavior of the LEAPIMAP4Server which, right now, it's just implemented in this test file as - LeapIMAPServer. We will move the implementation, together with + LEAPIMAPServer. We will move the implementation, together with authentication bits, to leap.mail.imap.server so it can be instantiated from the tac file. @@ -243,6 +233,7 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): """ succeed = ('testbox', 'test/box', 'test/', 'test/box/box', 'foobox') fail = ('testbox', 'test/box') + acc = self.server.theAccount def cb(): self.result.append(1) @@ -254,46 +245,54 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return self.client.login(TEST_USER, TEST_PASSWD) def create(): + create_deferreds = [] for name in succeed + fail: d = self.client.create(name) d.addCallback(strip(cb)).addErrback(eb) - d.addCallbacks(self._cbStopClient, self._ebGeneral) + create_deferreds.append(d) + dd = defer.gatherResults(create_deferreds) + dd.addCallbacks(self._cbStopClient, self._ebGeneral) + return dd self.result = [] - d1 = self.connected.addCallback(strip(login)).addCallback( - strip(create)) + d1 = self.connected.addCallback(strip(login)) + d1.addCallback(strip(create)) d2 = self.loopback() - d = defer.gatherResults([d1, d2]) + d = defer.gatherResults([d1, d2], consumeErrors=True) + d.addCallback(lambda _: acc.account.list_all_mailbox_names()) return d.addCallback(self._cbTestCreate, succeed, fail) - def _cbTestCreate(self, ignored, succeed, fail): + def _cbTestCreate(self, mailboxes, succeed, fail): self.assertEqual(self.result, [1] * len(succeed) + [0] * len(fail)) - mboxes = list(LeapIMAPServer.theAccount.mailboxes) - answers = ([u'INBOX', u'foobox', 'test', u'test/box', - u'test/box/box', 'testbox']) - self.assertEqual(mboxes, [a for a in answers]) + answers = ([u'INBOX', u'testbox', u'test/box', u'test', + u'test/box/box', 'foobox']) + self.assertEqual(sorted(mailboxes), sorted([a for a in answers])) def testDelete(self): """ Test whether we can delete mailboxes """ - LeapIMAPServer.theAccount.addMailbox('delete/me') + def add_mailbox(): + return self.server.theAccount.addMailbox('test-delete/me') def login(): return self.client.login(TEST_USER, TEST_PASSWD) def delete(): - return self.client.delete('delete/me') + return self.client.delete('test-delete/me') - d1 = self.connected.addCallback(strip(login)) + acc = self.server.theAccount.account + + d1 = self.connected.addCallback(add_mailbox) + d1.addCallback(strip(login)) d1.addCallbacks(strip(delete), self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) - d.addCallback( - lambda _: self.assertEqual( - LeapIMAPServer.theAccount.mailboxes, ['INBOX'])) + d.addCallback(lambda _: acc.list_all_mailbox_names()) + d.addCallback(lambda mboxes: self.assertEqual( + mboxes, ['INBOX'])) return d def testIllegalInboxDelete(self): @@ -352,24 +351,34 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): Try deleting a mailbox with sub-folders, and \NoSelect flag set. An exception is expected. """ - LeapIMAPServer.theAccount.addMailbox('delete') - to_delete = LeapIMAPServer.theAccount.getMailbox('delete') - to_delete.setFlags((r'\Noselect',)) - to_delete.getFlags() - LeapIMAPServer.theAccount.addMailbox('delete/me') + acc = self.server.theAccount def login(): return self.client.login(TEST_USER, TEST_PASSWD) - def delete(): + def create_mailboxes(): + d1 = acc.addMailbox('delete') + d2 = acc.addMailbox('delete/me') + d = defer.gatherResults([d1, d2]) + return d + + def get_noselect_mailbox(mboxes): + mbox = mboxes[0] + return mbox.setFlags((r'\Noselect',)) + + def delete_mbox(ignored): return self.client.delete('delete') def deleteFailed(failure): self.failure = failure self.failure = None + d1 = self.connected.addCallback(strip(login)) - d1.addCallback(strip(delete)).addErrback(deleteFailed) + d1.addCallback(strip(create_mailboxes)) + d1.addCallback(get_noselect_mailbox) + + d1.addCallback(delete_mbox).addErrback(deleteFailed) d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) @@ -381,11 +390,15 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): self.assertEqual(str(self.failure.value), expected)) return d + # FIXME --- this test sometimes FAILS (timing issue). + # Some of the deferreds used in the rename op is not waiting for the + # operations properly def testRename(self): """ Test whether we can rename a mailbox """ - LeapIMAPServer.theAccount.addMailbox('oldmbox') + def create_mbox(): + return self.server.theAccount.addMailbox('oldmbox') def login(): return self.client.login(TEST_USER, TEST_PASSWD) @@ -393,15 +406,16 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def rename(): return self.client.rename('oldmbox', 'newname') - d1 = self.connected.addCallback(strip(login)) + d1 = self.connected.addCallback(strip(create_mbox)) + d1.addCallback(strip(login)) d1.addCallbacks(strip(rename), self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) d.addCallback(lambda _: - self.assertEqual( - LeapIMAPServer.theAccount.mailboxes, - ['INBOX', 'newname'])) + self.server.theAccount.account.list_all_mailbox_names()) + d.addCallback(lambda mboxes: + self.assertItemsEqual(mboxes, ['INBOX', 'newname'])) return d def testIllegalInboxRename(self): @@ -435,8 +449,12 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): """ Try to rename hierarchical mailboxes """ - LeapIMAPServer.theAccount.create('oldmbox/m1') - LeapIMAPServer.theAccount.create('oldmbox/m2') + acc = self.server.theAccount + + def add_mailboxes(): + return defer.gatherResults([ + acc.addMailbox('oldmbox/m1'), + acc.addMailbox('oldmbox/m2')]) def login(): return self.client.login(TEST_USER, TEST_PASSWD) @@ -444,45 +462,62 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def rename(): return self.client.rename('oldmbox', 'newname') - d1 = self.connected.addCallback(strip(login)) + d1 = self.connected.addCallback(strip(add_mailboxes)) + d1.addCallback(strip(login)) d1.addCallbacks(strip(rename), self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) + d.addCallback(lambda _: acc.account.list_all_mailbox_names()) return d.addCallback(self._cbTestHierarchicalRename) - def _cbTestHierarchicalRename(self, ignored): - mboxes = LeapIMAPServer.theAccount.mailboxes + def _cbTestHierarchicalRename(self, mailboxes): expected = ['INBOX', 'newname', 'newname/m1', 'newname/m2'] - self.assertEqual(mboxes, [s for s in expected]) + self.assertEqual(sorted(mailboxes), sorted([s for s in expected])) def testSubscribe(self): """ Test whether we can mark a mailbox as subscribed to """ + acc = self.server.theAccount + + def add_mailbox(): + return acc.addMailbox('this/mbox') + def login(): return self.client.login(TEST_USER, TEST_PASSWD) def subscribe(): return self.client.subscribe('this/mbox') - d1 = self.connected.addCallback(strip(login)) + def get_subscriptions(ignored): + return self.server.theAccount.getSubscriptions() + + d1 = self.connected.addCallback(strip(add_mailbox)) + d1.addCallback(strip(login)) d1.addCallbacks(strip(subscribe), self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) - d.addCallback(lambda _: - self.assertEqual( - LeapIMAPServer.theAccount.subscriptions, - ['this/mbox'])) + d.addCallback(get_subscriptions) + d.addCallback(lambda subscriptions: + self.assertEqual(subscriptions, + ['this/mbox'])) return d def testUnsubscribe(self): """ Test whether we can unsubscribe from a set of mailboxes """ - LeapIMAPServer.theAccount.subscribe('this/mbox') - LeapIMAPServer.theAccount.subscribe('that/mbox') + acc = self.server.theAccount + + def add_mailboxes(): + return defer.gatherResults([ + acc.addMailbox('this/mbox'), + acc.addMailbox('that/mbox')]) + + dc1 = lambda: acc.subscribe('this/mbox') + dc2 = lambda: acc.subscribe('that/mbox') def login(): return self.client.login(TEST_USER, TEST_PASSWD) @@ -490,24 +525,35 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def unsubscribe(): return self.client.unsubscribe('this/mbox') - d1 = self.connected.addCallback(strip(login)) + def get_subscriptions(ignored): + return acc.getSubscriptions() + + d1 = self.connected.addCallback(strip(add_mailboxes)) + d1.addCallback(strip(login)) + d1.addCallback(strip(dc1)) + d1.addCallback(strip(dc2)) d1.addCallbacks(strip(unsubscribe), self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) - d.addCallback(lambda _: - self.assertEqual( - LeapIMAPServer.theAccount.subscriptions, - ['that/mbox'])) + d.addCallback(get_subscriptions) + d.addCallback(lambda subscriptions: + self.assertEqual(subscriptions, + ['that/mbox'])) return d def testSelect(self): """ Try to select a mailbox """ - self.server.theAccount.addMailbox('TESTMAILBOX-SELECT', creation_ts=42) + mbox_name = "TESTMAILBOXSELECT" self.selectedArgs = None + acc = self.server.theAccount + + def add_mailbox(): + return acc.addMailbox(mbox_name, creation_ts=42) + def login(): return self.client.login(TEST_USER, TEST_PASSWD) @@ -515,29 +561,26 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def selected(args): self.selectedArgs = args self._cbStopClient(None) - d = self.client.select('TESTMAILBOX-SELECT') + d = self.client.select(mbox_name) d.addCallback(selected) return d - d1 = self.connected.addCallback(strip(login)) + d1 = self.connected.addCallback(strip(add_mailbox)) + d1.addCallback(strip(login)) d1.addCallback(strip(select)) - d1.addErrback(self._ebGeneral) + #d1.addErrback(self._ebGeneral) d2 = self.loopback() - return defer.gatherResults([d1, d2]).addCallback(self._cbTestSelect) + + d = defer.gatherResults([d1, d2]) + d.addCallback(self._cbTestSelect) + return d def _cbTestSelect(self, ignored): - mbox = LeapIMAPServer.theAccount.getMailbox('TESTMAILBOX-SELECT') - self.assertEqual(self.server.mbox.messages.mbox, mbox.messages.mbox) - # XXX UIDVALIDITY should be "42" if the creation_ts is passed along - # to the memory store. However, the current state of the account - # implementation is incomplete and we're writing to soledad store - # directly there. We should handle the UIDVALIDITY timestamping - # mechanism in a separate test suite. + self.assertTrue(self.selectedArgs is not None) self.assertEqual(self.selectedArgs, { - 'EXISTS': 0, 'RECENT': 0, 'UIDVALIDITY': 0, - # 'EXISTS': 0, 'RECENT': 0, 'UIDVALIDITY': 42, + 'EXISTS': 0, 'RECENT': 0, 'UIDVALIDITY': 42, 'FLAGS': ('\\Seen', '\\Answered', '\\Flagged', '\\Deleted', '\\Draft', '\\Recent', 'List'), 'READ-WRITE': True @@ -555,13 +598,16 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): caps.update(c) self.server.transport.loseConnection() return self.client.getCapabilities().addCallback(gotCaps) - d1 = self.connected.addCallback( + + d1 = self.connected + d1.addCallback( strip(getCaps)).addErrback(self._ebGeneral) + d = defer.gatherResults([self.loopback(), d1]) expected = {'IMAP4rev1': None, 'NAMESPACE': None, 'LITERAL+': None, 'IDLE': None} - - return d.addCallback(lambda _: self.assertEqual(expected, caps)) + d.addCallback(lambda _: self.assertEqual(expected, caps)) + return d def testCapabilityWithAuth(self): caps = {} @@ -582,7 +628,8 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): 'IDLE': None, 'LITERAL+': None, 'AUTH': ['CRAM-MD5']} - return d.addCallback(lambda _: self.assertEqual(expCap, caps)) + d.addCallback(lambda _: self.assertEqual(expCap, caps)) + return d # # authentication @@ -630,7 +677,6 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return d.addCallback(self._cbTestLogin) def _cbTestLogin(self, ignored): - self.assertEqual(self.server.account, LeapIMAPServer.theAccount) self.assertEqual(self.server.state, 'auth') def testFailedLogin(self): @@ -668,7 +714,6 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return d.addCallback(self._cbTestLoginRequiringQuoting) def _cbTestLoginRequiringQuoting(self, ignored): - self.assertEqual(self.server.account, LeapIMAPServer.theAccount) self.assertEqual(self.server.state, 'auth') # @@ -715,11 +760,13 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): for details. """ # TODO implement the IMAP4ClientExamineTests testcase. - - self.server.theAccount.addMailbox('test-mailbox-e', - creation_ts=42) + mbox_name = "test_mailbox_e" + acc = self.server.theAccount self.examinedArgs = None + def add_mailbox(): + return acc.addMailbox(mbox_name, creation_ts=42) + def login(): return self.client.login(TEST_USER, TEST_PASSWD) @@ -727,11 +774,12 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def examined(args): self.examinedArgs = args self._cbStopClient(None) - d = self.client.examine('test-mailbox-e') + d = self.client.examine(mbox_name) d.addCallback(examined) return d - d1 = self.connected.addCallback(strip(login)) + d1 = self.connected.addCallback(strip(add_mailbox)) + d1.addCallback(strip(login)) d1.addCallback(strip(examine)) d1.addErrback(self._ebGeneral) d2 = self.loopback() @@ -739,28 +787,19 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return d.addCallback(self._cbTestExamine) def _cbTestExamine(self, ignored): - mbox = self.server.theAccount.getMailbox('test-mailbox-e') - self.assertEqual(self.server.mbox.messages.mbox, mbox.messages.mbox) - - # XXX UIDVALIDITY should be "42" if the creation_ts is passed along - # to the memory store. However, the current state of the account - # implementation is incomplete and we're writing to soledad store - # directly there. We should handle the UIDVALIDITY timestamping - # mechanism in a separate test suite. self.assertEqual(self.examinedArgs, { - 'EXISTS': 0, 'RECENT': 0, 'UIDVALIDITY': 0, - # 'EXISTS': 0, 'RECENT': 0, 'UIDVALIDITY': 42, + 'EXISTS': 0, 'RECENT': 0, 'UIDVALIDITY': 42, 'FLAGS': ('\\Seen', '\\Answered', '\\Flagged', '\\Deleted', '\\Draft', '\\Recent', 'List'), 'READ-WRITE': False}) - def _listSetup(self, f): - LeapIMAPServer.theAccount.addMailbox('root/subthingl', - creation_ts=42) - LeapIMAPServer.theAccount.addMailbox('root/another-thing', - creation_ts=42) - LeapIMAPServer.theAccount.addMailbox('non-root/subthing', - creation_ts=42) + def _listSetup(self, f, f2=None): + + acc = self.server.theAccount + + dc1 = lambda: acc.addMailbox('root_subthing', creation_ts=42) + dc2 = lambda: acc.addMailbox('root_another_thing', creation_ts=42) + dc3 = lambda: acc.addMailbox('non_root_subthing', creation_ts=42) def login(): return self.client.login(TEST_USER, TEST_PASSWD) @@ -770,6 +809,13 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): self.listed = None d1 = self.connected.addCallback(strip(login)) + d1.addCallback(strip(dc1)) + d1.addCallback(strip(dc2)) + d1.addCallback(strip(dc3)) + + if f2 is not None: + d1.addCallback(f2) + d1.addCallbacks(strip(f), self._ebGeneral) d1.addCallbacks(listed, self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) @@ -782,12 +828,13 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): """ def list(): return self.client.list('root', '%') + d = self._listSetup(list) d.addCallback(lambda listed: self.assertEqual( sortNest(listed), sortNest([ - (SoledadMailbox.INIT_FLAGS, "/", "root/subthingl"), - (SoledadMailbox.INIT_FLAGS, "/", "root/another-thing") + (IMAPMailbox.init_flags, "/", "root_subthing"), + (IMAPMailbox.init_flags, "/", "root_another_thing") ]) )) return d @@ -796,20 +843,29 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): """ Test LSub command """ - LeapIMAPServer.theAccount.subscribe('root/subthingl2') + acc = self.server.theAccount + + def subs_mailbox(): + # why not client.subscribe instead? + return acc.subscribe('root_subthing') def lsub(): return self.client.lsub('root', '%') - d = self._listSetup(lsub) + + d = self._listSetup(lsub, strip(subs_mailbox)) d.addCallback(self.assertEqual, - [(SoledadMailbox.INIT_FLAGS, "/", "root/subthingl2")]) + [(IMAPMailbox.init_flags, "/", "root_subthing")]) return d def testStatus(self): """ Test Status command """ - LeapIMAPServer.theAccount.addMailbox('root/subthings') + acc = self.server.theAccount + + def add_mailbox(): + return acc.addMailbox('root_subthings') + # XXX FIXME ---- should populate this a little bit, # with unseen etc... @@ -818,13 +874,15 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def status(): return self.client.status( - 'root/subthings', 'MESSAGES', 'UIDNEXT', 'UNSEEN') + 'root_subthings', 'MESSAGES', 'UIDNEXT', 'UNSEEN') def statused(result): self.statused = result self.statused = None - d1 = self.connected.addCallback(strip(login)) + + d1 = self.connected.addCallback(strip(add_mailbox)) + d1.addCallback(strip(login)) d1.addCallbacks(strip(status), self._ebGeneral) d1.addCallbacks(statused, self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) @@ -881,56 +939,84 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): """ infile = util.sibpath(__file__, 'rfc822.message') message = open(infile) - LeapIMAPServer.theAccount.addMailbox('root/subthing') + acc = self.server.theAccount + mailbox_name = "root/subthing" + + def add_mailbox(): + return acc.addMailbox(mailbox_name) def login(): return self.client.login(TEST_USER, TEST_PASSWD) def append(): return self.client.append( - 'root/subthing', - message, + mailbox_name, message, ('\\SEEN', '\\DELETED'), 'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)', ) - d1 = self.connected.addCallback(strip(login)) + d1 = self.connected.addCallback(strip(add_mailbox)) + d1.addCallback(strip(login)) d1.addCallbacks(strip(append), self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) + + d.addCallback(lambda _: acc.getMailbox(mailbox_name)) + d.addCallback(lambda mb: mb.fetch(imap4.MessageSet(start=1), True)) return d.addCallback(self._cbTestFullAppend, infile) - def _cbTestFullAppend(self, ignored, infile): - mb = LeapIMAPServer.theAccount.getMailbox('root/subthing') - self.assertEqual(1, len(mb.messages)) + def _cbTestFullAppend(self, fetched, infile): + self.assertTrue(len(fetched) == 1) + self.assertTrue(len(fetched[0]) == 2) + uid, msg = fetched[0] + parsed = self.parser.parse(open(infile)) + expected_body = parsed.get_payload() + expected_headers = dict(parsed.items()) - msg = mb.messages.get_msg_by_uid(1) - self.assertEqual( - set(('\\Recent', '\\SEEN', '\\DELETED')), - set(msg.getFlags())) + def assert_flags(flags): + self.assertEqual( + set(('\\SEEN', '\\DELETED')), + set(flags)) - self.assertEqual( - 'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)', - msg.getInternalDate()) + def assert_date(date): + self.assertEqual( + 'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)', + date) - parsed = self.parser.parse(open(infile)) - body = parsed.get_payload() - headers = dict(parsed.items()) - self.assertEqual( - body, - msg.getBodyFile().read()) - gotheaders = msg.getHeaders(True) + def assert_body(body): + gotbody = body.read() + self.assertEqual(expected_body, gotbody) + + def assert_headers(headers): + self.assertItemsEqual(expected_headers, headers) + + d = defer.maybeDeferred(msg.getFlags) + d.addCallback(assert_flags) + + d.addCallback(lambda _: defer.maybeDeferred(msg.getInternalDate)) + d.addCallback(assert_date) + + d.addCallback( + lambda _: defer.maybeDeferred( + msg.getBodyFile, self._soledad)) + d.addCallback(assert_body) + + d.addCallback(lambda _: defer.maybeDeferred(msg.getHeaders, True)) + d.addCallback(assert_headers) - self.assertItemsEqual( - headers, gotheaders) + return d def testPartialAppend(self): """ Test partially appending a message to the mailbox """ infile = util.sibpath(__file__, 'rfc822.message') - LeapIMAPServer.theAccount.addMailbox('PARTIAL/SUBTHING') + + acc = self.server.theAccount + + def add_mailbox(): + return acc.addMailbox('PARTIAL/SUBTHING') def login(): return self.client.login(TEST_USER, TEST_PASSWD) @@ -945,33 +1031,46 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): (), self.client._IMAP4Client__cbContinueAppend, message ) ) - d1 = self.connected.addCallback(strip(login)) + d1 = self.connected.addCallback(strip(add_mailbox)) + d1.addCallback(strip(login)) d1.addCallbacks(strip(append), self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) + + d.addCallback(lambda _: acc.getMailbox("PARTIAL/SUBTHING")) + d.addCallback(lambda mb: mb.fetch(imap4.MessageSet(start=1), True)) return d.addCallback( self._cbTestPartialAppend, infile) - def _cbTestPartialAppend(self, ignored, infile): - mb = LeapIMAPServer.theAccount.getMailbox('PARTIAL/SUBTHING') - self.assertEqual(1, len(mb.messages)) - msg = mb.messages.get_msg_by_uid(1) - self.assertEqual( - set(('\\SEEN', '\\Recent')), - set(msg.getFlags()) - ) + def _cbTestPartialAppend(self, fetched, infile): + self.assertTrue(len(fetched) == 1) + self.assertTrue(len(fetched[0]) == 2) + uid, msg = fetched[0] parsed = self.parser.parse(open(infile)) - body = parsed.get_payload() - self.assertEqual( - body, - msg.getBodyFile().read()) + expected_body = parsed.get_payload() + + def assert_flags(flags): + self.assertEqual( + set((['\\SEEN'])), set(flags)) + + def assert_body(body): + gotbody = body.read() + self.assertEqual(expected_body, gotbody) + + d = defer.maybeDeferred(msg.getFlags) + d.addCallback(assert_flags) + + d.addCallback(lambda _: defer.maybeDeferred(msg.getBodyFile)) + d.addCallback(assert_body) + return d def testCheck(self): """ Test check command """ - LeapIMAPServer.theAccount.addMailbox('root/subthing') + def add_mailbox(): + return self.server.theAccount.addMailbox('root/subthing') def login(): return self.client.login(TEST_USER, TEST_PASSWD) @@ -982,89 +1081,48 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def check(): return self.client.check() - d = self.connected.addCallback(strip(login)) + d = self.connected.addCallbacks( + strip(add_mailbox), self._ebGeneral) + d.addCallbacks(lambda _: login(), self._ebGeneral) d.addCallbacks(strip(select), self._ebGeneral) d.addCallbacks(strip(check), self._ebGeneral) d.addCallbacks(self._cbStopClient, self._ebGeneral) - return self.loopback() - - # Okay, that was fun - - def testClose(self): - """ - Test closing the mailbox. We expect to get deleted all messages flagged - as such. - """ - name = 'mailbox-close' - self.server.theAccount.addMailbox(name) - - m = LeapIMAPServer.theAccount.getMailbox(name) - - def login(): - return self.client.login(TEST_USER, TEST_PASSWD) - - def select(): - return self.client.select(name) - - def add_messages(): - d1 = m.messages.add_msg( - 'test 1', subject="Message 1", - flags=('\\Deleted', 'AnotherFlag')) - d2 = m.messages.add_msg( - 'test 2', subject="Message 2", - flags=('AnotherFlag',)) - d3 = m.messages.add_msg( - 'test 3', subject="Message 3", - flags=('\\Deleted',)) - d = defer.gatherResults([d1, d2, d3]) - return d - - def close(): - return self.client.close() - - d = self.connected.addCallback(strip(login)) - d.addCallbacks(strip(select), self._ebGeneral) - d.addCallbacks(strip(add_messages), self._ebGeneral) - d.addCallbacks(strip(close), self._ebGeneral) - d.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() - return defer.gatherResults([d, d2]).addCallback(self._cbTestClose, m) - - def _cbTestClose(self, ignored, m): - self.assertEqual(len(m.messages), 1) - msg = m.messages.get_msg_by_uid(2) - self.assertTrue(msg is not None) + return defer.gatherResults([d, d2]) - self.assertEqual( - dict(msg.hdoc.content)['subject'], - 'Message 2') - self.failUnless(m.closed) + # Okay, that was much fun indeed def testExpunge(self): """ Test expunge command """ - name = 'mailbox-expunge' - self.server.theAccount.addMailbox(name) - m = LeapIMAPServer.theAccount.getMailbox(name) + acc = self.server.theAccount + mailbox_name = 'mailboxexpunge' + + def add_mailbox(): + return acc.addMailbox(mailbox_name) def login(): return self.client.login(TEST_USER, TEST_PASSWD) def select(): - return self.client.select('mailbox-expunge') + return self.client.select(mailbox_name) + + def save_mailbox(mailbox): + self.mailbox = mailbox + + def get_mailbox(): + d = acc.getMailbox(mailbox_name) + d.addCallback(save_mailbox) + return d def add_messages(): - d1 = m.messages.add_msg( - 'test 1', subject="Message 1", - flags=('\\Deleted', 'AnotherFlag')) - d2 = m.messages.add_msg( - 'test 2', subject="Message 2", - flags=('AnotherFlag',)) - d3 = m.messages.add_msg( - 'test 3', subject="Message 3", - flags=('\\Deleted',)) - d = defer.gatherResults([d1, d2, d3]) + d = self.mailbox.addMessage( + 'test 1', flags=('\\Deleted', 'AnotherFlag')) + d.addCallback(lambda _: self.mailbox.addMessage( + 'test 2', flags=('AnotherFlag',))) + d.addCallback(lambda _: self.mailbox.addMessage( + 'test 3', flags=('\\Deleted',))) return d def expunge(): @@ -1075,47 +1133,54 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): self.results = results self.results = None - d1 = self.connected.addCallback(strip(login)) - d1.addCallbacks(strip(select), self._ebGeneral) + d1 = self.connected.addCallback(strip(add_mailbox)) + d1.addCallback(strip(login)) + d1.addCallback(strip(get_mailbox)) d1.addCallbacks(strip(add_messages), self._ebGeneral) + d1.addCallbacks(strip(select), self._ebGeneral) d1.addCallbacks(strip(expunge), self._ebGeneral) d1.addCallbacks(expunged, self._ebGeneral) d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) - return d.addCallback(self._cbTestExpunge, m) + d.addCallback(lambda _: self.mailbox.getMessageCount()) + return d.addCallback(self._cbTestExpunge) - def _cbTestExpunge(self, ignored, m): + def _cbTestExpunge(self, count): # we only left 1 mssage with no deleted flag - self.assertEqual(len(m.messages), 1) - msg = m.messages.get_msg_by_uid(2) - - msg = list(m.messages)[0] - self.assertTrue(msg is not None) - - self.assertEqual( - msg.hdoc.content['subject'], - 'Message 2') - + self.assertEqual(count, 1) # the uids of the deleted messages self.assertItemsEqual(self.results, [1, 3]) -class AccountTestCase(IMAP4HelperMixin, unittest.TestCase): +# TODO -------- Fix this testcase +class AccountTestCase(IMAP4HelperMixin): """ Test the Account. """ def _create_empty_mailbox(self): - LeapIMAPServer.theAccount.addMailbox('') + return self.server.theAccount.addMailbox('') def _create_one_mailbox(self): - LeapIMAPServer.theAccount.addMailbox('one') + return self.server.theAccount.addMailbox('one') def test_illegalMailboxCreate(self): - self.assertRaises(AssertionError, self._create_empty_mailbox) + # FIXME --- account.addMailbox needs to raise a failure, + # not the direct exception. + self.stashed = None + + def stash(result): + self.stashed = result + + d = self._create_empty_mailbox() + d.addBoth(stash) + d.addCallback(lambda _: self.failUnless(isinstance(self.stashed, + failure.Failure))) + return d + #self.assertRaises(AssertionError, self._create_empty_mailbox) -class IMAP4ServerSearchTestCase(IMAP4HelperMixin, unittest.TestCase): +class IMAP4ServerSearchTestCase(IMAP4HelperMixin): """ Tests for the behavior of the search_* functions in L{imap5.IMAP4Server}. """ |