diff options
Diffstat (limited to 'src/leap/mail/imap/tests/utils.py')
| -rw-r--r-- | src/leap/mail/imap/tests/utils.py | 225 |
1 files changed, 225 insertions, 0 deletions
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) + + |