diff options
Diffstat (limited to 'src/leap/mail/tests')
| -rw-r--r-- | src/leap/mail/tests/common.py | 106 | ||||
| -rw-r--r-- | src/leap/mail/tests/rfc822.message | 86 | ||||
| -rw-r--r-- | src/leap/mail/tests/test_mail.py | 95 | ||||
| -rw-r--r-- | src/leap/mail/tests/test_mailbox_indexer.py | 241 |
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 |