diff options
| -rw-r--r-- | src/leap/mail/imap/tests/test_imap.py | 318 | ||||
| -rw-r--r-- | src/leap/mail/imap/tests/test_imap_store_fetch.py | 71 | ||||
| -rw-r--r-- | src/leap/mail/imap/tests/utils.py | 225 |
3 files changed, 317 insertions, 297 deletions
diff --git a/src/leap/mail/imap/tests/test_imap.py b/src/leap/mail/imap/tests/test_imap.py index bad8a5b..980e46e 100644 --- a/src/leap/mail/imap/tests/test_imap.py +++ b/src/leap/mail/imap/tests/test_imap.py @@ -25,7 +25,6 @@ XXX add authors from the original twisted tests. @license: GPLv3, see included LICENSE file """ # XXX review license of the original tests!!! -from email import parser try: from cStringIO import StringIO @@ -34,13 +33,9 @@ except ImportError: import os import types -import tempfile -import shutil -from mock import Mock from twisted.mail import imap4 -from twisted.protocols import loopback from twisted.internet import defer from twisted.trial import unittest from twisted.python import util @@ -51,14 +46,12 @@ from twisted import cred # import u1db -from leap.common.testing.basetest import BaseLeapTest -from leap.mail.imap.account import SoledadBackedAccount 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.tests.utils import IMAP4HelperMixin -from leap.soledad.client import Soledad TEST_USER = "testuser@leap.se" TEST_PASSWD = "1234" @@ -79,46 +72,6 @@ def sortNest(l): return l -def initialize_soledad(email, gnupg_home, tempdir): - """ - 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 - """ - - uuid = "foobar-uuid" - passphrase = u"verysecretpassphrase" - secret_path = os.path.join(tempdir, "secret.gpg") - local_db_path = os.path.join(tempdir, "soledad.u1db") - server_url = "http://provider" - cert_file = "" - - class MockSharedDB(object): - - get_doc = Mock(return_value=None) - put_doc = Mock() - lock = Mock(return_value=('atoken', 300)) - unlock = Mock(return_value=True) - - def __call__(self): - return self - - Soledad._shared_db = MockSharedDB() - - _soledad = Soledad( - uuid, - passphrase, - secret_path, - local_db_path, - server_url, - cert_file) - - return _soledad - - class TestRealm: """ @@ -131,171 +84,6 @@ class TestRealm: # -# 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) - self.deferred = deferred - self.events = [] - - def serverGreeting(self, caps): - self.deferred.callback(None) - - def modeChanged(self, writeable): - self.events.append(['modeChanged', writeable]) - self.transport.loseConnection() - - def flagsChanged(self, newFlags): - self.events.append(['flagsChanged', newFlags]) - self.transport.loseConnection() - - def newMessages(self, exists, recent): - self.events.append(['newMessages', exists, recent]) - self.transport.loseConnection() - - -class IMAP4HelperMixin(BaseLeapTest): - - """ - MixIn containing several utilities to be shared across - different TestCases - """ - - serverCTX = None - clientCTX = None - - # setUpClass cannot be a classmethod in trial, see: - # https://twistedmatrix.com/trac/ticket/1870 - - - 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. - """ - self.old_path = os.environ['PATH'] - self.old_home = os.environ['HOME'] - self.tempdir = tempfile.mkdtemp(prefix="leap_tests-") - self.home = self.tempdir - bin_tdir = os.path.join( - self.tempdir, - 'bin') - os.environ["PATH"] = bin_tdir - os.environ["HOME"] = self.tempdir - - # Soledad: config info - self.gnupg_home = "%s/gnupg" % self.tempdir - self.email = 'leap@leap.se' - - # initialize soledad by hand so we can control keys - self._soledad = initialize_soledad( - self.email, - self.gnupg_home, - self.tempdir) - UUID = 'deadbeef', - USERID = TEST_USER - memstore = MemoryStore() - - ########### - - d = defer.Deferred() - self.server = LeapIMAPServer( - uuid=UUID, userid=USERID, - contextFactory=self.serverCTX, - # XXX do we really need this?? - soledad=self._soledad) - - self.client = SimpleClient(d, contextFactory=self.clientCTX) - self.connected = d - - # 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. - - theAccount = SoledadBackedAccount( - USERID, - soledad=self._soledad, - memstore=memstore) - LeapIMAPServer.theAccount = theAccount - - # in case we get something from previous tests... - for mb in self.server.theAccount.mailboxes: - self.server.theAccount.delete(mb) - - # email parser - self.parser = parser.Parser() - - def tearDown(self): - """ - tearDown method called after each test. - - Deletes all documents in the Index, and deletes - instances of server and client. - """ - try: - self._soledad.close() - os.environ["PATH"] = self.old_path - os.environ["HOME"] = self.old_home - # safety check - assert 'leap_tests-' in self.tempdir - shutil.rmtree(self.tempdir) - except Exception: - print "ERROR WHILE CLOSING SOLEDAD" - - def populateMessages(self): - """ - 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('', uid=1, subject="test1") - self._soledad.messages.add_msg('', uid=2, subject="test2") - self._soledad.messages.add_msg('', uid=3, subject="test3") - # XXX should change Flags too - self._soledad.messages.add_msg('', uid=4, subject="test4") - - def delete_all_docs(self): - """ - 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() - - def _ebGeneral(self, failure): - self.client.transport.loseConnection() - self.server.transport.loseConnection() - # can we do something similar? - # I guess this was ok with trial, but not in noseland... - # log.err(failure, "Problem with %r" % (self.function,)) - raise failure.value - # failure.trap(Exception) - - def loopback(self): - return loopback.loopbackAsync(self.server, self.client) - - -# # TestCases # @@ -353,17 +141,17 @@ class MessageCollectionTestCase(IMAP4HelperMixin, unittest.TestCase): def add_first(): d = defer.gatherResults([ - mc.add_msg('Stuff 1', uid=1, subject="test1"), - mc.add_msg('Stuff 2', uid=2, subject="test2"), - mc.add_msg('Stuff 3', uid=3, subject="test3"), - mc.add_msg('Stuff 4', uid=4, subject="test4")]) + mc.add_msg('Stuff 1', subject="test1"), + mc.add_msg('Stuff 2', subject="test2"), + mc.add_msg('Stuff 3', subject="test3"), + mc.add_msg('Stuff 4', subject="test4")]) return d def add_second(result): d = defer.gatherResults([ - mc.add_msg('Stuff 5', uid=5, subject="test5"), - mc.add_msg('Stuff 6', uid=6, subject="test6"), - mc.add_msg('Stuff 7', uid=7, subject="test7")]) + mc.add_msg('Stuff 5', subject="test5"), + mc.add_msg('Stuff 6', subject="test6"), + mc.add_msg('Stuff 7', subject="test7")]) return d def check_second(result): @@ -383,20 +171,20 @@ class MessageCollectionTestCase(IMAP4HelperMixin, unittest.TestCase): self.assertEqual(countrecent(), 0) - d = mc.add_msg('Stuff', uid=1, subject="test1") + d = mc.add_msg('Stuff', subject="test1") # For the semantics defined in the RFC, we auto-add the # recent flag by default. def add2(_): - return mc.add_msg('Stuff', subject="test2", uid=2, + return mc.add_msg('Stuff', subject="test2", flags=('\\Deleted',)) def add3(_): - return mc.add_msg('Stuff', subject="test3", uid=3, + return mc.add_msg('Stuff', subject="test3", flags=('\\Recent',)) def add4(_): - return mc.add_msg('Stuff', subject="test4", uid=4, + return mc.add_msg('Stuff', subject="test4", flags=('\\Deleted', '\\Recent')) d.addCallback(lambda r: eq(countrecent(), 1)) @@ -415,9 +203,9 @@ class MessageCollectionTestCase(IMAP4HelperMixin, unittest.TestCase): self.assertEqual(self.messages.count(), 0) def add_1(): - d1 = mc.add_msg('msg 1', uid=1, subject="test1") - d2 = mc.add_msg('msg 2', uid=2, subject="test2") - d3 = mc.add_msg('msg 3', uid=3, subject="test3") + d1 = mc.add_msg('msg 1', subject="test1") + d2 = mc.add_msg('msg 2', subject="test2") + d3 = mc.add_msg('msg 3', subject="test3") d = defer.gatherResults([d1, d2, d3]) return d @@ -1225,13 +1013,13 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def add_messages(): d1 = m.messages.add_msg( - 'test 1', uid=1, subject="Message 1", + 'test 1', subject="Message 1", flags=('\\Deleted', 'AnotherFlag')) d2 = m.messages.add_msg( - 'test 2', uid=2, subject="Message 2", + 'test 2', subject="Message 2", flags=('AnotherFlag',)) d3 = m.messages.add_msg( - 'test 3', uid=3, subject="Message 3", + 'test 3', subject="Message 3", flags=('\\Deleted',)) d = defer.gatherResults([d1, d2, d3]) return d @@ -1273,13 +1061,13 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): def add_messages(): d1 = m.messages.add_msg( - 'test 1', uid=1, subject="Message 1", + 'test 1', subject="Message 1", flags=('\\Deleted', 'AnotherFlag')) d2 = m.messages.add_msg( - 'test 2', uid=2, subject="Message 2", + 'test 2', subject="Message 2", flags=('AnotherFlag',)) d3 = m.messages.add_msg( - 'test 3', uid=3, subject="Message 3", + 'test 3', subject="Message 3", flags=('\\Deleted',)) d = defer.gatherResults([d1, d2, d3]) return d @@ -1318,70 +1106,6 @@ class LeapIMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase): self.assertItemsEqual(self.results, [1, 3]) -class StoreAndFetchTestCase(unittest.TestCase, IMAP4HelperMixin): - """ - Several tests to check that the internal storage representation - is able to render the message structures as we expect them. - """ - - def setUp(self): - IMAP4HelperMixin.setUp(self) - self.received_messages = self.received_uid = None - self.result = None - - def addListener(self, x): - pass - - def removeListener(self, x): - pass - - def _addSignedMessage(self, _): - self.server.state = 'select' - infile = util.sibpath(__file__, 'rfc822.multi-signed.message') - raw = open(infile).read() - MBOX_NAME = "multipart/SIGNED" - - self.server.theAccount.addMailbox(MBOX_NAME) - mbox = self.server.theAccount.getMailbox(MBOX_NAME) - self.server.mbox = mbox - # return a deferred that will fire with UID - return self.server.mbox.messages.add_msg(raw) - - def _fetchWork(self, uids): - - def result(R): - self.result = R - - self.connected.addCallback( - self._addSignedMessage).addCallback( - lambda uid: self.function( - uids, uid=uid) # do NOT use seq numbers! - ).addCallback(result).addCallback( - self._cbStopClient).addErrback(self._ebGeneral) - - d = loopback.loopbackTCP(self.server, self.client, noisy=False) - d.addCallback(lambda x: self.assertEqual(self.result, self.expected)) - return d - - def testMultiBody(self): - """ - Test that a multipart signed message is retrieved the same - as we stored it. - """ - self.function = self.client.fetchBody - messages = '1' - - # XXX review. This probably should give everything? - - self.expected = {1: { - 'RFC822.TEXT': 'This is an example of a signed message,\n' - 'with attachments.\n\n\n--=20\n' - 'Nihil sine chao! =E2=88=B4\n', - 'UID': '1'}} - # print "test multi: fetch uid", messages - return self._fetchWork(messages) - - class IMAP4ServerSearchTestCase(IMAP4HelperMixin, unittest.TestCase): """ diff --git a/src/leap/mail/imap/tests/test_imap_store_fetch.py b/src/leap/mail/imap/tests/test_imap_store_fetch.py new file mode 100644 index 0000000..6da8581 --- /dev/null +++ b/src/leap/mail/imap/tests/test_imap_store_fetch.py @@ -0,0 +1,71 @@ +from twisted.protocols import loopback +from twisted.python import util + +from leap.mail.imap.tests.utils import IMAP4HelperMixin + +TEST_USER = "testuser@leap.se" +TEST_PASSWD = "1234" + + +class StoreAndFetchTestCase(IMAP4HelperMixin): + """ + Several tests to check that the internal storage representation + is able to render the message structures as we expect them. + """ + + def setUp(self): + IMAP4HelperMixin.setUp(self) + self.received_messages = self.received_uid = None + self.result = None + + def addListener(self, x): + pass + + def removeListener(self, x): + pass + + def _addSignedMessage(self, _): + self.server.state = 'select' + infile = util.sibpath(__file__, 'rfc822.multi-signed.message') + raw = open(infile).read() + MBOX_NAME = "multipart/SIGNED" + + self.server.theAccount.addMailbox(MBOX_NAME) + mbox = self.server.theAccount.getMailbox(MBOX_NAME) + self.server.mbox = mbox + # return a deferred that will fire with UID + return self.server.mbox.messages.add_msg(raw) + + def _fetchWork(self, uids): + + def result(R): + self.result = R + + self.connected.addCallback( + self._addSignedMessage).addCallback( + lambda uid: self.function( + uids, uid=uid) # do NOT use seq numbers! + ).addCallback(result).addCallback( + self._cbStopClient).addErrback(self._ebGeneral) + + d = loopback.loopbackTCP(self.server, self.client, noisy=False) + d.addCallback(lambda x: self.assertEqual(self.result, self.expected)) + return d + + def testMultiBody(self): + """ + Test that a multipart signed message is retrieved the same + as we stored it. + """ + self.function = self.client.fetchBody + messages = '1' + + # XXX review. This probably should give everything? + + self.expected = {1: { + 'RFC822.TEXT': 'This is an example of a signed message,\n' + 'with attachments.\n\n\n--=20\n' + 'Nihil sine chao! =E2=88=B4\n', + 'UID': '1'}} + # print "test multi: fetch uid", messages + return self._fetchWork(messages) diff --git a/src/leap/mail/imap/tests/utils.py b/src/leap/mail/imap/tests/utils.py new file mode 100644 index 0000000..0932bd4 --- /dev/null +++ b/src/leap/mail/imap/tests/utils.py @@ -0,0 +1,225 @@ +import os +import tempfile +import shutil + +from email import parser + +from mock import Mock +from twisted.mail import imap4 +from twisted.internet import defer +from twisted.protocols import loopback + +from leap.common.testing.basetest import BaseLeapTest +from leap.mail.imap.account import SoledadBackedAccount +from leap.mail.imap.memorystore import MemoryStore +from leap.mail.imap.server import LeapIMAPServer +from leap.soledad.client import Soledad + +TEST_USER = "testuser@leap.se" +TEST_PASSWD = "1234" + +# +# 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) + self.deferred = deferred + self.events = [] + + def serverGreeting(self, caps): + self.deferred.callback(None) + + def modeChanged(self, writeable): + self.events.append(['modeChanged', writeable]) + self.transport.loseConnection() + + def flagsChanged(self, newFlags): + self.events.append(['flagsChanged', newFlags]) + self.transport.loseConnection() + + def newMessages(self, exists, recent): + self.events.append(['newMessages', exists, recent]) + self.transport.loseConnection() + + +def initialize_soledad(email, gnupg_home, tempdir): + """ + 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 + """ + + uuid = "foobar-uuid" + passphrase = u"verysecretpassphrase" + secret_path = os.path.join(tempdir, "secret.gpg") + local_db_path = os.path.join(tempdir, "soledad.u1db") + server_url = "http://provider" + cert_file = "" + + class MockSharedDB(object): + + get_doc = Mock(return_value=None) + put_doc = Mock() + lock = Mock(return_value=('atoken', 300)) + unlock = Mock(return_value=True) + + def __call__(self): + return self + + Soledad._shared_db = MockSharedDB() + + _soledad = Soledad( + uuid, + passphrase, + secret_path, + local_db_path, + server_url, + cert_file) + + return _soledad + + +# XXX this is not properly a mixin, since helper already inherits from +# uniittest.Testcase +class IMAP4HelperMixin(BaseLeapTest): + """ + MixIn containing several utilities to be shared across + different TestCases + """ + + serverCTX = None + clientCTX = None + + # setUpClass cannot be a classmethod in trial, see: + # https://twistedmatrix.com/trac/ticket/1870 + + 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. + """ + self.old_path = os.environ['PATH'] + self.old_home = os.environ['HOME'] + self.tempdir = tempfile.mkdtemp(prefix="leap_tests-") + self.home = self.tempdir + bin_tdir = os.path.join( + self.tempdir, + 'bin') + os.environ["PATH"] = bin_tdir + os.environ["HOME"] = self.tempdir + + # Soledad: config info + self.gnupg_home = "%s/gnupg" % self.tempdir + self.email = 'leap@leap.se' + + # initialize soledad by hand so we can control keys + self._soledad = initialize_soledad( + self.email, + self.gnupg_home, + self.tempdir) + UUID = 'deadbeef', + USERID = TEST_USER + memstore = MemoryStore() + + ########### + + d = defer.Deferred() + self.server = LeapIMAPServer( + uuid=UUID, userid=USERID, + contextFactory=self.serverCTX, + # XXX do we really need this?? + soledad=self._soledad) + + self.client = SimpleClient(d, contextFactory=self.clientCTX) + self.connected = d + + # 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. + + theAccount = SoledadBackedAccount( + USERID, + soledad=self._soledad, + memstore=memstore) + LeapIMAPServer.theAccount = theAccount + + # in case we get something from previous tests... + for mb in self.server.theAccount.mailboxes: + self.server.theAccount.delete(mb) + + # email parser + self.parser = parser.Parser() + + def tearDown(self): + """ + tearDown method called after each test. + + Deletes all documents in the Index, and deletes + instances of server and client. + """ + try: + self._soledad.close() + os.environ["PATH"] = self.old_path + os.environ["HOME"] = self.old_home + # safety check + assert 'leap_tests-' in self.tempdir + shutil.rmtree(self.tempdir) + except Exception: + print "ERROR WHILE CLOSING SOLEDAD" + + def populateMessages(self): + """ + 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") + + def delete_all_docs(self): + """ + 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() + + def _ebGeneral(self, failure): + self.client.transport.loseConnection() + self.server.transport.loseConnection() + # can we do something similar? + # I guess this was ok with trial, but not in noseland... + # log.err(failure, "Problem with %r" % (self.function,)) + raise failure.value + # failure.trap(Exception) + + def loopback(self): + return loopback.loopbackAsync(self.server, self.client) + + |