diff options
Diffstat (limited to 'src/leap/mail/imap/tests')
| -rw-r--r-- | src/leap/mail/imap/tests/__init__.py | 15 | ||||
| -rw-r--r-- | src/leap/mail/imap/tests/test_imap.py | 955 |
2 files changed, 639 insertions, 331 deletions
diff --git a/src/leap/mail/imap/tests/__init__.py b/src/leap/mail/imap/tests/__init__.py index 9a4c663..315d649 100644 --- a/src/leap/mail/imap/tests/__init__.py +++ b/src/leap/mail/imap/tests/__init__.py @@ -48,18 +48,19 @@ class BaseSoledadIMAPTest(BaseLeapTest): document_factory=LeapDocument) self._db2 = u1db.open(self.db2_file, create=True, document_factory=LeapDocument) + # initialize soledad by hand so we can control keys self._soledad = Soledad(self.email, gnupg_home=self.gnupg_home, - initialize=False, + bootstrap=False, prefix=self.tempdir) self._soledad._init_dirs() self._soledad._gpg = GPGWrapper(gnupghome=self.gnupg_home) - self._soledad._gpg.import_keys(PUBLIC_KEY) - self._soledad._gpg.import_keys(PRIVATE_KEY) - self._soledad._load_openpgp_keypair() - if not self._soledad._has_secret(): - self._soledad._gen_secret() - self._soledad._load_secret() + + if not self._soledad._has_privkey(): + self._soledad._set_privkey(PRIVATE_KEY) + if not self._soledad._has_symkey(): + self._soledad._gen_symkey() + self._soledad._load_symkey() self._soledad._init_db() def tearDown(self): diff --git a/src/leap/mail/imap/tests/test_imap.py b/src/leap/mail/imap/tests/test_imap.py index 7bfa1d7..6b6c24e 100644 --- a/src/leap/mail/imap/tests/test_imap.py +++ b/src/leap/mail/imap/tests/test_imap.py @@ -1,37 +1,54 @@ -#-*- encoding: utf-8 -*- +# -*- coding: utf-8 -*- +# test_imap.py +# Copyright (C) 2013 LEAP +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. """ -leap/email/imap/tests/test_imap.py ----------------------------------- Test case for leap.email.imap.server +TestCases taken from twisted tests and modified to make them work +against SoledadBackedAccount. @authors: Kali Kaneko, <kali@leap.se> +XXX add authors from the original twisted tests. + @license: GPLv3, see included LICENSE file -@copyright: © 2013 Kali Kaneko, see COPYLEFT file """ +# XXX review license of the original tests!!! try: from cStringIO import StringIO except ImportError: from StringIO import StringIO -import codecs -import locale +#import codecs +#import locale import os import types import tempfile import shutil -from zope.interface import implements +#from zope.interface import implements -from twisted.mail.imap4 import MessageSet +#from twisted.mail.imap4 import MessageSet from twisted.mail import imap4 from twisted.protocols import loopback from twisted.internet import defer -from twisted.internet import error -from twisted.internet import reactor -from twisted.internet import interfaces -from twisted.internet.task import Clock +#from twisted.internet import error +#from twisted.internet import reactor +#from twisted.internet import interfaces +#from twisted.internet.task import Clock from twisted.trial import unittest from twisted.python import util, log from twisted.python import failure @@ -42,19 +59,20 @@ import twisted.cred.checkers import twisted.cred.credentials import twisted.cred.portal -from twisted.test.proto_helpers import StringTransport, StringTransportWithDisconnection +#from twisted.test.proto_helpers import StringTransport, StringTransportWithDisconnection -import u1db +#import u1db from leap.common.testing.basetest import BaseLeapTest from leap.mail.imap.server import SoledadMailbox -from leap.mail.tests.imap import PUBLIC_KEY -from leap.mail.tests.imap import PRIVATE_KEY +from leap.mail.imap.server import SoledadBackedAccount +from leap.mail.imap.server import MessageCollection +#from leap.mail.imap.tests import PUBLIC_KEY +#from leap.mail.imap.tests import PRIVATE_KEY from leap.soledad import Soledad -from leap.soledad.util import GPGWrapper -from leap.soledad.backends.leap_backend import LeapDocument +from leap.soledad import SoledadCrypto def strip(f): @@ -74,57 +92,61 @@ def sortNest(l): def initialize_soledad(email, gnupg_home, tempdir): """ - initializes soledad by hand + Initializes soledad by hand + + @param email: ID for the user + @param gnupg_home: path to home used by gnupg + @param tempdir: path to temporal dir + @rtype: Soledad instance """ - _soledad = Soledad(email, gnupg_home=gnupg_home, - initialize=False, - prefix=tempdir) + + uuid = "foobar-uuid" + passphrase = "verysecretpassphrase" + secret_path = os.path.join(tempdir, "secret.gpg") + local_db_path = os.path.join(tempdir, "soledad.u1db") + server_url = "http://provider" + cert_file = "" + + _soledad = Soledad( + uuid, # user's uuid, obtained through signal events + passphrase, # how to get this? + secret_path, # how to get this? + local_db_path, # how to get this? + server_url, # can be None for now + cert_file, + bootstrap=False) _soledad._init_dirs() - _soledad._gpg = GPGWrapper(gnupghome=gnupg_home) - _soledad._gpg.import_keys(PUBLIC_KEY) - _soledad._gpg.import_keys(PRIVATE_KEY) - _soledad._load_openpgp_keypair() - if not _soledad._has_secret(): - _soledad._gen_secret() - _soledad._load_secret() + _soledad._crypto = SoledadCrypto(_soledad) + _soledad._shared_db = None + _soledad._init_keys() _soledad._init_db() + return _soledad ########################################## -# account, simpleserver +# Simple LEAP IMAP4 Server for testing ########################################## +class SimpleLEAPServer(imap4.IMAP4Server): + """ + A Simple IMAP4 Server with mailboxes backed by Soledad. -class SoledadBackedAccount(imap4.MemoryAccount): - #mailboxFactory = SimpleMailbox - mailboxFactory = SoledadMailbox - soledadInstance = None - - # XXX should reimplement IAccount -> SoledadAccount - # and receive the soledad instance on the constructor. - # SoledadMailbox should allow to filter by mailbox name - # _soledad db should include mailbox field - # and a document with "INDEX" info (mailboxes / subscriptions) - - def _emptyMailbox(self, name, id): - return self.mailboxFactory(self.soledadInstance) - - def select(self, name, rw=1): - # XXX rethink this. - # Need to be classmethods... - mbox = imap4.MemoryAccount.select(self, name) - if mbox is not None: - mbox.rw = rw - return mbox + This should be pretty close to the real LeapIMAP4Server that we + will be instantiating as a service, minus the authentication bits. + """ + def __init__(self, *args, **kw): + soledad = kw.pop('soledad', None) -class SimpleLEAPServer(imap4.IMAP4Server): - def __init__(self, *args, **kw): imap4.IMAP4Server.__init__(self, *args, **kw) realm = TestRealm() - realm.theAccount = SoledadBackedAccount('testuser') - # XXX soledadInstance here? + + # XXX Why I AM PASSING THE ACCOUNT TO + # REALM? I AM NOT USING THAT NOW, AM I??? + realm.theAccount = SoledadBackedAccount( + 'testuser', + soledad=soledad) portal = cred.portal.Portal(realm) c = cred.checkers.InMemoryUsernamePasswordDatabaseDontUse() @@ -150,17 +172,25 @@ class SimpleLEAPServer(imap4.IMAP4Server): class TestRealm: + """ + A minimal auth realm for testing purposes only + """ theAccount = None def requestAvatar(self, avatarId, mind, *interfaces): return imap4.IAccount, self.theAccount, lambda: None -###################### -# Test LEAP Server -###################### + +###################################### +# Simple IMAP4 Client for testing +###################################### class SimpleClient(imap4.IMAP4Client): + """ + A Simple IMAP4 Client to test our + Soledad-LEAPServer + """ def __init__(self, deferred, contextFactory=None): imap4.IMAP4Client.__init__(self, contextFactory) @@ -184,12 +214,28 @@ class SimpleClient(imap4.IMAP4Client): class IMAP4HelperMixin(BaseLeapTest): + """ + MixIn containing several utilities to be shared across + different TestCases + """ serverCTX = None clientCTX = None @classmethod def setUpClass(cls): + """ + TestCase initialization setup. + Sets up a new environment. + Initializes a SINGLE Soledad Instance that will be shared + by all tests in this base class. + This breaks orthogonality, avoiding us to use trial, so we should + move away from this test design. But it's a quick way to get + started without knowing / mocking the soledad api. + + We do also some duplication with BaseLeapTest cause trial and nose + seem not to deal well with deriving classmethods. + """ cls.old_path = os.environ['PATH'] cls.old_home = os.environ['HOME'] cls.tempdir = tempfile.mkdtemp(prefix="leap_tests-") @@ -217,10 +263,20 @@ class IMAP4HelperMixin(BaseLeapTest): cls.gnupg_home, cls.tempdir) - cls.sm = SoledadMailbox(soledad=cls._soledad) + # now we're passing the mailbox name, so we + # should get this into a partial or something. + #cls.sm = SoledadMailbox("mailbox", soledad=cls._soledad) + # XXX REFACTOR --- self.server (in setUp) is initializing + # a SoledadBackedAccount @classmethod def tearDownClass(cls): + """ + TestCase teardown method. + + Restores the old path and home environment variables. + Removes the temporal dir created for tests. + """ #cls._db1.close() #cls._db2.close() cls._soledad.close() @@ -232,33 +288,79 @@ class IMAP4HelperMixin(BaseLeapTest): shutil.rmtree(cls.tempdir) def setUp(self): + """ + Setup method for each test. + + Initializes and run a LEAP IMAP4 Server, + but passing the same Soledad instance (it's costly to initialize), + so we have to be sure to restore state across tests. + """ d = defer.Deferred() - self.server = SimpleLEAPServer(contextFactory=self.serverCTX) + self.server = SimpleLEAPServer( + contextFactory=self.serverCTX, + # XXX do we really need this?? + soledad=self._soledad) + self.client = SimpleClient(d, contextFactory=self.clientCTX) self.connected = d - theAccount = SoledadBackedAccount('testuser') - theAccount.soledadInstance = self._soledad + # XXX REVIEW-ME. + # We're adding theAccount here to server + # but it was also passed to initialization + # as it was passed to realm. + # I THINK we ONLY need to do it at one place now. - # XXX used for something??? - #theAccount.mboxType = SoledadMailbox + theAccount = SoledadBackedAccount( + 'testuser', + soledad=self._soledad) SimpleLEAPServer.theAccount = theAccount + # in case we get something from previous tests... + for mb in self.server.theAccount.mailboxes: + self.server.theAccount.delete(mb) + def tearDown(self): + """ + tearDown method called after each test. + + Deletes all documents in the Index, and deletes + instances of server and client. + """ self.delete_all_docs() + acct = self.server.theAccount + for mb in acct.mailboxes: + acct.delete(mb) + + # FIXME add again + #for subs in acct.subscriptions: + #acct.unsubscribe(subs) + del self.server del self.client del self.connected def populateMessages(self): - self._soledad.messages.add_msg(subject="test1") - self._soledad.messages.add_msg(subject="test2") - self._soledad.messages.add_msg(subject="test3") + """ + Populates soledad instance with several simple messages + """ + # XXX we should encapsulate this thru SoledadBackedAccount + # instead. + + # XXX we also should put this in a mailbox! + + self._soledad.messages.add_msg('', subject="test1") + self._soledad.messages.add_msg('', subject="test2") + self._soledad.messages.add_msg('', subject="test3") # XXX should change Flags too - self._soledad.messages.add_msg(subject="test4") + self._soledad.messages.add_msg('', subject="test4") def delete_all_docs(self): - self.server.theAccount.messages.deleteAllDocs() + """ + Deletes all the docs in the testing instance of the + SoledadBackedAccount. + """ + self.server.theAccount.deleteAllMessages( + iknowhatiamdoing=True) def _cbStopClient(self, ignore): self.client.transport.loseConnection() @@ -272,206 +374,83 @@ class IMAP4HelperMixin(BaseLeapTest): return loopback.loopbackAsync(self.server, self.client) -class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): - - def testCapability(self): - caps = {} - - def getCaps(): - def gotCaps(c): - caps.update(c) - self.server.transport.loseConnection() - return self.client.getCapabilities().addCallback(gotCaps) - d1 = self.connected.addCallback( - strip(getCaps)).addErrback(self._ebGeneral) - d = defer.gatherResults([self.loopback(), d1]) - expected = {'IMAP4rev1': None, 'NAMESPACE': None, 'IDLE': None} - - return d.addCallback(lambda _: self.assertEqual(expected, caps)) - - def testCapabilityWithAuth(self): - caps = {} - self.server.challengers[ - 'CRAM-MD5'] = cred.credentials.CramMD5Credentials - - def getCaps(): - def gotCaps(c): - caps.update(c) - self.server.transport.loseConnection() - return self.client.getCapabilities().addCallback(gotCaps) - d1 = self.connected.addCallback( - strip(getCaps)).addErrback(self._ebGeneral) - d = defer.gatherResults([self.loopback(), d1]) - - expCap = {'IMAP4rev1': None, 'NAMESPACE': None, - 'IDLE': None, 'AUTH': ['CRAM-MD5']} - - return d.addCallback(lambda _: self.assertEqual(expCap, caps)) - - def testLogout(self): - self.loggedOut = 0 - - def logout(): - def setLoggedOut(): - self.loggedOut = 1 - self.client.logout().addCallback(strip(setLoggedOut)) - self.connected.addCallback(strip(logout)).addErrback(self._ebGeneral) - d = self.loopback() - return d.addCallback(lambda _: self.assertEqual(self.loggedOut, 1)) - - def testNoop(self): - self.responses = None - - def noop(): - def setResponses(responses): - self.responses = responses - self.server.transport.loseConnection() - self.client.noop().addCallback(setResponses) - self.connected.addCallback(strip(noop)).addErrback(self._ebGeneral) - d = self.loopback() - return d.addCallback(lambda _: self.assertEqual(self.responses, [])) - - def testLogin(self): - def login(): - d = self.client.login('testuser', 'password-test') - d.addCallback(self._cbStopClient) - d1 = self.connected.addCallback(strip(login)).addErrback(self._ebGeneral) - d = defer.gatherResults([d1, self.loopback()]) - return d.addCallback(self._cbTestLogin) - - def _cbTestLogin(self, ignored): - self.assertEqual(self.server.account, SimpleLEAPServer.theAccount) - self.assertEqual(self.server.state, 'auth') - - def testFailedLogin(self): - def login(): - d = self.client.login('testuser', 'wrong-password') - d.addBoth(self._cbStopClient) - - d1 = self.connected.addCallback(strip(login)).addErrback(self._ebGeneral) - d2 = self.loopback() - d = defer.gatherResults([d1, d2]) - return d.addCallback(self._cbTestFailedLogin) - - def _cbTestFailedLogin(self, ignored): - self.assertEqual(self.server.account, None) - self.assertEqual(self.server.state, 'unauth') - - - def testLoginRequiringQuoting(self): - self.server._username = '{test}user' - self.server._password = '{test}password' - - def login(): - d = self.client.login('{test}user', '{test}password') - d.addBoth(self._cbStopClient) - - d1 = self.connected.addCallback(strip(login)).addErrback(self._ebGeneral) - d = defer.gatherResults([self.loopback(), d1]) - return d.addCallback(self._cbTestLoginRequiringQuoting) - - def _cbTestLoginRequiringQuoting(self, ignored): - self.assertEqual(self.server.account, SimpleLEAPServer.theAccount) - self.assertEqual(self.server.state, 'auth') - - - def testNamespace(self): - self.namespaceArgs = None - def login(): - return self.client.login('testuser', 'password-test') - def namespace(): - def gotNamespace(args): - self.namespaceArgs = args - self._cbStopClient(None) - return self.client.namespace().addCallback(gotNamespace) - - d1 = self.connected.addCallback(strip(login)) - d1.addCallback(strip(namespace)) - d1.addErrback(self._ebGeneral) - d2 = self.loopback() - d = defer.gatherResults([d1, d2]) - d.addCallback(lambda _: self.assertEqual(self.namespaceArgs, - [[['', '/']], [], []])) - return d - - def testSelect(self): - SimpleLEAPServer.theAccount.addMailbox('test-mailbox') - self.selectedArgs = None - - def login(): - return self.client.login('testuser', 'password-test') - - def select(): - def selected(args): - self.selectedArgs = args - self._cbStopClient(None) - d = self.client.select('test-mailbox') - d.addCallback(selected) - return d +# +# TestCases +# - d1 = self.connected.addCallback(strip(login)) - d1.addCallback(strip(select)) - d1.addErrback(self._ebGeneral) - d2 = self.loopback() - return defer.gatherResults([d1, d2]).addCallback(self._cbTestSelect) - - def _cbTestSelect(self, ignored): - mbox = SimpleLEAPServer.theAccount.mailboxes['TEST-MAILBOX'] - self.assertEqual(self.server.mbox, mbox) - self.assertEqual(self.selectedArgs, { - 'EXISTS': 9, 'RECENT': 3, 'UIDVALIDITY': 42, - 'FLAGS': ('\\Seen', '\\Answered', '\\Flagged', - '\\Deleted', '\\Draft', '\\Recent', 'List'), - 'READ-WRITE': 1 - }) - - def test_examine(self): +class MessageCollectionTestCase(IMAP4HelperMixin, unittest.TestCase): + """ + Tests for the MessageCollection class + """ + def setUp(self): """ - L{IMAP4Client.examine} issues an I{EXAMINE} command to the server and - returns a L{Deferred} which fires with a C{dict} with as many of the - following keys as the server includes in its response: C{'FLAGS'}, - C{'EXISTS'}, C{'RECENT'}, C{'UNSEEN'}, C{'READ-WRITE'}, C{'READ-ONLY'}, - C{'UIDVALIDITY'}, and C{'PERMANENTFLAGS'}. - - Unfortunately the server doesn't generate all of these so it's hard to - test the client's handling of them here. See - L{IMAP4ClientExamineTests} below. - - See U{RFC 3501<http://www.faqs.org/rfcs/rfc3501.html>}, section 6.3.2, - for details. + setUp method for each test + We override mixin method since we are only testing + MessageCollection interface in this particular TestCase """ - SimpleLEAPServer.theAccount.addMailbox('test-mailbox') - self.examinedArgs = None - - def login(): - return self.client.login('testuser', 'password-test') + self.messages = MessageCollection("testmbox", self._soledad._db) - def examine(): - def examined(args): - self.examinedArgs = args - self._cbStopClient(None) - d = self.client.examine('test-mailbox') - d.addCallback(examined) - return d + def tearDown(self): + """ + tearDown method for each test + Delete the message collection + """ + del self.messages - d1 = self.connected.addCallback(strip(login)) - d1.addCallback(strip(examine)) - d1.addErrback(self._ebGeneral) - d2 = self.loopback() - d = defer.gatherResults([d1, d2]) - return d.addCallback(self._cbTestExamine) + def testEmptyMessage(self): + """ + Test empty message and collection + """ + em = self.messages.get_empty_msg() + self.assertEqual(em, + {"subject": "", "seen": False, + "flags": [], "mailbox": "inbox", + "mbox-uid": 1, + "raw": ""}) + self.assertEqual(self.messages.count(), 0) + + def testFilterByMailbox(self): + """ + Test that queries filter by selected mailbox + """ + mc = self.messages + mc.add_msg('', subject="test1") + mc.add_msg('', subject="test2") + mc.add_msg('', subject="test3") + self.assertEqual(self.messages.count(), 3) + + newmsg = mc.get_empty_msg() + newmsg['mailbox'] = "mailbox/foo" + newmsg['subject'] = "test another mailbox" + mc.db.create_doc(newmsg) + self.assertEqual(mc.count(), 3) + self.assertEqual(len(mc.db.get_from_index(mc.MAILBOX_INDEX, "*")), + 4) + + +class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): + """ + Tests for the generic behavior of the LeapIMAP4Server + which, right now, it's just implemented in this test file as + SimpleLEAPServer. We will move the implementation, together with + authentication bits, to leap.mail.imap.server so it can be instantiated + from the tac file. + + Right now this TestCase tries to mimmick as close as possible the + organization from the twisted.mail.imap tests so we can achieve + a complete implementation. The order in which they appear reflect + the intended order of implementation. + """ - def _cbTestExamine(self, ignored): - mbox = SimpleLEAPServer.theAccount.mailboxes['TEST-MAILBOX'] - self.assertEqual(self.server.mbox, mbox) - self.assertEqual(self.examinedArgs, { - 'EXISTS': 9, 'RECENT': 3, 'UIDVALIDITY': 42, - 'FLAGS': ('\\Seen', '\\Answered', '\\Flagged', - '\\Deleted', '\\Draft', '\\Recent', 'List'), - 'READ-WRITE': False}) + # + # mailboxes operations + # def testCreate(self): - succeed = ('testbox', 'test/box', 'test/', 'test/box/box', 'INBOX') + """ + Test whether we can create mailboxes + """ + succeed = ('testbox', 'test/box', 'test/', 'test/box/box', 'FOOBOX') fail = ('testbox', 'test/box') def cb(): @@ -498,13 +477,17 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def _cbTestCreate(self, ignored, succeed, fail): self.assertEqual(self.result, [1] * len(succeed) + [0] * len(fail)) - mbox = SimpleLEAPServer.theAccount.mailboxes.keys() - answers = ['inbox', 'testbox', 'test/box', 'test', 'test/box/box'] + + mbox = SimpleLEAPServer.theAccount.mailboxes + answers = ['foobox', 'testbox', 'test/box', 'test', 'test/box/box'] mbox.sort() answers.sort() self.assertEqual(mbox, [a.upper() for a in answers]) def testDelete(self): + """ + Test whether we can delete mailboxes + """ SimpleLEAPServer.theAccount.addMailbox('delete/me') def login(): @@ -518,11 +501,16 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): d1.addCallbacks(self._cbStopClient, self._ebGeneral) d2 = self.loopback() d = defer.gatherResults([d1, d2]) - d.addCallback(lambda _: - self.assertEqual(SimpleLEAPServer.theAccount.mailboxes.keys(), [])) + d.addCallback( + lambda _: self.assertEqual( + SimpleLEAPServer.theAccount.mailboxes, [])) return d def testIllegalInboxDelete(self): + """ + Test what happens if we try to delete the user Inbox. + We expect that operation to fail. + """ self.stashed = None def login(): @@ -545,12 +533,16 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return d def testNonExistentDelete(self): - + """ + Test what happens if we try to delete a non-existent mailbox. + We expect an error raised stating 'No such inbox' + """ def login(): return self.client.login('testuser', 'password-test') def delete(): return self.client.delete('delete/me') + self.failure = failure def deleteFailed(failure): self.failure = failure @@ -562,13 +554,17 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): d2 = self.loopback() d = defer.gatherResults([d1, d2]) d.addCallback(lambda _: self.assertEqual(str(self.failure.value), - 'No such mailbox')) + 'No such mailbox')) return d def testIllegalDelete(self): - m = SoledadMailbox() - m.flags = (r'\Noselect',) - SimpleLEAPServer.theAccount.addMailbox('delete', m) + """ + Try deleting a mailbox with sub-folders, and \NoSelect flag set. + An exception is expected + """ + SimpleLEAPServer.theAccount.addMailbox('delete') + to_delete = SimpleLEAPServer.theAccount.getMailbox('delete') + to_delete.setFlags((r'\Noselect',)) SimpleLEAPServer.theAccount.addMailbox('delete/me') def login(): @@ -593,6 +589,9 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return d def testRename(self): + """ + Test whether we can rename a mailbox + """ SimpleLEAPServer.theAccount.addMailbox('oldmbox') def login(): @@ -608,11 +607,15 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): d = defer.gatherResults([d1, d2]) d.addCallback(lambda _: self.assertEqual( - SimpleLEAPServer.theAccount.mailboxes.keys(), - ['NEWNAME'])) + SimpleLEAPServer.theAccount.mailboxes, + ['NEWNAME'])) return d def testIllegalInboxRename(self): + """ + Try to rename inbox. We expect it to fail. Then it would be not + an inbox anymore, would it? + """ self.stashed = None def login(): @@ -632,10 +635,13 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): d = defer.gatherResults([d1, d2]) d.addCallback(lambda _: self.failUnless(isinstance( - self.stashed, failure.Failure))) + self.stashed, failure.Failure))) return d def testHierarchicalRename(self): + """ + Try to rename hierarchical mailboxes + """ SimpleLEAPServer.theAccount.create('oldmbox/m1') SimpleLEAPServer.theAccount.create('oldmbox/m2') @@ -653,13 +659,15 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return d.addCallback(self._cbTestHierarchicalRename) def _cbTestHierarchicalRename(self, ignored): - mboxes = SimpleLEAPServer.theAccount.mailboxes.keys() + mboxes = SimpleLEAPServer.theAccount.mailboxes expected = ['newname', 'newname/m1', 'newname/m2'] mboxes.sort() self.assertEqual(mboxes, [s.upper() for s in expected]) def testSubscribe(self): - + """ + Test whether we can mark a mailbox as subscribed to + """ def login(): return self.client.login('testuser', 'password-test') @@ -672,14 +680,21 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): d2 = self.loopback() d = defer.gatherResults([d1, d2]) d.addCallback(lambda _: - self.assertEqual(SimpleLEAPServer.theAccount.subscriptions, - ['THIS/MBOX'])) + self.assertEqual( + SimpleLEAPServer.theAccount.subscriptions, + ['THIS/MBOX'])) return d def testUnsubscribe(self): - SimpleLEAPServer.theAccount.subscriptions = ['THIS/MBOX', 'THAT/MBOX'] + """ + Test whether we can unsubscribe from a set of mailboxes + """ + SimpleLEAPServer.theAccount.subscribe('THIS/MBOX') + SimpleLEAPServer.theAccount.subscribe('THAT/MBOX') + def login(): return self.client.login('testuser', 'password-test') + def unsubscribe(): return self.client.unsubscribe('this/mbox') @@ -689,14 +704,255 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): d2 = self.loopback() d = defer.gatherResults([d1, d2]) d.addCallback(lambda _: - self.assertEqual(SimpleLEAPServer.theAccount.subscriptions, - ['THAT/MBOX'])) + self.assertEqual( + SimpleLEAPServer.theAccount.subscriptions, + ['THAT/MBOX'])) + return d + + def testSelect(self): + """ + Try to select a mailbox + """ + self.server.theAccount.addMailbox('TESTMAILBOX-SELECT', creation_ts=42) + self.selectedArgs = None + + def login(): + return self.client.login('testuser', 'password-test') + + def select(): + def selected(args): + self.selectedArgs = args + self._cbStopClient(None) + d = self.client.select('TESTMAILBOX-SELECT') + d.addCallback(selected) + return d + + d1 = self.connected.addCallback(strip(login)) + d1.addCallback(strip(select)) + d1.addErrback(self._ebGeneral) + + d2 = self.loopback() + return defer.gatherResults([d1, d2]).addCallback(self._cbTestSelect) + + def _cbTestSelect(self, ignored): + mbox = SimpleLEAPServer.theAccount.getMailbox('TESTMAILBOX-SELECT') + self.assertEqual(self.server.mbox.messages.mbox, mbox.messages.mbox) + self.assertEqual(self.selectedArgs, { + 'EXISTS': 0, 'RECENT': 0, 'UIDVALIDITY': 42, + 'FLAGS': ('\\Seen', '\\Answered', '\\Flagged', + '\\Deleted', '\\Draft', '\\Recent', 'List'), + 'READ-WRITE': True + }) + + # + # capabilities + # + + def testCapability(self): + caps = {} + + def getCaps(): + def gotCaps(c): + caps.update(c) + self.server.transport.loseConnection() + return self.client.getCapabilities().addCallback(gotCaps) + d1 = self.connected.addCallback( + strip(getCaps)).addErrback(self._ebGeneral) + d = defer.gatherResults([self.loopback(), d1]) + expected = {'IMAP4rev1': None, 'NAMESPACE': None, 'IDLE': None} + + return d.addCallback(lambda _: self.assertEqual(expected, caps)) + + def testCapabilityWithAuth(self): + caps = {} + self.server.challengers[ + 'CRAM-MD5'] = cred.credentials.CramMD5Credentials + + def getCaps(): + def gotCaps(c): + caps.update(c) + self.server.transport.loseConnection() + return self.client.getCapabilities().addCallback(gotCaps) + d1 = self.connected.addCallback( + strip(getCaps)).addErrback(self._ebGeneral) + + d = defer.gatherResults([self.loopback(), d1]) + + expCap = {'IMAP4rev1': None, 'NAMESPACE': None, + 'IDLE': None, 'AUTH': ['CRAM-MD5']} + + return d.addCallback(lambda _: self.assertEqual(expCap, caps)) + + # + # authentication + # + + def testLogout(self): + """ + Test log out + """ + self.loggedOut = 0 + + def logout(): + def setLoggedOut(): + self.loggedOut = 1 + self.client.logout().addCallback(strip(setLoggedOut)) + self.connected.addCallback(strip(logout)).addErrback(self._ebGeneral) + d = self.loopback() + return d.addCallback(lambda _: self.assertEqual(self.loggedOut, 1)) + + def testNoop(self): + """ + Test noop command + """ + self.responses = None + + def noop(): + def setResponses(responses): + self.responses = responses + self.server.transport.loseConnection() + self.client.noop().addCallback(setResponses) + self.connected.addCallback(strip(noop)).addErrback(self._ebGeneral) + d = self.loopback() + return d.addCallback(lambda _: self.assertEqual(self.responses, [])) + + def testLogin(self): + """ + Test login + """ + def login(): + d = self.client.login('testuser', 'password-test') + d.addCallback(self._cbStopClient) + d1 = self.connected.addCallback( + strip(login)).addErrback(self._ebGeneral) + d = defer.gatherResults([d1, self.loopback()]) + return d.addCallback(self._cbTestLogin) + + def _cbTestLogin(self, ignored): + self.assertEqual(self.server.account, SimpleLEAPServer.theAccount) + self.assertEqual(self.server.state, 'auth') + + def testFailedLogin(self): + """ + Test bad login + """ + def login(): + d = self.client.login('testuser', 'wrong-password') + d.addBoth(self._cbStopClient) + + d1 = self.connected.addCallback( + strip(login)).addErrback(self._ebGeneral) + d2 = self.loopback() + d = defer.gatherResults([d1, d2]) + return d.addCallback(self._cbTestFailedLogin) + + def _cbTestFailedLogin(self, ignored): + self.assertEqual(self.server.account, None) + self.assertEqual(self.server.state, 'unauth') + + def testLoginRequiringQuoting(self): + """ + Test login requiring quoting + """ + self.server._username = '{test}user' + self.server._password = '{test}password' + + def login(): + d = self.client.login('{test}user', '{test}password') + d.addBoth(self._cbStopClient) + + d1 = self.connected.addCallback( + strip(login)).addErrback(self._ebGeneral) + d = defer.gatherResults([self.loopback(), d1]) + return d.addCallback(self._cbTestLoginRequiringQuoting) + + def _cbTestLoginRequiringQuoting(self, ignored): + self.assertEqual(self.server.account, SimpleLEAPServer.theAccount) + self.assertEqual(self.server.state, 'auth') + + # + # Inspection + # + + def testNamespace(self): + """ + Test retrieving namespace + """ + self.namespaceArgs = None + + def login(): + return self.client.login('testuser', 'password-test') + + def namespace(): + def gotNamespace(args): + self.namespaceArgs = args + self._cbStopClient(None) + return self.client.namespace().addCallback(gotNamespace) + + d1 = self.connected.addCallback(strip(login)) + d1.addCallback(strip(namespace)) + d1.addErrback(self._ebGeneral) + d2 = self.loopback() + d = defer.gatherResults([d1, d2]) + d.addCallback(lambda _: self.assertEqual(self.namespaceArgs, + [[['', '/']], [], []])) return d + def testExamine(self): + """ + L{IMAP4Client.examine} issues an I{EXAMINE} command to the server and + returns a L{Deferred} which fires with a C{dict} with as many of the + following keys as the server includes in its response: C{'FLAGS'}, + C{'EXISTS'}, C{'RECENT'}, C{'UNSEEN'}, C{'READ-WRITE'}, C{'READ-ONLY'}, + C{'UIDVALIDITY'}, and C{'PERMANENTFLAGS'}. + + Unfortunately the server doesn't generate all of these so it's hard to + test the client's handling of them here. See + L{IMAP4ClientExamineTests} below. + + See U{RFC 3501<http://www.faqs.org/rfcs/rfc3501.html>}, section 6.3.2, + for details. + """ + self.server.theAccount.addMailbox('test-mailbox-e', + creation_ts=42) + #import ipdb; ipdb.set_trace() + + self.examinedArgs = None + + def login(): + return self.client.login('testuser', 'password-test') + + def examine(): + def examined(args): + self.examinedArgs = args + self._cbStopClient(None) + d = self.client.examine('test-mailbox-e') + d.addCallback(examined) + return d + + d1 = self.connected.addCallback(strip(login)) + d1.addCallback(strip(examine)) + d1.addErrback(self._ebGeneral) + d2 = self.loopback() + d = defer.gatherResults([d1, d2]) + 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) + self.assertEqual(self.examinedArgs, { + 'EXISTS': 0, 'RECENT': 0, 'UIDVALIDITY': 42, + 'FLAGS': ('\\Seen', '\\Answered', '\\Flagged', + '\\Deleted', '\\Draft', '\\Recent', 'List'), + 'READ-WRITE': False}) + def _listSetup(self, f): - SimpleLEAPServer.theAccount.addMailbox('root/subthing') - SimpleLEAPServer.theAccount.addMailbox('root/another-thing') - SimpleLEAPServer.theAccount.addMailbox('non-root/subthing') + SimpleLEAPServer.theAccount.addMailbox('root/subthingl', + creation_ts=42) + SimpleLEAPServer.theAccount.addMailbox('root/another-thing', + creation_ts=42) + SimpleLEAPServer.theAccount.addMailbox('non-root/subthing', + creation_ts=42) def login(): return self.client.login('testuser', 'password-test') @@ -713,37 +969,51 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return defer.gatherResults([d1, d2]).addCallback(lambda _: self.listed) def testList(self): + """ + Test List command + """ def list(): return self.client.list('root', '%') d = self._listSetup(list) d.addCallback(lambda listed: self.assertEqual( sortNest(listed), sortNest([ - (SoledadMailbox.flags, "/", "ROOT/SUBTHING"), - (SoledadMailbox.flags, "/", "ROOT/ANOTHER-THING") + (SoledadMailbox.INIT_FLAGS, "/", "ROOT/SUBTHINGL"), + (SoledadMailbox.INIT_FLAGS, "/", "ROOT/ANOTHER-THING") ]) )) return d + # XXX implement subscriptions + ''' def testLSub(self): - SimpleLEAPServer.theAccount.subscribe('ROOT/SUBTHING') + """ + Test LSub command + """ + SimpleLEAPServer.theAccount.subscribe('ROOT/SUBTHINGL') def lsub(): return self.client.lsub('root', '%') d = self._listSetup(lsub) d.addCallback(self.assertEqual, - [(SoledadMailbox.flags, "/", "ROOT/SUBTHING")]) + [(SoledadMailbox.INIT_FLAGS, "/", "ROOT/SUBTHINGL")]) return d + ''' def testStatus(self): - SimpleLEAPServer.theAccount.addMailbox('root/subthing') + """ + Test Status command + """ + SimpleLEAPServer.theAccount.addMailbox('root/subthings') + # XXX FIXME ---- should populate this a little bit, + # with unseen etc... def login(): return self.client.login('testuser', 'password-test') def status(): return self.client.status( - 'root/subthing', 'MESSAGES', 'UIDNEXT', 'UNSEEN') + 'root/subthings', 'MESSAGES', 'UIDNEXT', 'UNSEEN') def statused(result): self.statused = result @@ -757,11 +1027,14 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): d = defer.gatherResults([d1, d2]) d.addCallback(lambda _: self.assertEqual( self.statused, - {'MESSAGES': 9, 'UIDNEXT': '10', 'UNSEEN': 4} + {'MESSAGES': 0, 'UIDNEXT': '1', 'UNSEEN': 0} )) return d def testFailedStatus(self): + """ + Test failed status command with a non-existent mailbox + """ def login(): return self.client.login('testuser', 'password-test') @@ -793,7 +1066,14 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): ('Could not open mailbox',) ) + # + # messages + # + def testFullAppend(self): + """ + Test appending a full message to the mailbox + """ infile = util.sibpath(__file__, 'rfc822.message') message = open(infile) SimpleLEAPServer.theAccount.addMailbox('root/subthing') @@ -805,7 +1085,7 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return self.client.append( 'root/subthing', message, - ('\\SEEN', '\\DELETED'), + ['\\SEEN', '\\DELETED'], 'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)', ) @@ -817,15 +1097,24 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return d.addCallback(self._cbTestFullAppend, infile) def _cbTestFullAppend(self, ignored, infile): - mb = SimpleLEAPServer.theAccount.mailboxes['ROOT/SUBTHING'] + mb = SimpleLEAPServer.theAccount.getMailbox('ROOT/SUBTHING') self.assertEqual(1, len(mb.messages)) + + #import ipdb; ipdb.set_trace() self.assertEqual( - (['\\SEEN', '\\DELETED'], 'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)', 0), - mb.messages[0][1:] - ) - self.assertEqual(open(infile).read(), mb.messages[0][0].getvalue()) + ['\\SEEN', '\\DELETED'], + mb.messages[1]['flags']) + + self.assertEqual( + 'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)', + mb.messages[1]['date']) + + self.assertEqual(open(infile).read(), mb.messages[1]['raw']) def testPartialAppend(self): + """ + Test partially appending a message to the mailbox + """ infile = util.sibpath(__file__, 'rfc822.message') message = open(infile) SimpleLEAPServer.theAccount.addMailbox('PARTIAL/SUBTHING') @@ -838,7 +1127,8 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return self.client.sendCommand( imap4.Command( 'APPEND', - 'PARTIAL/SUBTHING (\\SEEN) "Right now" {%d}' % os.path.getsize(infile), + 'PARTIAL/SUBTHING (\\SEEN) "Right now" ' + '{%d}' % os.path.getsize(infile), (), self.client._IMAP4Client__cbContinueAppend, message ) ) @@ -850,15 +1140,20 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): return d.addCallback(self._cbTestPartialAppend, infile) def _cbTestPartialAppend(self, ignored, infile): - mb = SimpleLEAPServer.theAccount.mailboxes['PARTIAL/SUBTHING'] + mb = SimpleLEAPServer.theAccount.getMailbox('PARTIAL/SUBTHING') self.assertEqual(1, len(mb.messages)) self.assertEqual( - (['\\SEEN'], 'Right now', 0), - mb.messages[0][1:] + ['\\SEEN',], + mb.messages[1]['flags'] ) - self.assertEqual(open(infile).read(), mb.messages[0][0].getvalue()) + self.assertEqual( + 'Right now', mb.messages[1]['date']) + self.assertEqual(open(infile).read(), mb.messages[1]['raw']) def testCheck(self): + """ + Test check command + """ SimpleLEAPServer.theAccount.addMailbox('root/subthing') def login(): @@ -879,19 +1174,25 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): # Okay, that was fun def testClose(self): - m = SoledadMailbox() - m.messages = [ - ('Message 1', ('\\Deleted', 'AnotherFlag'), None, 0), - ('Message 2', ('AnotherFlag',), None, 1), - ('Message 3', ('\\Deleted',), None, 2), - ] - SimpleLEAPServer.theAccount.addMailbox('mailbox', m) + """ + Test closing the mailbox. We expect to get deleted all messages flagged + as such. + """ + name = 'mailbox-close' + self.server.theAccount.addMailbox(name) + #import ipdb; ipdb.set_trace() + + m = SimpleLEAPServer.theAccount.getMailbox(name) + m.messages.add_msg('', subject="Message 1", + flags=('\\Deleted', 'AnotherFlag')) + m.messages.add_msg('', subject="Message 2", flags=('AnotherFlag',)) + m.messages.add_msg('', subject="Message 3", flags=('\\Deleted',)) def login(): return self.client.login('testuser', 'password-test') def select(): - return self.client.select('mailbox') + return self.client.select(name) def close(): return self.client.close() @@ -905,24 +1206,29 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def _cbTestClose(self, ignored, m): self.assertEqual(len(m.messages), 1) - self.assertEqual(m.messages[0], - ('Message 2', ('AnotherFlag',), None, 1)) + self.assertEqual( + m.messages[1]['subject'], + 'Message 2') + self.failUnless(m.closed) def testExpunge(self): - m = SoledadMailbox() - m.messages = [ - ('Message 1', ('\\Deleted', 'AnotherFlag'), None, 0), - ('Message 2', ('AnotherFlag',), None, 1), - ('Message 3', ('\\Deleted',), None, 2), - ] - SimpleLEAPServer.theAccount.addMailbox('mailbox', m) + """ + Test expunge command + """ + name = 'mailbox-expunge' + SimpleLEAPServer.theAccount.addMailbox(name) + m = SimpleLEAPServer.theAccount.getMailbox(name) + m.messages.add_msg('', subject="Message 1", + flags=('\\Deleted', 'AnotherFlag')) + m.messages.add_msg('', subject="Message 2", flags=('AnotherFlag',)) + m.messages.add_msg('', subject="Message 3", flags=('\\Deleted',)) def login(): return self.client.login('testuser', 'password-test') def select(): - return self.client.select('mailbox') + return self.client.select('mailbox-expunge') def expunge(): return self.client.expunge() @@ -943,15 +1249,16 @@ class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def _cbTestExpunge(self, ignored, m): self.assertEqual(len(m.messages), 1) - self.assertEqual(m.messages[0], - ('Message 2', ('AnotherFlag',), None, 1)) - - self.assertEqual(self.results, [0, 2]) - + self.assertEqual( + m.messages[1]['subject'], + 'Message 2') + self.assertEqual(self.results, [0, 1]) + # XXX fix this thing with the indexes... class IMAP4ServerSearchTestCase(IMAP4HelperMixin, unittest.TestCase): """ Tests for the behavior of the search_* functions in L{imap4.IMAP4Server}. """ + # XXX coming soon to your screens! pass |