summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--mail/src/leap/mail/imap/tests/test_imap.py318
-rw-r--r--mail/src/leap/mail/imap/tests/test_imap_store_fetch.py71
-rw-r--r--mail/src/leap/mail/imap/tests/utils.py225
3 files changed, 317 insertions, 297 deletions
diff --git a/mail/src/leap/mail/imap/tests/test_imap.py b/mail/src/leap/mail/imap/tests/test_imap.py
index bad8a5b5..980e46e7 100644
--- a/mail/src/leap/mail/imap/tests/test_imap.py
+++ b/mail/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/mail/src/leap/mail/imap/tests/test_imap_store_fetch.py b/mail/src/leap/mail/imap/tests/test_imap_store_fetch.py
new file mode 100644
index 00000000..6da8581a
--- /dev/null
+++ b/mail/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/mail/src/leap/mail/imap/tests/utils.py b/mail/src/leap/mail/imap/tests/utils.py
new file mode 100644
index 00000000..0932bd4e
--- /dev/null
+++ b/mail/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)
+
+