summaryrefslogtreecommitdiff
path: root/src/leap/mail/tests
diff options
context:
space:
mode:
Diffstat (limited to 'src/leap/mail/tests')
-rw-r--r--src/leap/mail/tests/common.py106
-rw-r--r--src/leap/mail/tests/rfc822.message86
-rw-r--r--src/leap/mail/tests/test_mail.py95
-rw-r--r--src/leap/mail/tests/test_mailbox_indexer.py241
4 files changed, 528 insertions, 0 deletions
diff --git a/src/leap/mail/tests/common.py b/src/leap/mail/tests/common.py
new file mode 100644
index 0000000..fefa7ee
--- /dev/null
+++ b/src/leap/mail/tests/common.py
@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+# common.py
+# Copyright (C) 2014 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/>.
+"""
+Common utilities for testing Soledad.
+"""
+import os
+import shutil
+import tempfile
+
+from leap.common.testing.basetest import BaseLeapTest
+from leap.soledad.client import Soledad
+
+# TODO move to common module, or Soledad itself
+# XXX remove duplication
+
+TEST_USER = "testuser@leap.se"
+TEST_PASSWD = "1234"
+
+
+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 = "https://provider"
+ cert_file = ""
+
+ soledad = Soledad(
+ uuid,
+ passphrase,
+ secret_path,
+ local_db_path,
+ server_url,
+ cert_file,
+ syncable=False)
+
+ return soledad
+
+
+class SoledadTestMixin(BaseLeapTest):
+ """
+ It is **VERY** important that this base is added *AFTER* unittest.TestCase
+ """
+
+ def setUp(self):
+ self.results = []
+
+ 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)
+
+ def tearDown(self):
+ """
+ tearDown method called after each test.
+ """
+ self.results = []
+ try:
+ self._soledad.close()
+ except Exception as exc:
+ print "ERROR WHILE CLOSING SOLEDAD"
+ # logging.exception(exc)
+ finally:
+ os.environ["PATH"] = self.old_path
+ os.environ["HOME"] = self.old_home
+ # safety check
+ assert 'leap_tests-' in self.tempdir
+ shutil.rmtree(self.tempdir)
diff --git a/src/leap/mail/tests/rfc822.message b/src/leap/mail/tests/rfc822.message
new file mode 100644
index 0000000..ee97ab9
--- /dev/null
+++ b/src/leap/mail/tests/rfc822.message
@@ -0,0 +1,86 @@
+Return-Path: <twisted-commits-admin@twistedmatrix.com>
+Delivered-To: exarkun@meson.dyndns.org
+Received: from localhost [127.0.0.1]
+ by localhost with POP3 (fetchmail-6.2.1)
+ for exarkun@localhost (single-drop); Thu, 20 Mar 2003 14:50:20 -0500 (EST)
+Received: from pyramid.twistedmatrix.com (adsl-64-123-27-105.dsl.austtx.swbell.net [64.123.27.105])
+ by intarweb.us (Postfix) with ESMTP id 4A4A513EA4
+ for <exarkun@meson.dyndns.org>; Thu, 20 Mar 2003 14:49:27 -0500 (EST)
+Received: from localhost ([127.0.0.1] helo=pyramid.twistedmatrix.com)
+ by pyramid.twistedmatrix.com with esmtp (Exim 3.35 #1 (Debian))
+ id 18w648-0007Vl-00; Thu, 20 Mar 2003 13:51:04 -0600
+Received: from acapnotic by pyramid.twistedmatrix.com with local (Exim 3.35 #1 (Debian))
+ id 18w63j-0007VK-00
+ for <twisted-commits@twistedmatrix.com>; Thu, 20 Mar 2003 13:50:39 -0600
+To: twisted-commits@twistedmatrix.com
+From: etrepum CVS <etrepum@twistedmatrix.com>
+Reply-To: twisted-python@twistedmatrix.com
+X-Mailer: CVSToys
+Message-Id: <E18w63j-0007VK-00@pyramid.twistedmatrix.com>
+Subject: [Twisted-commits] rebuild now works on python versions from 2.2.0 and up.
+Sender: twisted-commits-admin@twistedmatrix.com
+Errors-To: twisted-commits-admin@twistedmatrix.com
+X-BeenThere: twisted-commits@twistedmatrix.com
+X-Mailman-Version: 2.0.11
+Precedence: bulk
+List-Help: <mailto:twisted-commits-request@twistedmatrix.com?subject=help>
+List-Post: <mailto:twisted-commits@twistedmatrix.com>
+List-Subscribe: <http://twistedmatrix.com/cgi-bin/mailman/listinfo/twisted-commits>,
+ <mailto:twisted-commits-request@twistedmatrix.com?subject=subscribe>
+List-Id: <twisted-commits.twistedmatrix.com>
+List-Unsubscribe: <http://twistedmatrix.com/cgi-bin/mailman/listinfo/twisted-commits>,
+ <mailto:twisted-commits-request@twistedmatrix.com?subject=unsubscribe>
+List-Archive: <http://twistedmatrix.com/pipermail/twisted-commits/>
+Date: Thu, 20 Mar 2003 13:50:39 -0600
+
+Modified files:
+Twisted/twisted/python/rebuild.py 1.19 1.20
+
+Log message:
+rebuild now works on python versions from 2.2.0 and up.
+
+
+ViewCVS links:
+http://twistedmatrix.com/users/jh.twistd/viewcvs/cgi/viewcvs.cgi/twisted/python/rebuild.py.diff?r1=text&tr1=1.19&r2=text&tr2=1.20&cvsroot=Twisted
+
+Index: Twisted/twisted/python/rebuild.py
+diff -u Twisted/twisted/python/rebuild.py:1.19 Twisted/twisted/python/rebuild.py:1.20
+--- Twisted/twisted/python/rebuild.py:1.19 Fri Jan 17 13:50:49 2003
++++ Twisted/twisted/python/rebuild.py Thu Mar 20 11:50:08 2003
+@@ -206,15 +206,27 @@
+ clazz.__dict__.clear()
+ clazz.__getattr__ = __getattr__
+ clazz.__module__ = module.__name__
++ if newclasses:
++ import gc
++ if (2, 2, 0) <= sys.version_info[:3] < (2, 2, 2):
++ hasBrokenRebuild = 1
++ gc_objects = gc.get_objects()
++ else:
++ hasBrokenRebuild = 0
+ for nclass in newclasses:
+ ga = getattr(module, nclass.__name__)
+ if ga is nclass:
+ log.msg("WARNING: new-class %s not replaced by reload!" % reflect.qual(nclass))
+ else:
+- import gc
+- for r in gc.get_referrers(nclass):
+- if isinstance(r, nclass):
++ if hasBrokenRebuild:
++ for r in gc_objects:
++ if not getattr(r, '__class__', None) is nclass:
++ continue
+ r.__class__ = ga
++ else:
++ for r in gc.get_referrers(nclass):
++ if getattr(r, '__class__', None) is nclass:
++ r.__class__ = ga
+ if doLog:
+ log.msg('')
+ log.msg(' (fixing %s): ' % str(module.__name__))
+
+
+_______________________________________________
+Twisted-commits mailing list
+Twisted-commits@twistedmatrix.com
+http://twistedmatrix.com/cgi-bin/mailman/listinfo/twisted-commits
diff --git a/src/leap/mail/tests/test_mail.py b/src/leap/mail/tests/test_mail.py
new file mode 100644
index 0000000..ce2366c
--- /dev/null
+++ b/src/leap/mail/tests/test_mail.py
@@ -0,0 +1,95 @@
+# -*- coding: utf-8 -*-
+# test_mail.py
+# Copyright (C) 2014 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/>.
+"""
+Tests for the mail module.
+"""
+import os
+from functools import partial
+
+from leap.mail.adaptors.soledad import SoledadMailAdaptor
+from leap.mail.mail import MessageCollection
+from leap.mail.mailbox_indexer import MailboxIndexer
+from leap.mail.tests.common import SoledadTestMixin
+
+from twisted.internet import defer
+from twisted.trial import unittest
+
+HERE = os.path.split(os.path.abspath(__file__))[0]
+
+
+class MessageCollectionTestCase(unittest.TestCase, SoledadTestMixin):
+ """
+ Tests for the SoledadDocumentWrapper.
+ """
+
+ def get_collection(self, mbox_collection=True):
+ """
+ Get a collection for tests.
+ """
+ adaptor = SoledadMailAdaptor()
+ store = self._soledad
+ adaptor.store = store
+ if mbox_collection:
+ mbox_indexer = MailboxIndexer(store)
+ mbox_name = "TestMbox"
+ else:
+ mbox_indexer = mbox_name = None
+
+ def get_collection_from_mbox_wrapper(wrapper):
+ return MessageCollection(
+ adaptor, store,
+ mbox_indexer=mbox_indexer, mbox_wrapper=wrapper)
+
+ d = adaptor.initialize_store(store)
+ if mbox_collection:
+ d.addCallback(lambda _: mbox_indexer.create_table(mbox_name))
+ d.addCallback(lambda _: adaptor.get_or_create_mbox(store, mbox_name))
+ d.addCallback(get_collection_from_mbox_wrapper)
+ return d
+
+ def test_is_mailbox_collection(self):
+
+ def assert_is_mbox_collection(collection):
+ self.assertTrue(collection.is_mailbox_collection())
+
+ d = self.get_collection()
+ d.addCallback(assert_is_mbox_collection)
+ return d
+
+ def assert_collection_count(self, _, expected, collection):
+
+ def _assert_count(count):
+ self.assertEqual(count, expected)
+ d = collection.count()
+ d.addCallback(_assert_count)
+ return d
+
+ def test_add_msg(self):
+
+ with open(os.path.join(HERE, "rfc822.message")) as f:
+ raw = f.read()
+
+ def add_msg_to_collection_and_assert_count(collection):
+ d = collection.add_msg(raw)
+ d.addCallback(partial(
+ self.assert_collection_count,
+ expected=1, collection=collection))
+ return d
+
+ d = self.get_collection()
+ d.addCallback(add_msg_to_collection_and_assert_count)
+ return d
diff --git a/src/leap/mail/tests/test_mailbox_indexer.py b/src/leap/mail/tests/test_mailbox_indexer.py
new file mode 100644
index 0000000..47a3bdc
--- /dev/null
+++ b/src/leap/mail/tests/test_mailbox_indexer.py
@@ -0,0 +1,241 @@
+# -*- coding: utf-8 -*-
+# test_mailbox_indexer.py
+# Copyright (C) 2014 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/>.
+"""
+Tests for the mailbox_indexer module.
+"""
+from functools import partial
+
+from twisted.trial import unittest
+
+from leap.mail import mailbox_indexer as mi
+from leap.mail.tests.common import SoledadTestMixin
+
+hash_test0 = '590c9f8430c7435807df8ba9a476e3f1295d46ef210f6efae2043a4c085a569e'
+hash_test1 = '1b4f0e9851971998e732078544c96b36c3d01cedf7caa332359d6f1d83567014'
+hash_test2 = '60303ae22b998861bce3b28f33eec1be758a213c86c93c076dbe9f558c11c752'
+hash_test3 = 'fd61a03af4f77d870fc21e05e7e80678095c92d808cfb3b5c279ee04c74aca13'
+hash_test4 = 'a4e624d686e03ed2767c0abd85c14426b0b1157d2ce81d27bb4fe4f6f01d688a'
+
+
+def fmt_hash(mailbox, hash):
+ return "M-" + mailbox + "-" + hash
+
+
+class MailboxIndexerTestCase(unittest.TestCase, SoledadTestMixin):
+ """
+ Tests for the MailboxUID class.
+ """
+ def get_mbox_uid(self):
+ m_uid = mi.MailboxIndexer(self._soledad)
+ return m_uid
+
+ def list_mail_tables_cb(self, ignored):
+ def filter_mailuid_tables(tables):
+ filtered = [
+ table[0] for table in tables if
+ table[0].startswith(mi.MailboxIndexer.table_preffix)]
+ return filtered
+
+ sql = "SELECT name FROM sqlite_master WHERE type='table';"
+ d = self._soledad.raw_sqlcipher_query(sql)
+ d.addCallback(filter_mailuid_tables)
+ return d
+
+ def select_uid_rows(self, mailbox):
+ sql = "SELECT * FROM %s%s;" % (
+ mi.MailboxIndexer.table_preffix, mailbox)
+ d = self._soledad.raw_sqlcipher_query(sql)
+ return d
+
+ def test_create_table(self):
+ def assert_table_created(tables):
+ self.assertEqual(
+ tables, ["leapmail_uid_inbox"])
+
+ m_uid = self.get_mbox_uid()
+ d = m_uid.create_table('inbox')
+ d.addCallback(self.list_mail_tables_cb)
+ d.addCallback(assert_table_created)
+ return d
+
+ def test_create_and_delete_table(self):
+ def assert_table_deleted(tables):
+ self.assertEqual(tables, [])
+
+ m_uid = self.get_mbox_uid()
+ d = m_uid.create_table('inbox')
+ d.addCallback(lambda _: m_uid.delete_table('inbox'))
+ d.addCallback(self.list_mail_tables_cb)
+ d.addCallback(assert_table_deleted)
+ return d
+
+ def test_rename_table(self):
+ def assert_table_renamed(tables):
+ self.assertEqual(
+ tables, ["leapmail_uid_foomailbox"])
+
+ m_uid = self.get_mbox_uid()
+ d = m_uid.create_table('inbox')
+ d.addCallback(lambda _: m_uid.rename_table('inbox', 'foomailbox'))
+ d.addCallback(self.list_mail_tables_cb)
+ d.addCallback(assert_table_renamed)
+ return d
+
+ def test_insert_doc(self):
+ m_uid = self.get_mbox_uid()
+ mbox = 'foomailbox'
+
+ h1 = fmt_hash(mbox, hash_test0)
+ h2 = fmt_hash(mbox, hash_test1)
+ h3 = fmt_hash(mbox, hash_test2)
+ h4 = fmt_hash(mbox, hash_test3)
+ h5 = fmt_hash(mbox, hash_test4)
+
+ def assert_uid_rows(rows):
+ expected = [(1, h1), (2, h2), (3, h3), (4, h4), (5, h5)]
+ self.assertEquals(rows, expected)
+
+ d = m_uid.create_table(mbox)
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h1))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h2))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h3))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h4))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h5))
+ d.addCallback(lambda _: self.select_uid_rows(mbox))
+ d.addCallback(assert_uid_rows)
+ return d
+
+ def test_insert_doc_return(self):
+ m_uid = self.get_mbox_uid()
+ mbox = 'foomailbox'
+
+ def assert_rowid(rowid, expected=None):
+ self.assertEqual(rowid, expected)
+
+ h1 = fmt_hash(mbox, hash_test0)
+ h2 = fmt_hash(mbox, hash_test1)
+ h3 = fmt_hash(mbox, hash_test2)
+
+ d = m_uid.create_table(mbox)
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h1))
+ d.addCallback(partial(assert_rowid, expected=1))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h2))
+ d.addCallback(partial(assert_rowid, expected=2))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h3))
+ d.addCallback(partial(assert_rowid, expected=3))
+ return d
+
+ def test_delete_doc(self):
+ m_uid = self.get_mbox_uid()
+ mbox = 'foomailbox'
+
+ h1 = fmt_hash(mbox, hash_test0)
+ h2 = fmt_hash(mbox, hash_test1)
+ h3 = fmt_hash(mbox, hash_test2)
+ h4 = fmt_hash(mbox, hash_test3)
+ h5 = fmt_hash(mbox, hash_test4)
+
+ def assert_uid_rows(rows):
+ expected = [(4, h4), (5, h5)]
+ self.assertEquals(rows, expected)
+
+ d = m_uid.create_table(mbox)
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h1))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h2))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h3))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h4))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h5))
+
+ d.addCallbacks(lambda _: m_uid.delete_doc_by_uid(mbox, 1))
+ d.addCallbacks(lambda _: m_uid.delete_doc_by_uid(mbox, 2))
+ d.addCallbacks(lambda _: m_uid.delete_doc_by_hash(mbox, h3))
+
+ d.addCallback(lambda _: self.select_uid_rows(mbox))
+ d.addCallback(assert_uid_rows)
+ return d
+
+ def test_get_doc_id_from_uid(self):
+ m_uid = self.get_mbox_uid()
+ mbox = 'foomailbox'
+
+ h1 = fmt_hash(mbox, hash_test0)
+
+ def assert_doc_hash(res):
+ self.assertEqual(res, h1)
+
+ d = m_uid.create_table(mbox)
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h1))
+ d.addCallback(lambda _: m_uid.get_doc_id_from_uid(mbox, 1))
+ d.addCallback(assert_doc_hash)
+ return d
+
+ def test_count(self):
+ m_uid = self.get_mbox_uid()
+ mbox = 'foomailbox'
+
+ h1 = fmt_hash(mbox, hash_test0)
+ h2 = fmt_hash(mbox, hash_test1)
+ h3 = fmt_hash(mbox, hash_test2)
+ h4 = fmt_hash(mbox, hash_test3)
+ h5 = fmt_hash(mbox, hash_test4)
+
+ d = m_uid.create_table(mbox)
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h1))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h2))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h3))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h4))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h5))
+
+ def assert_count_after_inserts(count):
+ self.assertEquals(count, 5)
+
+ d.addCallback(lambda _: m_uid.count(mbox))
+ d.addCallback(assert_count_after_inserts)
+
+ d.addCallbacks(lambda _: m_uid.delete_doc_by_uid(mbox, 1))
+ d.addCallbacks(lambda _: m_uid.delete_doc_by_uid(mbox, 2))
+
+ def assert_count_after_deletions(count):
+ self.assertEquals(count, 3)
+
+ d.addCallback(lambda _: m_uid.count(mbox))
+ d.addCallback(assert_count_after_deletions)
+ return d
+
+ def test_get_next_uid(self):
+ m_uid = self.get_mbox_uid()
+ mbox = 'foomailbox'
+
+ h1 = fmt_hash(mbox, hash_test0)
+ h2 = fmt_hash(mbox, hash_test1)
+ h3 = fmt_hash(mbox, hash_test2)
+ h4 = fmt_hash(mbox, hash_test3)
+ h5 = fmt_hash(mbox, hash_test4)
+
+ d = m_uid.create_table(mbox)
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h1))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h2))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h3))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h4))
+ d.addCallback(lambda _: m_uid.insert_doc(mbox, h5))
+
+ def assert_next_uid(result, expected=1):
+ self.assertEquals(result, expected)
+
+ d.addCallback(lambda _: m_uid.get_next_uid(mbox))
+ d.addCallback(partial(assert_next_uid, expected=6))
+ return d