From b915e3d5bd1e37c732b44559af5587f6c6a90fc3 Mon Sep 17 00:00:00 2001
From: drebs <drebs@leap.se>
Date: Tue, 25 Nov 2014 14:58:31 -0200
Subject: Adapt tests for new api.
---
common/src/leap/soledad/common/tests/__init__.py | 284 +-------
common/src/leap/soledad/common/tests/test_couch.py | 164 +----
.../tests/test_couch_operations_atomicity.py | 206 +++---
.../src/leap/soledad/common/tests/test_crypto.py | 17 +-
common/src/leap/soledad/common/tests/test_http.py | 5 -
.../leap/soledad/common/tests/test_http_client.py | 11 +-
common/src/leap/soledad/common/tests/test_https.py | 42 +-
.../src/leap/soledad/common/tests/test_server.py | 352 ++++-----
.../src/leap/soledad/common/tests/test_soledad.py | 166 ++---
.../leap/soledad/common/tests/test_soledad_app.py | 59 ++
.../leap/soledad/common/tests/test_soledad_doc.py | 24 +-
.../leap/soledad/common/tests/test_sqlcipher.py | 62 +-
.../soledad/common/tests/test_sqlcipher_sync.py | 49 +-
common/src/leap/soledad/common/tests/test_sync.py | 153 ++--
.../soledad/common/tests/test_sync_deferred.py | 150 ++--
.../leap/soledad/common/tests/test_sync_target.py | 285 ++++----
.../src/leap/soledad/common/tests/test_target.py | 797 ---------------------
.../soledad/common/tests/test_target_soledad.py | 102 ---
.../soledad/common/tests/u1db_tests/__init__.py | 4 +-
.../soledad/common/tests/u1db_tests/test_https.py | 2 +-
common/src/leap/soledad/common/tests/util.py | 352 ++++++++-
21 files changed, 1116 insertions(+), 2170 deletions(-)
create mode 100644 common/src/leap/soledad/common/tests/test_soledad_app.py
delete mode 100644 common/src/leap/soledad/common/tests/test_target.py
delete mode 100644 common/src/leap/soledad/common/tests/test_target_soledad.py
(limited to 'common/src')
diff --git a/common/src/leap/soledad/common/tests/__init__.py b/common/src/leap/soledad/common/tests/__init__.py
index f8253409..acebb77b 100644
--- a/common/src/leap/soledad/common/tests/__init__.py
+++ b/common/src/leap/soledad/common/tests/__init__.py
@@ -19,291 +19,9 @@
"""
Tests to make sure Soledad provides U1DB functionality and more.
"""
-import os
-import random
-import string
-import u1db
-from mock import Mock
-
-
-from leap.soledad.common.document import SoledadDocument
-from leap.soledad.common.crypto import ENC_SCHEME_KEY
-from leap.soledad.client import Soledad
-from leap.soledad.client.crypto import decrypt_doc_dict
-from leap.common.testing.basetest import BaseLeapTest
-
-
-#-----------------------------------------------------------------------------
-# Some tests inherit from BaseSoledadTest in order to have a working Soledad
-# instance in each test.
-#-----------------------------------------------------------------------------
-
-ADDRESS = 'leap@leap.se'
-
-
-class BaseSoledadTest(BaseLeapTest):
- """
- Instantiates Soledad for usage in tests.
- """
- defer_sync_encryption = False
-
- def setUp(self):
- # config info
- self.db1_file = os.path.join(self.tempdir, "db1.u1db")
- self.db2_file = os.path.join(self.tempdir, "db2.u1db")
- self.email = ADDRESS
- # open test dbs
- self._db1 = u1db.open(self.db1_file, create=True,
- document_factory=SoledadDocument)
- self._db2 = u1db.open(self.db2_file, create=True,
- document_factory=SoledadDocument)
- # get a random prefix for each test, so we do not mess with
- # concurrency during initialization and shutting down of
- # each local db.
- self.rand_prefix = ''.join(
- map(lambda x: random.choice(string.ascii_letters), range(6)))
- # initialize soledad by hand so we can control keys
- self._soledad = self._soledad_instance(
- prefix=self.rand_prefix, user=self.email)
-
- def tearDown(self):
- self._db1.close()
- self._db2.close()
- self._soledad.close()
-
- # XXX should not access "private" attrs
- for f in [self._soledad._local_db_path, self._soledad._secrets_path]:
- if os.path.isfile(f):
- os.unlink(f)
- def get_default_shared_mock(self, put_doc_side_effect):
- """
- Get a default class for mocking the shared DB
- """
- class defaultMockSharedDB(object):
- get_doc = Mock(return_value=None)
- put_doc = Mock(side_effect=put_doc_side_effect)
- lock = Mock(return_value=('atoken', 300))
- unlock = Mock(return_value=True)
- def __call__(self):
- return self
- return defaultMockSharedDB
-
- def _soledad_instance(self, user=ADDRESS, passphrase=u'123',
- prefix='',
- secrets_path='secrets.json',
- local_db_path='soledad.u1db', server_url='',
- cert_file=None, secret_id=None,
- shared_db_class=None):
-
- def _put_doc_side_effect(doc):
- self._doc_put = doc
-
- if shared_db_class is not None:
- MockSharedDB = shared_db_class
- else:
- MockSharedDB = self.get_default_shared_mock(
- _put_doc_side_effect)
-
- Soledad._shared_db = MockSharedDB()
- return Soledad(
- user,
- passphrase,
- secrets_path=os.path.join(
- self.tempdir, prefix, secrets_path),
- local_db_path=os.path.join(
- self.tempdir, prefix, local_db_path),
- server_url=server_url, # Soledad will fail if not given an url.
- cert_file=cert_file,
- secret_id=secret_id,
- defer_encryption=self.defer_sync_encryption)
-
- def assertGetEncryptedDoc(
- self, db, doc_id, doc_rev, content, has_conflicts):
- """
- Assert that the document in the database looks correct.
- """
- exp_doc = self.make_document(doc_id, doc_rev, content,
- has_conflicts=has_conflicts)
- doc = db.get_doc(doc_id)
-
- if ENC_SCHEME_KEY in doc.content:
- # XXX check for SYM_KEY too
- key = self._soledad._crypto.doc_passphrase(doc.doc_id)
- secret = self._soledad._crypto.secret
- decrypted = decrypt_doc_dict(
- doc.content, doc.doc_id, doc.rev,
- key, secret)
- doc.set_json(decrypted)
- self.assertEqual(exp_doc.doc_id, doc.doc_id)
- self.assertEqual(exp_doc.rev, doc.rev)
- self.assertEqual(exp_doc.has_conflicts, doc.has_conflicts)
- self.assertEqual(exp_doc.content, doc.content)
-
-
-# Key material for testing
-KEY_FINGERPRINT = "E36E738D69173C13D709E44F2F455E2824D18DDF"
-PUBLIC_KEY = """
------BEGIN PGP PUBLIC KEY BLOCK-----
-Version: GnuPG v1.4.10 (GNU/Linux)
-
-mQINBFC9+dkBEADNRfwV23TWEoGc/x0wWH1P7PlXt8MnC2Z1kKaKKmfnglVrpOiz
-iLWoiU58sfZ0L5vHkzXHXCBf6Eiy/EtUIvdiWAn+yASJ1mk5jZTBKO/WMAHD8wTO
-zpMsFmWyg3xc4DkmFa9KQ5EVU0o/nqPeyQxNMQN7px5pPwrJtJFmPxnxm+aDkPYx
-irDmz/4DeDNqXliazGJKw7efqBdlwTHkl9Akw2gwy178pmsKwHHEMOBOFFvX61AT
-huKqHYmlCGSliwbrJppTG7jc1/ls3itrK+CWTg4txREkSpEVmfcASvw/ZqLbjgfs
-d/INMwXnR9U81O8+7LT6yw/ca4ppcFoJD7/XJbkRiML6+bJ4Dakiy6i727BzV17g
-wI1zqNvm5rAhtALKfACha6YO43aJzairO4II1wxVHvRDHZn2IuKDDephQ3Ii7/vb
-hUOf6XCSmchkAcpKXUOvbxm1yfB1LRa64mMc2RcZxf4mW7KQkulBsdV5QG2276lv
-U2UUy2IutXcGP5nXC+f6sJJGJeEToKJ57yiO/VWJFjKN8SvP+7AYsQSqINUuEf6H
-T5gCPCraGMkTUTPXrREvu7NOohU78q6zZNaL3GW8ai7eSeANSuQ8Vzffx7Wd8Y7i
-Pw9sYj0SMFs1UgjbuL6pO5ueHh+qyumbtAq2K0Bci0kqOcU4E9fNtdiovQARAQAB
-tBxMZWFwIFRlc3QgS2V5IDxsZWFwQGxlYXAuc2U+iQI3BBMBCAAhBQJQvfnZAhsD
-BQsJCAcDBRUKCQgLBRYCAwEAAh4BAheAAAoJEC9FXigk0Y3fT7EQAKH3IuRniOpb
-T/DDIgwwjz3oxB/W0DDMyPXowlhSOuM0rgGfntBpBb3boezEXwL86NPQxNGGruF5
-hkmecSiuPSvOmQlqlS95NGQp6hNG0YaKColh+Q5NTspFXCAkFch9oqUje0LdxfSP
-QfV9UpeEvGyPmk1I9EJV/YDmZ4+Djge1d7qhVZInz4Rx1NrSyF/Tc2EC0VpjQFsU
-Y9Kb2YBBR7ivG6DBc8ty0jJXi7B4WjkFcUEJviQpMF2dCLdonCehYs1PqsN1N7j+
-eFjQd+hqVMJgYuSGKjvuAEfClM6MQw7+FmFwMyLgK/Ew/DttHEDCri77SPSkOGSI
-txCzhTg6798f6mJr7WcXmHX1w1Vcib5FfZ8vTDFVhz/XgAgArdhPo9V6/1dgSSiB
-KPQ/spsco6u5imdOhckERE0lnAYvVT6KE81TKuhF/b23u7x+Wdew6kK0EQhYA7wy
-7LmlaNXc7rMBQJ9Z60CJ4JDtatBWZ0kNrt2VfdDHVdqBTOpl0CraNUjWE5YMDasr
-K2dF5IX8D3uuYtpZnxqg0KzyLg0tzL0tvOL1C2iudgZUISZNPKbS0z0v+afuAAnx
-2pTC3uezbh2Jt8SWTLhll4i0P4Ps5kZ6HQUO56O+/Z1cWovX+mQekYFmERySDR9n
-3k1uAwLilJmRmepGmvYbB8HloV8HqwgguQINBFC9+dkBEAC0I/xn1uborMgDvBtf
-H0sEhwnXBC849/32zic6udB6/3Efk9nzbSpL3FSOuXITZsZgCHPkKarnoQ2ztMcS
-sh1ke1C5gQGms75UVmM/nS+2YI4vY8OX/GC/on2vUyncqdH+bR6xH5hx4NbWpfTs
-iQHmz5C6zzS/kuabGdZyKRaZHt23WQ7JX/4zpjqbC99DjHcP9BSk7tJ8wI4bkMYD
-uFVQdT9O6HwyKGYwUU4sAQRAj7XCTGvVbT0dpgJwH4RmrEtJoHAx4Whg8mJ710E0
-GCmzf2jqkNuOw76ivgk27Kge+Hw00jmJjQhHY0yVbiaoJwcRrPKzaSjEVNgrpgP3
-lXPRGQArgESsIOTeVVHQ8fhK2YtTeCY9rIiO+L0OX2xo9HK7hfHZZWL6rqymXdyS
-fhzh/f6IPyHFWnvj7Brl7DR8heMikygcJqv+ed2yx7iLyCUJ10g12I48+aEj1aLe
-dP7lna32iY8/Z0SHQLNH6PXO9SlPcq2aFUgKqE75A/0FMk7CunzU1OWr2ZtTLNO1
-WT/13LfOhhuEq9jTyTosn0WxBjJKq18lnhzCXlaw6EAtbA7CUwsD3CTPR56aAXFK
-3I7KXOVAqggrvMe5Tpdg5drfYpI8hZovL5aAgb+7Y5ta10TcJdUhS5K3kFAWe/td
-U0cmWUMDP1UMSQ5Jg6JIQVWhSwARAQABiQIfBBgBCAAJBQJQvfnZAhsMAAoJEC9F
-Xigk0Y3fRwsP/i0ElYCyxeLpWJTwo1iCLkMKz2yX1lFVa9nT1BVTPOQwr/IAc5OX
-NdtbJ14fUsKL5pWgW8OmrXtwZm1y4euI1RPWWubG01ouzwnGzv26UcuHeqC5orZj
-cOnKtL40y8VGMm8LoicVkRJH8blPORCnaLjdOtmA3rx/v2EXrJpSa3AhOy0ZSRXk
-ZSrK68AVNwamHRoBSYyo0AtaXnkPX4+tmO8X8BPfj125IljubvwZPIW9VWR9UqCE
-VPfDR1XKegVb6VStIywF7kmrknM1C5qUY28rdZYWgKorw01hBGV4jTW0cqde3N51
-XT1jnIAa+NoXUM9uQoGYMiwrL7vNsLlyyiW5ayDyV92H/rIuiqhFgbJsHTlsm7I8
-oGheR784BagAA1NIKD1qEO9T6Kz9lzlDaeWS5AUKeXrb7ZJLI1TTCIZx5/DxjLqM
-Tt/RFBpVo9geZQrvLUqLAMwdaUvDXC2c6DaCPXTh65oCZj/hqzlJHH+RoTWWzKI+
-BjXxgUWF9EmZUBrg68DSmI+9wuDFsjZ51BcqvJwxyfxtTaWhdoYqH/UQS+D1FP3/
-diZHHlzwVwPICzM9ooNTgbrcDzyxRkIVqsVwBq7EtzcvgYUyX53yG25Giy6YQaQ2
-ZtQ/VymwFL3XdUWV6B/hU4PVAFvO3qlOtdJ6TpE+nEWgcWjCv5g7RjXX
-=MuOY
------END PGP PUBLIC KEY BLOCK-----
-"""
-PRIVATE_KEY = """
------BEGIN PGP PRIVATE KEY BLOCK-----
-Version: GnuPG v1.4.10 (GNU/Linux)
-
-lQcYBFC9+dkBEADNRfwV23TWEoGc/x0wWH1P7PlXt8MnC2Z1kKaKKmfnglVrpOiz
-iLWoiU58sfZ0L5vHkzXHXCBf6Eiy/EtUIvdiWAn+yASJ1mk5jZTBKO/WMAHD8wTO
-zpMsFmWyg3xc4DkmFa9KQ5EVU0o/nqPeyQxNMQN7px5pPwrJtJFmPxnxm+aDkPYx
-irDmz/4DeDNqXliazGJKw7efqBdlwTHkl9Akw2gwy178pmsKwHHEMOBOFFvX61AT
-huKqHYmlCGSliwbrJppTG7jc1/ls3itrK+CWTg4txREkSpEVmfcASvw/ZqLbjgfs
-d/INMwXnR9U81O8+7LT6yw/ca4ppcFoJD7/XJbkRiML6+bJ4Dakiy6i727BzV17g
-wI1zqNvm5rAhtALKfACha6YO43aJzairO4II1wxVHvRDHZn2IuKDDephQ3Ii7/vb
-hUOf6XCSmchkAcpKXUOvbxm1yfB1LRa64mMc2RcZxf4mW7KQkulBsdV5QG2276lv
-U2UUy2IutXcGP5nXC+f6sJJGJeEToKJ57yiO/VWJFjKN8SvP+7AYsQSqINUuEf6H
-T5gCPCraGMkTUTPXrREvu7NOohU78q6zZNaL3GW8ai7eSeANSuQ8Vzffx7Wd8Y7i
-Pw9sYj0SMFs1UgjbuL6pO5ueHh+qyumbtAq2K0Bci0kqOcU4E9fNtdiovQARAQAB
-AA/+JHtlL39G1wsH9R6UEfUQJGXR9MiIiwZoKcnRB2o8+DS+OLjg0JOh8XehtuCs
-E/8oGQKtQqa5bEIstX7IZoYmYFiUQi9LOzIblmp2vxOm+HKkxa4JszWci2/ZmC3t
-KtaA4adl9XVnshoQ7pijuCMUKB3naBEOAxd8s9d/JeReGIYkJErdrnVfNk5N71Ds
-FmH5Ll3XtEDvgBUQP3nkA6QFjpsaB94FHjL3gDwum/cxzj6pCglcvHOzEhfY0Ddb
-J967FozQTaf2JW3O+w3LOqtcKWpq87B7+O61tVidQPSSuzPjCtFF0D2LC9R/Hpky
-KTMQ6CaKja4MPhjwywd4QPcHGYSqjMpflvJqi+kYIt8psUK/YswWjnr3r4fbuqVY
-VhtiHvnBHQjz135lUqWvEz4hM3Xpnxydx7aRlv5NlevK8+YIO5oFbWbGNTWsPZI5
-jpoFBpSsnR1Q5tnvtNHauvoWV+XN2qAOBTG+/nEbDYH6Ak3aaE9jrpTdYh0CotYF
-q7csANsDy3JvkAzeU6WnYpsHHaAjqOGyiZGsLej1UcXPFMosE/aUo4WQhiS8Zx2c
-zOVKOi/X5vQ2GdNT9Qolz8AriwzsvFR+bxPzyd8V6ALwDsoXvwEYinYBKK8j0OPv
-OOihSR6HVsuP9NUZNU9ewiGzte/+/r6pNXHvR7wTQ8EWLcEIAN6Zyrb0bHZTIlxt
-VWur/Ht2mIZrBaO50qmM5RD3T5oXzWXi/pjLrIpBMfeZR9DWfwQwjYzwqi7pxtYx
-nJvbMuY505rfnMoYxb4J+cpRXV8MS7Dr1vjjLVUC9KiwSbM3gg6emfd2yuA93ihv
-Pe3mffzLIiQa4mRE3wtGcioC43nWuV2K2e1KjxeFg07JhrezA/1Cak505ab/tmvP
-4YmjR5c44+yL/YcQ3HdFgs4mV+nVbptRXvRcPpolJsgxPccGNdvHhsoR4gwXMS3F
-RRPD2z6x8xeN73Q4KH3bm01swQdwFBZbWVfmUGLxvN7leCdfs9+iFJyqHiCIB6Iv
-mQfp8F0IAOwSo8JhWN+V1dwML4EkIrM8wUb4yecNLkyR6TpPH/qXx4PxVMC+vy6x
-sCtjeHIwKE+9vqnlhd5zOYh7qYXEJtYwdeDDmDbL8oks1LFfd+FyAuZXY33DLwn0
-cRYsr2OEZmaajqUB3NVmj3H4uJBN9+paFHyFSXrH68K1Fk2o3n+RSf2EiX+eICwI
-L6rqoF5sSVUghBWdNegV7qfy4anwTQwrIMGjgU5S6PKW0Dr/3iO5z3qQpGPAj5OW
-ATqPWkDICLbObPxD5cJlyyNE2wCA9VVc6/1d6w4EVwSq9h3/WTpATEreXXxTGptd
-LNiTA1nmakBYNO2Iyo3djhaqBdWjk+EIAKtVEnJH9FAVwWOvaj1RoZMA5DnDMo7e
-SnhrCXl8AL7Z1WInEaybasTJXn1uQ8xY52Ua4b8cbuEKRKzw/70NesFRoMLYoHTO
-dyeszvhoDHberpGRTciVmpMu7Hyi33rM31K9epA4ib6QbbCHnxkWOZB+Bhgj1hJ8
-xb4RBYWiWpAYcg0+DAC3w9gfxQhtUlZPIbmbrBmrVkO2GVGUj8kH6k4UV6kUHEGY
-HQWQR0HcbKcXW81ZXCCD0l7ROuEWQtTe5Jw7dJ4/QFuqZnPutXVRNOZqpl6eRShw
-7X2/a29VXBpmHA95a88rSQsL+qm7Fb3prqRmuMCtrUZgFz7HLSTuUMR867QcTGVh
-cCBUZXN0IEtleSA8bGVhcEBsZWFwLnNlPokCNwQTAQgAIQUCUL352QIbAwULCQgH
-AwUVCgkICwUWAgMBAAIeAQIXgAAKCRAvRV4oJNGN30+xEACh9yLkZ4jqW0/wwyIM
-MI896MQf1tAwzMj16MJYUjrjNK4Bn57QaQW926HsxF8C/OjT0MTRhq7heYZJnnEo
-rj0rzpkJapUveTRkKeoTRtGGigqJYfkOTU7KRVwgJBXIfaKlI3tC3cX0j0H1fVKX
-hLxsj5pNSPRCVf2A5mePg44HtXe6oVWSJ8+EcdTa0shf03NhAtFaY0BbFGPSm9mA
-QUe4rxugwXPLctIyV4uweFo5BXFBCb4kKTBdnQi3aJwnoWLNT6rDdTe4/nhY0Hfo
-alTCYGLkhio77gBHwpTOjEMO/hZhcDMi4CvxMPw7bRxAwq4u+0j0pDhkiLcQs4U4
-Ou/fH+pia+1nF5h19cNVXIm+RX2fL0wxVYc/14AIAK3YT6PVev9XYEkogSj0P7Kb
-HKOruYpnToXJBERNJZwGL1U+ihPNUyroRf29t7u8flnXsOpCtBEIWAO8Muy5pWjV
-3O6zAUCfWetAieCQ7WrQVmdJDa7dlX3Qx1XagUzqZdAq2jVI1hOWDA2rKytnReSF
-/A97rmLaWZ8aoNCs8i4NLcy9Lbzi9QtornYGVCEmTTym0tM9L/mn7gAJ8dqUwt7n
-s24dibfElky4ZZeItD+D7OZGeh0FDuejvv2dXFqL1/pkHpGBZhEckg0fZ95NbgMC
-4pSZkZnqRpr2GwfB5aFfB6sIIJ0HGARQvfnZARAAtCP8Z9bm6KzIA7wbXx9LBIcJ
-1wQvOPf99s4nOrnQev9xH5PZ820qS9xUjrlyE2bGYAhz5Cmq56ENs7THErIdZHtQ
-uYEBprO+VFZjP50vtmCOL2PDl/xgv6J9r1Mp3KnR/m0esR+YceDW1qX07IkB5s+Q
-us80v5LmmxnWcikWmR7dt1kOyV/+M6Y6mwvfQ4x3D/QUpO7SfMCOG5DGA7hVUHU/
-Tuh8MihmMFFOLAEEQI+1wkxr1W09HaYCcB+EZqxLSaBwMeFoYPJie9dBNBgps39o
-6pDbjsO+or4JNuyoHvh8NNI5iY0IR2NMlW4mqCcHEazys2koxFTYK6YD95Vz0RkA
-K4BErCDk3lVR0PH4StmLU3gmPayIjvi9Dl9saPRyu4Xx2WVi+q6spl3ckn4c4f3+
-iD8hxVp74+wa5ew0fIXjIpMoHCar/nndsse4i8glCddINdiOPPmhI9Wi3nT+5Z2t
-9omPP2dEh0CzR+j1zvUpT3KtmhVICqhO+QP9BTJOwrp81NTlq9mbUyzTtVk/9dy3
-zoYbhKvY08k6LJ9FsQYySqtfJZ4cwl5WsOhALWwOwlMLA9wkz0eemgFxStyOylzl
-QKoIK7zHuU6XYOXa32KSPIWaLy+WgIG/u2ObWtdE3CXVIUuSt5BQFnv7XVNHJllD
-Az9VDEkOSYOiSEFVoUsAEQEAAQAP/1AagnZQZyzHDEgw4QELAspYHCWLXE5aZInX
-wTUJhK31IgIXNn9bJ0hFiSpQR2xeMs9oYtRuPOu0P8oOFMn4/z374fkjZy8QVY3e
-PlL+3EUeqYtkMwlGNmVw5a/NbNuNfm5Darb7pEfbYd1gPcni4MAYw7R2SG/57GbC
-9gucvspHIfOSfBNLBthDzmK8xEKe1yD2eimfc2T7IRYb6hmkYfeds5GsqvGI6mwI
-85h4uUHWRc5JOlhVM6yX8hSWx0L60Z3DZLChmc8maWnFXd7C8eQ6P1azJJbW71Ih
-7CoK0XW4LE82vlQurSRFgTwfl7wFYszW2bOzCuhHDDtYnwH86Nsu0DC78ZVRnvxn
-E8Ke/AJgrdhIOo4UAyR+aZD2+2mKd7/waOUTUrUtTzc7i8N3YXGi/EIaNReBXaq+
-ZNOp24BlFzRp+FCF/pptDW9HjPdiV09x0DgICmeZS4Gq/4vFFIahWctg52NGebT0
-Idxngjj+xDtLaZlLQoOz0n5ByjO/Wi0ANmMv1sMKCHhGvdaSws2/PbMR2r4caj8m
-KXpIgdinM/wUzHJ5pZyF2U/qejsRj8Kw8KH/tfX4JCLhiaP/mgeTuWGDHeZQERAT
-xPmRFHaLP9/ZhvGNh6okIYtrKjWTLGoXvKLHcrKNisBLSq+P2WeFrlme1vjvJMo/
-jPwLT5o9CADQmcbKZ+QQ1ZM9v99iDZol7SAMZX43JC019sx6GK0u6xouJBcLfeB4
-OXacTgmSYdTa9RM9fbfVpti01tJ84LV2SyL/VJq/enJF4XQPSynT/tFTn1PAor6o
-tEAAd8fjKdJ6LnD5wb92SPHfQfXqI84rFEO8rUNIE/1ErT6DYifDzVCbfD2KZdoF
-cOSp7TpD77sY1bs74ocBX5ejKtd+aH99D78bJSMM4pSDZsIEwnomkBHTziubPwJb
-OwnATy0LmSMAWOw5rKbsh5nfwCiUTM20xp0t5JeXd+wPVWbpWqI2EnkCEN+RJr9i
-7dp/ymDQ+Yt5wrsN3NwoyiexPOG91WQVCADdErHsnglVZZq9Z8Wx7KwecGCUurJ2
-H6lKudv5YOxPnAzqZS5HbpZd/nRTMZh2rdXCr5m2YOuewyYjvM757AkmUpM09zJX
-MQ1S67/UX2y8/74TcRF97Ncx9HeELs92innBRXoFitnNguvcO6Esx4BTe1OdU6qR
-ER3zAmVf22Le9ciXbu24DN4mleOH+OmBx7X2PqJSYW9GAMTsRB081R6EWKH7romQ
-waxFrZ4DJzZ9ltyosEJn5F32StyLrFxpcrdLUoEaclZCv2qka7sZvi0EvovDVEBU
-e10jOx9AOwf8Gj2ufhquQ6qgVYCzbP+YrodtkFrXRS3IsljIchj1M2ffB/0bfoUs
-rtER9pLvYzCjBPg8IfGLw0o754Qbhh/ReplCRTusP/fQMybvCvfxreS3oyEriu/G
-GufRomjewZ8EMHDIgUsLcYo2UHZsfF7tcazgxMGmMvazp4r8vpgrvW/8fIN/6Adu
-tF+WjWDTvJLFJCe6O+BFJOWrssNrrra1zGtLC1s8s+Wfpe+bGPL5zpHeebGTwH1U
-22eqgJArlEKxrfarz7W5+uHZJHSjF/K9ZvunLGD0n9GOPMpji3UO3zeM8IYoWn7E
-/EWK1XbjnssNemeeTZ+sDh+qrD7BOi+vCX1IyBxbfqnQfJZvmcPWpruy1UsO+aIC
-0GY8Jr3OL69dDQ21jueJAh8EGAEIAAkFAlC9+dkCGwwACgkQL0VeKCTRjd9HCw/+
-LQSVgLLF4ulYlPCjWIIuQwrPbJfWUVVr2dPUFVM85DCv8gBzk5c121snXh9Swovm
-laBbw6ate3BmbXLh64jVE9Za5sbTWi7PCcbO/bpRy4d6oLmitmNw6cq0vjTLxUYy
-bwuiJxWREkfxuU85EKdouN062YDevH+/YResmlJrcCE7LRlJFeRlKsrrwBU3BqYd
-GgFJjKjQC1peeQ9fj62Y7xfwE9+PXbkiWO5u/Bk8hb1VZH1SoIRU98NHVcp6BVvp
-VK0jLAXuSauSczULmpRjbyt1lhaAqivDTWEEZXiNNbRyp17c3nVdPWOcgBr42hdQ
-z25CgZgyLCsvu82wuXLKJblrIPJX3Yf+si6KqEWBsmwdOWybsjygaF5HvzgFqAAD
-U0goPWoQ71PorP2XOUNp5ZLkBQp5etvtkksjVNMIhnHn8PGMuoxO39EUGlWj2B5l
-Cu8tSosAzB1pS8NcLZzoNoI9dOHrmgJmP+GrOUkcf5GhNZbMoj4GNfGBRYX0SZlQ
-GuDrwNKYj73C4MWyNnnUFyq8nDHJ/G1NpaF2hiof9RBL4PUU/f92JkceXPBXA8gL
-Mz2ig1OButwPPLFGQhWqxXAGrsS3Ny+BhTJfnfIbbkaLLphBpDZm1D9XKbAUvdd1
-RZXoH+FTg9UAW87eqU610npOkT6cRaBxaMK/mDtGNdc=
-=JTFu
------END PGP PRIVATE KEY BLOCK-----
-"""
+import os
def load_tests():
diff --git a/common/src/leap/soledad/common/tests/test_couch.py b/common/src/leap/soledad/common/tests/test_couch.py
index 10d6c136..d2aef9bb 100644
--- a/common/src/leap/soledad/common/tests/test_couch.py
+++ b/common/src/leap/soledad/common/tests/test_couch.py
@@ -20,134 +20,21 @@
Test ObjectStore and Couch backend bits.
"""
-import re
-import copy
-import shutil
-from base64 import b64decode
-from mock import Mock
-from urlparse import urljoin
+import simplejson as json
+
+from urlparse import urljoin
from u1db import errors as u1db_errors
from couchdb.client import Server
-from leap.common.files import mkdir_p
+from testscenarios import TestWithScenarios
+
+from leap.soledad.common import couch, errors
from leap.soledad.common.tests import u1db_tests as tests
from leap.soledad.common.tests.u1db_tests import test_backends
from leap.soledad.common.tests.u1db_tests import test_sync
-from leap.soledad.common import couch, errors
-import simplejson as json
-
-
-#-----------------------------------------------------------------------------
-# A wrapper for running couchdb locally.
-#-----------------------------------------------------------------------------
-
-import re
-import os
-import tempfile
-import subprocess
-import time
-import unittest
-
-
-# from: https://github.com/smcq/paisley/blob/master/paisley/test/util.py
-# TODO: include license of above project.
-class CouchDBWrapper(object):
- """
- Wrapper for external CouchDB instance which is started and stopped for
- testing.
- """
-
- def start(self):
- """
- Start a CouchDB instance for a test.
- """
- self.tempdir = tempfile.mkdtemp(suffix='.couch.test')
-
- path = os.path.join(os.path.dirname(__file__),
- 'couchdb.ini.template')
- handle = open(path)
- conf = handle.read() % {
- 'tempdir': self.tempdir,
- }
- handle.close()
-
- confPath = os.path.join(self.tempdir, 'test.ini')
- handle = open(confPath, 'w')
- handle.write(conf)
- handle.close()
-
- # create the dirs from the template
- mkdir_p(os.path.join(self.tempdir, 'lib'))
- mkdir_p(os.path.join(self.tempdir, 'log'))
- args = ['couchdb', '-n', '-a', confPath]
- null = open('/dev/null', 'w')
-
- self.process = subprocess.Popen(
- args, env=None, stdout=null.fileno(), stderr=null.fileno(),
- close_fds=True)
- # find port
- logPath = os.path.join(self.tempdir, 'log', 'couch.log')
- while not os.path.exists(logPath):
- if self.process.poll() is not None:
- got_stdout, got_stderr = "", ""
- if self.process.stdout is not None:
- got_stdout = self.process.stdout.read()
-
- if self.process.stderr is not None:
- got_stderr = self.process.stderr.read()
- raise Exception("""
-couchdb exited with code %d.
-stdout:
-%s
-stderr:
-%s""" % (
- self.process.returncode, got_stdout, got_stderr))
- time.sleep(0.01)
- while os.stat(logPath).st_size == 0:
- time.sleep(0.01)
- PORT_RE = re.compile(
- 'Apache CouchDB has started on http://127.0.0.1:(?P<port>\d+)')
-
- handle = open(logPath)
- line = handle.read()
- handle.close()
- m = PORT_RE.search(line)
- if not m:
- self.stop()
- raise Exception("Cannot find port in line %s" % line)
- self.port = int(m.group('port'))
-
- def stop(self):
- """
- Terminate the CouchDB instance.
- """
- self.process.terminate()
- self.process.communicate()
- shutil.rmtree(self.tempdir)
-
-
-class CouchDBTestCase(unittest.TestCase):
- """
- TestCase base class for tests against a real CouchDB server.
- """
-
- @classmethod
- def setUpClass(cls):
- """
- Make sure we have a CouchDB instance for a test.
- """
- cls.wrapper = CouchDBWrapper()
- cls.wrapper.start()
- #self.db = self.wrapper.db
-
- @classmethod
- def tearDownClass(cls):
- """
- Stop CouchDB instance for test.
- """
- cls.wrapper.stop()
+from leap.soledad.common.tests.util import CouchDBTestCase
#-----------------------------------------------------------------------------
@@ -239,7 +126,8 @@ COUCH_SCENARIOS = [
]
-class CouchTests(test_backends.AllDatabaseTests, CouchDBTestCase):
+class CouchTests(
+ TestWithScenarios, test_backends.AllDatabaseTests, CouchDBTestCase):
scenarios = COUCH_SCENARIOS
@@ -262,7 +150,8 @@ class CouchTests(test_backends.AllDatabaseTests, CouchDBTestCase):
test_backends.AllDatabaseTests.tearDown(self)
-class CouchDatabaseTests(test_backends.LocalDatabaseTests, CouchDBTestCase):
+class CouchDatabaseTests(
+ TestWithScenarios, test_backends.LocalDatabaseTests, CouchDBTestCase):
scenarios = COUCH_SCENARIOS
@@ -271,7 +160,7 @@ class CouchDatabaseTests(test_backends.LocalDatabaseTests, CouchDBTestCase):
test_backends.LocalDatabaseTests.tearDown(self)
-class CouchValidateGenNTransIdTests(
+class CouchValidateGenNTransIdTests(TestWithScenarios,
test_backends.LocalDatabaseValidateGenNTransIdTests, CouchDBTestCase):
scenarios = COUCH_SCENARIOS
@@ -281,7 +170,7 @@ class CouchValidateGenNTransIdTests(
test_backends.LocalDatabaseValidateGenNTransIdTests.tearDown(self)
-class CouchValidateSourceGenTests(
+class CouchValidateSourceGenTests(TestWithScenarios,
test_backends.LocalDatabaseValidateSourceGenTests, CouchDBTestCase):
scenarios = COUCH_SCENARIOS
@@ -291,7 +180,7 @@ class CouchValidateSourceGenTests(
test_backends.LocalDatabaseValidateSourceGenTests.tearDown(self)
-class CouchWithConflictsTests(
+class CouchWithConflictsTests(TestWithScenarios,
test_backends.LocalDatabaseWithConflictsTests, CouchDBTestCase):
scenarios = COUCH_SCENARIOS
@@ -325,23 +214,11 @@ simple_doc = tests.simple_doc
nested_doc = tests.nested_doc
-class CouchDatabaseSyncTargetTests(test_sync.DatabaseSyncTargetTests,
- CouchDBTestCase):
+class CouchDatabaseSyncTargetTests(
+ TestWithScenarios, test_sync.DatabaseSyncTargetTests, CouchDBTestCase):
scenarios = (tests.multiply_scenarios(COUCH_SCENARIOS, target_scenarios))
- def setUp(self):
- # we implement parents' setUp methods here to prevent from launching
- # more couch instances then needed.
- tests.TestCase.setUp(self)
- self.server = self.server_thread = None
- self.db, self.st = self.create_db_and_target(self)
- self.other_changes = []
-
- def tearDown(self):
- self.db.delete_database()
- test_sync.DatabaseSyncTargetTests.tearDown(self)
-
def test_sync_exchange_returns_many_new_docs(self):
# This test was replicated to allow dictionaries to be compared after
# JSON expansion (because one dictionary may have many different
@@ -372,7 +249,7 @@ from u1db.backends.inmemory import InMemoryIndex
class IndexedCouchDatabase(couch.CouchDatabase):
def __init__(self, url, dbname, replica_uid=None, ensure_ddocs=True):
- old_class.__init__(self, url, dbname, replica_uid=replica_uid,
+ old_class.__init__(self, url, dbname, replica_uid=replica_uid,
ensure_ddocs=ensure_ddocs)
self._indexes = {}
@@ -458,7 +335,8 @@ for name, scenario in COUCH_SCENARIOS:
scenario = dict(scenario)
-class CouchDatabaseSyncTests(test_sync.DatabaseSyncTests, CouchDBTestCase):
+class CouchDatabaseSyncTests(
+ TestWithScenarios, test_sync.DatabaseSyncTests, CouchDBTestCase):
scenarios = sync_scenarios
@@ -498,6 +376,7 @@ class CouchDatabaseExceptionsTests(CouchDBTestCase):
def tearDown(self):
self.db.delete_database()
self.db.close()
+ CouchDBTestCase.tearDown(self)
def test_missing_design_doc_raises(self):
"""
@@ -670,6 +549,3 @@ class CouchDatabaseExceptionsTests(CouchDBTestCase):
self.assertRaises(
errors.MissingDesignDocDeletedError,
self.db._do_set_replica_gen_and_trans_id, 1, 2, 3)
-
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py b/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py
index 6465eb80..83cee469 100644
--- a/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py
+++ b/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py
@@ -15,26 +15,25 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
-Test atomocity for couch operations.
+Test atomicity of couch operations.
"""
import os
-import mock
import tempfile
import threading
-
from urlparse import urljoin
-
+from twisted.internet import defer
from leap.soledad.client import Soledad
from leap.soledad.common.couch import CouchDatabase, CouchServerState
-from leap.soledad.common.tests.test_couch import CouchDBTestCase
-from leap.soledad.common.tests.u1db_tests import TestCaseWithServer
-from leap.soledad.common.tests.test_sync_target import (
+
+from leap.soledad.common.tests.util import (
make_token_soledad_app,
- make_leap_document_for_test,
- token_leap_sync_target,
+ make_soledad_document_for_test,
+ token_soledad_sync_target,
)
+from leap.soledad.common.tests.test_couch import CouchDBTestCase
+from leap.soledad.common.tests.u1db_tests import TestCaseWithServer
from leap.soledad.common.tests.test_server import _couch_ensure_database
@@ -52,15 +51,15 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
def make_app_after_state(state):
return make_token_soledad_app(state)
- make_document_for_test = make_leap_document_for_test
+ make_document_for_test = make_soledad_document_for_test
- sync_target = token_leap_sync_target
+ sync_target = token_soledad_sync_target
def _soledad_instance(self, user='user-uuid', passphrase=u'123',
prefix='',
- secrets_path=Soledad.STORAGE_SECRETS_FILE_NAME,
+ secrets_path='secrets.json',
local_db_path='soledad.u1db', server_url='',
- cert_file=None, auth_token=None, secret_id=None):
+ cert_file=None, auth_token=None):
"""
Instantiate Soledad.
"""
@@ -70,19 +69,6 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
def _put_doc_side_effect(doc):
self._doc_put = doc
- # we need a mocked shared db or else Soledad will try to access the
- # network to find if there are uploaded secrets.
- class MockSharedDB(object):
-
- get_doc = mock.Mock(return_value=None)
- put_doc = mock.Mock(side_effect=_put_doc_side_effect)
- lock = mock.Mock(return_value=('atoken', 300))
- unlock = mock.Mock()
-
- def __call__(self):
- return self
-
- Soledad._shared_db = MockSharedDB()
return Soledad(
user,
passphrase,
@@ -92,7 +78,7 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
server_url=server_url,
cert_file=cert_file,
auth_token=auth_token,
- secret_id=secret_id)
+ shared_db=self.get_default_shared_mock(_put_doc_side_effect))
def make_app(self):
self.request_state = CouchServerState(self._couch_url, 'shared',
@@ -126,7 +112,6 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
puts.
"""
doc = self.db.create_doc({'ops': 0})
- ops = 1
docs = [doc.doc_id]
for i in range(0, REPEAT_TIMES):
self.assertEqual(
@@ -183,24 +168,27 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
auth_token='auth-token',
server_url=self.getURL())
- def _create_docs_and_sync(sol, syncs):
- # create a lot of documents
- for i in range(0, REPEAT_TIMES):
- sol.create_doc({})
+ def _create_docs(results):
+ deferreds = []
+ for i in xrange(0, REPEAT_TIMES):
+ deferreds.append(sol.create_doc({}))
+ return defer.DeferredList(deferreds)
+
+ def _assert_transaction_and_sync_logs(results, sync_idx):
# assert sizes of transaction and sync logs
self.assertEqual(
- syncs*REPEAT_TIMES,
+ sync_idx*REPEAT_TIMES,
len(self.db._get_transaction_log()))
self.assertEqual(
- 1 if syncs > 0 else 0,
+ 1 if sync_idx > 0 else 0,
len(self.db._database.view('syncs/log').rows))
- # sync to the remote db
- sol.sync()
- gen, docs = self.db.get_all_docs()
- self.assertEqual((syncs+1)*REPEAT_TIMES, gen)
- self.assertEqual((syncs+1)*REPEAT_TIMES, len(docs))
+
+ def _assert_sync(results, sync_idx):
+ gen, docs = results
+ self.assertEqual((sync_idx+1)*REPEAT_TIMES, gen)
+ self.assertEqual((sync_idx+1)*REPEAT_TIMES, len(docs))
# assert sizes of transaction and sync logs
- self.assertEqual((syncs+1)*REPEAT_TIMES,
+ self.assertEqual((sync_idx+1)*REPEAT_TIMES,
len(self.db._get_transaction_log()))
sync_log_rows = self.db._database.view('syncs/log').rows
sync_log = sync_log_rows[0].value
@@ -210,14 +198,32 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
# assert sync_log has exactly 1 row
self.assertEqual(1, len(sync_log_rows))
# assert it has the correct replica_uid, gen and trans_id
- self.assertEqual(sol._db._replica_uid, replica_uid)
- sol_gen, sol_trans_id = sol._db._get_generation_info()
+ self.assertEqual(sol._dbpool.replica_uid, replica_uid)
+ conn_key = sol._dbpool._u1dbconnections.keys().pop()
+ conn = sol._dbpool._u1dbconnections[conn_key]
+ sol_gen, sol_trans_id = conn._get_generation_info()
self.assertEqual(sol_gen, known_gen)
self.assertEqual(sol_trans_id, known_trans_id)
+
+ # create some documents
+ d = _create_docs(None)
- _create_docs_and_sync(sol, 0)
- _create_docs_and_sync(sol, 1)
- sol.close()
+ # sync first time and assert success
+ d.addCallback(_assert_transaction_and_sync_logs, 0)
+ d.addCallback(lambda _: sol.sync())
+ d.addCallback(lambda _: sol.get_all_docs())
+ d.addCallback(_assert_sync, 0)
+
+ # create more docs, sync second time and assert success
+ d.addCallback(_create_docs)
+ d.addCallback(_assert_transaction_and_sync_logs, 1)
+ d.addCallback(lambda _: sol.sync())
+ d.addCallback(lambda _: sol.get_all_docs())
+ d.addCallback(_assert_sync, 1)
+
+ d.addCallback(lambda _: sol.close())
+
+ return d
#
# Concurrency tests
@@ -313,86 +319,76 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
"""
Assert that the sync_log is correct after concurrent syncs.
"""
- threads = []
docs = []
- pool = threading.BoundedSemaphore(value=1)
+
self.startServer()
+
sol = self._soledad_instance(
auth_token='auth-token',
server_url=self.getURL())
- def _run_method(self):
- # create a lot of documents
- doc = self._params['sol'].create_doc({})
- pool.acquire()
- docs.append(doc.doc_id)
- pool.release()
+ def _save_doc_ids(results):
+ for doc in results:
+ docs.append(doc.doc_id)
- # launch threads to create documents in parallel
+ # create documents in parallel
+ deferreds = []
for i in range(0, REPEAT_TIMES):
- thread = self._WorkerThread(
- {'sol': sol, 'syncs': i},
- _run_method)
- thread.start()
- threads.append(thread)
+ d = sol.create_doc({})
+ deferreds.append(d)
- # wait for threads to finish
- for thread in threads:
- thread.join()
+ # wait for documents creation and sync
+ d = defer.gatherResults(deferreds)
+ d.addCallback(_save_doc_ids)
+ d.addCallback(lambda _: sol.sync())
- # do the sync!
- sol.sync()
+ def _assert_logs(results):
+ transaction_log = self.db._get_transaction_log()
+ self.assertEqual(REPEAT_TIMES, len(transaction_log))
+ # assert all documents are in the remote log
+ self.assertEqual(REPEAT_TIMES, len(docs))
+ for doc_id in docs:
+ self.assertEqual(
+ 1,
+ len(filter(lambda t: t[0] == doc_id, transaction_log)))
- transaction_log = self.db._get_transaction_log()
- self.assertEqual(REPEAT_TIMES, len(transaction_log))
- # assert all documents are in the remote log
- self.assertEqual(REPEAT_TIMES, len(docs))
- for doc_id in docs:
- self.assertEqual(
- 1,
- len(filter(lambda t: t[0] == doc_id, transaction_log)))
- sol.close()
+ d.addCallback(_assert_logs)
+ d.addCallback(lambda _: sol.close())
+
+ return d
def test_concurrent_syncs_do_not_fail(self):
"""
Assert that concurrent attempts to sync end up being executed
sequentially and do not fail.
"""
- threads = []
docs = []
- pool = threading.BoundedSemaphore(value=1)
+
self.startServer()
+
sol = self._soledad_instance(
auth_token='auth-token',
server_url=self.getURL())
- def _run_method(self):
- # create a lot of documents
- doc = self._params['sol'].create_doc({})
- # do the sync!
- sol.sync()
- pool.acquire()
- docs.append(doc.doc_id)
- pool.release()
-
- # launch threads to create documents in parallel
- for i in range(0, REPEAT_TIMES):
- thread = self._WorkerThread(
- {'sol': sol, 'syncs': i},
- _run_method)
- thread.start()
- threads.append(thread)
-
- # wait for threads to finish
- for thread in threads:
- thread.join()
-
- transaction_log = self.db._get_transaction_log()
- self.assertEqual(REPEAT_TIMES, len(transaction_log))
- # assert all documents are in the remote log
- self.assertEqual(REPEAT_TIMES, len(docs))
- for doc_id in docs:
- self.assertEqual(
- 1,
- len(filter(lambda t: t[0] == doc_id, transaction_log)))
- sol.close()
+ deferreds = []
+ for i in xrange(0, REPEAT_TIMES):
+ d = sol.create_doc({})
+ d.addCallback(lambda doc: docs.append(doc.doc_id))
+ d.addCallback(lambda _: sol.sync())
+ deferreds.append(d)
+
+ def _assert_logs(results):
+ transaction_log = self.db._get_transaction_log()
+ self.assertEqual(REPEAT_TIMES, len(transaction_log))
+ # assert all documents are in the remote log
+ self.assertEqual(REPEAT_TIMES, len(docs))
+ for doc_id in docs:
+ self.assertEqual(
+ 1,
+ len(filter(lambda t: t[0] == doc_id, transaction_log)))
+
+ d = defer.gatherResults(deferreds)
+ d.addCallback(_assert_logs)
+ d.addCallback(lambda _: sol.close())
+
+ return d
diff --git a/common/src/leap/soledad/common/tests/test_crypto.py b/common/src/leap/soledad/common/tests/test_crypto.py
index f5fb4b7a..fdad8aac 100644
--- a/common/src/leap/soledad/common/tests/test_crypto.py
+++ b/common/src/leap/soledad/common/tests/test_crypto.py
@@ -23,7 +23,7 @@ import binascii
from leap.soledad.client import crypto
from leap.soledad.common.document import SoledadDocument
-from leap.soledad.common.tests import BaseSoledadTest
+from leap.soledad.common.tests.util import BaseSoledadTest
from leap.soledad.common.crypto import WrongMacError
from leap.soledad.common.crypto import UnknownMacMethodError
from leap.soledad.common.crypto import EncryptionMethods
@@ -82,7 +82,7 @@ class RecoveryDocumentTestCase(BaseSoledadTest):
rd = self._soledad.secrets._export_recovery_document()
s = self._soledad_instance()
s.secrets._import_recovery_document(rd)
- s.set_secret_id(self._soledad.secrets._secret_id)
+ s.secrets.set_secret_id(self._soledad.secrets._secret_id)
self.assertEqual(self._soledad.storage_secret,
s.storage_secret,
'Failed settinng secret for symmetric encryption.')
@@ -95,7 +95,7 @@ class SoledadSecretsTestCase(BaseSoledadTest):
# instantiate and save secret_id
sol = self._soledad_instance(user='user@leap.se')
self.assertTrue(len(sol.secrets._secrets) == 1)
- secret_id_1 = sol.secret_id
+ secret_id_1 = sol.secrets.secret_id
# assert id is hash of secret
self.assertTrue(
secret_id_1 == hashlib.sha256(sol.storage_secret).hexdigest())
@@ -104,9 +104,8 @@ class SoledadSecretsTestCase(BaseSoledadTest):
self.assertTrue(secret_id_1 != secret_id_2)
sol.close()
# re-instantiate
- sol = self._soledad_instance(
- user='user@leap.se',
- secret_id=secret_id_1)
+ sol = self._soledad_instance(user='user@leap.se')
+ sol.secrets.set_secret_id(secret_id_1)
# assert ids are valid
self.assertTrue(len(sol.secrets._secrets) == 2)
self.assertTrue(secret_id_1 in sol.secrets._secrets)
@@ -117,7 +116,7 @@ class SoledadSecretsTestCase(BaseSoledadTest):
secret_length = sol.secrets.GEN_SECRET_LENGTH
self.assertTrue(len(sol.storage_secret) == secret_length)
# assert format of secret 2
- sol.set_secret_id(secret_id_2)
+ sol.secrets.set_secret_id(secret_id_2)
self.assertTrue(sol.storage_secret is not None)
self.assertIsInstance(sol.storage_secret, str)
self.assertTrue(len(sol.storage_secret) == secret_length)
@@ -134,12 +133,12 @@ class SoledadSecretsTestCase(BaseSoledadTest):
"Should have a secret at this point")
# setting secret id to None should not interfere in the fact we have a
# secret.
- sol.set_secret_id(None)
+ sol.secrets.set_secret_id(None)
self.assertTrue(
sol.secrets._has_secret(),
"Should have a secret at this point")
# but not being able to decrypt correctly should
- sol.secrets._secrets[sol.secret_id] = None
+ sol.secrets._secrets[sol.secrets.secret_id] = None
self.assertFalse(sol.secrets._has_secret())
sol.close()
diff --git a/common/src/leap/soledad/common/tests/test_http.py b/common/src/leap/soledad/common/tests/test_http.py
index d21470e0..1f661b77 100644
--- a/common/src/leap/soledad/common/tests/test_http.py
+++ b/common/src/leap/soledad/common/tests/test_http.py
@@ -20,8 +20,6 @@ Test Leap backend bits: test http database
from u1db.remote import http_database
from leap.soledad.client import auth
-
-from leap.soledad.common.tests import u1db_tests as tests
from leap.soledad.common.tests.u1db_tests import test_http_database
@@ -59,6 +57,3 @@ class TestHTTPDatabaseWithCreds(
'token': 'auth-token',
}})
self.assertIn('token', db1._creds)
-
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_http_client.py b/common/src/leap/soledad/common/tests/test_http_client.py
index 3169398b..db731c32 100644
--- a/common/src/leap/soledad/common/tests/test_http_client.py
+++ b/common/src/leap/soledad/common/tests/test_http_client.py
@@ -21,8 +21,9 @@ import json
from u1db.remote import http_client
+from testscenarios import TestWithScenarios
+
from leap.soledad.client import auth
-from leap.soledad.common.tests import u1db_tests as tests
from leap.soledad.common.tests.u1db_tests import test_http_client
from leap.soledad.server.auth import SoledadTokenAuthMiddleware
@@ -31,7 +32,9 @@ from leap.soledad.server.auth import SoledadTokenAuthMiddleware
# The following tests come from `u1db.tests.test_http_client`.
#-----------------------------------------------------------------------------
-class TestSoledadClientBase(test_http_client.TestHTTPClientBase):
+class TestSoledadClientBase(
+ TestWithScenarios,
+ test_http_client.TestHTTPClientBase):
"""
This class should be used to test Token auth.
"""
@@ -90,7 +93,7 @@ class TestSoledadClientBase(test_http_client.TestHTTPClientBase):
"message": e.message})]
uuid, token = encoded.decode('base64').split(':', 1)
if uuid != 'user-uuid' and token != 'auth-token':
- return unauth_err("Incorrect address or token.")
+ return Exception("Incorrect address or token.")
start_response("200 OK", [('Content-Type', 'application/json')])
return [json.dumps([environ['PATH_INFO'], uuid, token])]
@@ -112,5 +115,3 @@ class TestSoledadClientBase(test_http_client.TestHTTPClientBase):
res, headers = cli._request('GET', ['doc', 'token'])
self.assertEqual(
['/dbase/doc/token', 'user-uuid', 'auth-token'], json.loads(res))
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_https.py b/common/src/leap/soledad/common/tests/test_https.py
index b6288188..4dd55754 100644
--- a/common/src/leap/soledad/common/tests/test_https.py
+++ b/common/src/leap/soledad/common/tests/test_https.py
@@ -14,30 +14,35 @@
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+
"""
Test Leap backend bits: https
"""
-from leap.soledad.common.tests import BaseSoledadTest
-from leap.soledad.common.tests import test_sync_target as test_st
-from leap.soledad.common.tests import u1db_tests as tests
-from leap.soledad.common.tests.u1db_tests import test_backends
-from leap.soledad.common.tests.u1db_tests import test_https
-from leap.soledad import client
-from leap.soledad.server import SoledadApp
from u1db.remote import http_client
+from leap.soledad import client
+
+from testscenarios import TestWithScenarios
+
+from leap.soledad.common.tests.u1db_tests import test_backends
+from leap.soledad.common.tests.u1db_tests import test_https
+from leap.soledad.common.tests.util import (
+ BaseSoledadTest,
+ make_soledad_document_for_test,
+ make_soledad_app,
+ make_token_soledad_app,
+)
-def make_soledad_app(state):
- return SoledadApp(state)
LEAP_SCENARIOS = [
('http', {
'make_database_for_test': test_backends.make_http_database_for_test,
'copy_database_for_test': test_backends.copy_http_database_for_test,
- 'make_document_for_test': test_st.make_leap_document_for_test,
- 'make_app_with_state': test_st.make_soledad_app}),
+ 'make_document_for_test': make_soledad_document_for_test,
+ 'make_app_with_state': make_soledad_app}),
]
@@ -55,14 +60,15 @@ def token_leap_https_sync_target(test, host, path):
class TestSoledadSyncTargetHttpsSupport(
+ TestWithScenarios,
test_https.TestHttpSyncTargetHttpsSupport,
BaseSoledadTest):
scenarios = [
('token_soledad_https',
{'server_def': test_https.https_server_def,
- 'make_app_with_state': test_st.make_token_soledad_app,
- 'make_document_for_test': test_st.make_leap_document_for_test,
+ 'make_app_with_state': make_token_soledad_app,
+ 'make_document_for_test': make_soledad_document_for_test,
'sync_target': token_leap_https_sync_target}),
]
@@ -71,8 +77,8 @@ class TestSoledadSyncTargetHttpsSupport(
# run smoothly with standard u1db.
test_https.TestHttpSyncTargetHttpsSupport.setUp(self)
# so here monkey patch again to test our functionality.
- http_client._VerifiedHTTPSConnection = client.VerifiedHTTPSConnection
- client.SOLEDAD_CERT = http_client.CA_CERTS
+ http_client._VerifiedHTTPSConnection = client.api.VerifiedHTTPSConnection
+ client.api.SOLEDAD_CERT = http_client.CA_CERTS
def test_working(self):
"""
@@ -83,7 +89,7 @@ class TestSoledadSyncTargetHttpsSupport(
"""
self.startServer()
db = self.request_state._create_database('test')
- self.patch(client, 'SOLEDAD_CERT', self.cacert_pem)
+ self.patch(client.api, 'SOLEDAD_CERT', self.cacert_pem)
remote_target = self.getSyncTarget('localhost', 'test')
remote_target.record_sync_info('other-id', 2, 'T-id')
self.assertEqual(
@@ -99,10 +105,8 @@ class TestSoledadSyncTargetHttpsSupport(
"""
self.startServer()
self.request_state._create_database('test')
- self.patch(client, 'SOLEDAD_CERT', self.cacert_pem)
+ self.patch(client.api, 'SOLEDAD_CERT', self.cacert_pem)
remote_target = self.getSyncTarget('127.0.0.1', 'test')
self.assertRaises(
http_client.CertificateError, remote_target.record_sync_info,
'other-id', 2, 'T-id')
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_server.py b/common/src/leap/soledad/common/tests/test_server.py
index acd0a54c..836bd74a 100644
--- a/common/src/leap/soledad/common/tests/test_server.py
+++ b/common/src/leap/soledad/common/tests/test_server.py
@@ -19,29 +19,28 @@ Tests for server-related functionality.
"""
import os
import tempfile
-import simplejson as json
import mock
import time
import binascii
from urlparse import urljoin
+from twisted.internet import defer
-from leap.common.testing.basetest import BaseLeapTest
from leap.soledad.common.couch import (
CouchServerState,
CouchDatabase,
)
-from leap.soledad.common.tests.u1db_tests import (
- TestCaseWithServer,
- simple_doc,
-)
+from leap.soledad.common.tests.u1db_tests import TestCaseWithServer
from leap.soledad.common.tests.test_couch import CouchDBTestCase
-from leap.soledad.common.tests.test_target_soledad import (
+from leap.soledad.common.tests.util import (
make_token_soledad_app,
- make_leap_document_for_test,
+ make_soledad_document_for_test,
+ token_soledad_sync_target,
+ BaseSoledadTest,
)
-from leap.soledad.common.tests.test_sync_target import token_leap_sync_target
-from leap.soledad.client import Soledad, crypto
+
+from leap.soledad.common import crypto
+from leap.soledad.client import Soledad
from leap.soledad.server import LockResource
from leap.soledad.server.auth import URLToAuthorization
@@ -58,7 +57,7 @@ def _couch_ensure_database(self, dbname):
CouchServerState.ensure_database = _couch_ensure_database
-class ServerAuthorizationTestCase(BaseLeapTest):
+class ServerAuthorizationTestCase(BaseSoledadTest):
"""
Tests related to Soledad server authorization.
"""
@@ -272,19 +271,24 @@ class EncryptedSyncTestCase(
Tests for encrypted sync using Soledad server backed by a couch database.
"""
+ # increase twisted.trial's timeout because large files syncing might take
+ # some time to finish.
+ timeout = 500
+
@staticmethod
def make_app_with_state(state):
return make_token_soledad_app(state)
- make_document_for_test = make_leap_document_for_test
+ make_document_for_test = make_soledad_document_for_test
- sync_target = token_leap_sync_target
+ sync_target = token_soledad_sync_target
def _soledad_instance(self, user='user-uuid', passphrase=u'123',
prefix='',
- secrets_path=Soledad.STORAGE_SECRETS_FILE_NAME,
- local_db_path='soledad.u1db', server_url='',
- cert_file=None, auth_token=None, secret_id=None):
+ secrets_path='secrets.json',
+ local_db_path='soledad.u1db',
+ server_url='',
+ cert_file=None, auth_token=None):
"""
Instantiate Soledad.
"""
@@ -294,20 +298,15 @@ class EncryptedSyncTestCase(
def _put_doc_side_effect(doc):
self._doc_put = doc
- # we need a mocked shared db or else Soledad will try to access the
- # network to find if there are uploaded secrets.
- class MockSharedDB(object):
-
- get_doc = mock.Mock(return_value=None)
- put_doc = mock.Mock(side_effect=_put_doc_side_effect)
- lock = mock.Mock(return_value=('atoken', 300))
- unlock = mock.Mock()
- close = mock.Mock()
-
- def __call__(self):
- return self
+ if not server_url:
+ # attempt to find the soledad server url
+ server_address = None
+ server = getattr(self, 'server', None)
+ if server:
+ server_address = getattr(self.server, 'server_address', None)
+ if server_address:
+ server_url = 'http://%s:%d' % (server_address)
- Soledad._shared_db = MockSharedDB()
return Soledad(
user,
passphrase,
@@ -317,7 +316,7 @@ class EncryptedSyncTestCase(
server_url=server_url,
cert_file=cert_file,
auth_token=auth_token,
- secret_id=secret_id)
+ shared_db=self.get_default_shared_mock(_put_doc_side_effect))
def make_app(self):
self.request_state = CouchServerState(self._couch_url, 'shared',
@@ -325,70 +324,122 @@ class EncryptedSyncTestCase(
return self.make_app_with_state(self.request_state)
def setUp(self):
- TestCaseWithServer.setUp(self)
+ # the order of the following initializations is crucial because of
+ # dependencies.
+ # XXX explain better
CouchDBTestCase.setUp(self)
- self.tempdir = tempfile.mkdtemp(prefix="leap_tests-")
self._couch_url = 'http://localhost:' + str(self.wrapper.port)
+ self.tempdir = tempfile.mkdtemp(prefix="leap_tests-")
+ TestCaseWithServer.setUp(self)
def tearDown(self):
CouchDBTestCase.tearDown(self)
TestCaseWithServer.tearDown(self)
- def test_encrypted_sym_sync(self):
+ def _test_encrypted_sym_sync(self, passphrase=u'123', doc_size=2,
+ number_of_docs=1):
"""
Test the complete syncing chain between two soledad dbs using a
Soledad server backed by a couch database.
"""
self.startServer()
+
# instantiate soledad and create a document
sol1 = self._soledad_instance(
# token is verified in test_target.make_token_soledad_app
- auth_token='auth-token'
- )
- _, doclist = sol1.get_all_docs()
- self.assertEqual([], doclist)
- doc1 = sol1.create_doc(json.loads(simple_doc))
+ auth_token='auth-token',
+ passphrase=passphrase)
+
+ # instantiate another soledad using the same secret as the previous
+ # one (so we can correctly verify the mac of the synced document)
+ sol2 = self._soledad_instance(
+ prefix='x',
+ auth_token='auth-token',
+ secrets_path=sol1._secrets_path,
+ passphrase=passphrase)
+
# ensure remote db exists before syncing
db = CouchDatabase.open_database(
urljoin(self._couch_url, 'user-user-uuid'),
create=True,
ensure_ddocs=True)
- # sync with server
- sol1._server_url = self.getURL()
- sol1.sync()
- # assert doc was sent to couch db
- _, doclist = db.get_all_docs()
- self.assertEqual(1, len(doclist))
- couchdoc = doclist[0]
- # assert document structure in couch server
- self.assertEqual(doc1.doc_id, couchdoc.doc_id)
- self.assertEqual(doc1.rev, couchdoc.rev)
- self.assertEqual(6, len(couchdoc.content))
- self.assertTrue(crypto.ENC_JSON_KEY in couchdoc.content)
- self.assertTrue(crypto.ENC_SCHEME_KEY in couchdoc.content)
- self.assertTrue(crypto.ENC_METHOD_KEY in couchdoc.content)
- self.assertTrue(crypto.ENC_IV_KEY in couchdoc.content)
- self.assertTrue(crypto.MAC_KEY in couchdoc.content)
- self.assertTrue(crypto.MAC_METHOD_KEY in couchdoc.content)
- # instantiate soledad with empty db, but with same secrets path
- sol2 = self._soledad_instance(prefix='x', auth_token='auth-token')
- _, doclist = sol2.get_all_docs()
- self.assertEqual([], doclist)
- sol2.secrets_path = sol1.secrets_path
- sol2.secrets._load_secrets()
- sol2.set_secret_id(sol1.secret_id)
- # sync the new instance
- sol2._server_url = self.getURL()
- sol2.sync()
- _, doclist = sol2.get_all_docs()
- self.assertEqual(1, len(doclist))
- doc2 = doclist[0]
- # assert incoming doc is equal to the first sent doc
- self.assertEqual(doc1, doc2)
- db.delete_database()
- db.close()
- sol1.close()
- sol2.close()
+
+ def _db1AssertEmptyDocList(results):
+ _, doclist = results
+ self.assertEqual([], doclist)
+
+ def _db1CreateDocs(results):
+ deferreds = []
+ for i in xrange(number_of_docs):
+ content = binascii.hexlify(os.urandom(doc_size/2))
+ deferreds.append(sol1.create_doc({'data': content}))
+ return defer.DeferredList(deferreds)
+
+ def _db1AssertDocsSyncedToServer(results):
+ _, sol_doclist = results
+ self.assertEqual(number_of_docs, len(sol_doclist))
+ # assert doc was sent to couch db
+ _, couch_doclist = db.get_all_docs()
+ self.assertEqual(number_of_docs, len(couch_doclist))
+ for i in xrange(number_of_docs):
+ soldoc = sol_doclist.pop()
+ couchdoc = couch_doclist.pop()
+ # assert document structure in couch server
+ self.assertEqual(soldoc.doc_id, couchdoc.doc_id)
+ self.assertEqual(soldoc.rev, couchdoc.rev)
+ self.assertEqual(6, len(couchdoc.content))
+ self.assertTrue(crypto.ENC_JSON_KEY in couchdoc.content)
+ self.assertTrue(crypto.ENC_SCHEME_KEY in couchdoc.content)
+ self.assertTrue(crypto.ENC_METHOD_KEY in couchdoc.content)
+ self.assertTrue(crypto.ENC_IV_KEY in couchdoc.content)
+ self.assertTrue(crypto.MAC_KEY in couchdoc.content)
+ self.assertTrue(crypto.MAC_METHOD_KEY in couchdoc.content)
+
+ d = sol1.get_all_docs()
+ d.addCallback(_db1AssertEmptyDocList)
+ d.addCallback(_db1CreateDocs)
+ d.addCallback(lambda _: sol1.sync())
+ d.addCallback(lambda _: sol1.get_all_docs())
+ d.addCallback(_db1AssertDocsSyncedToServer)
+
+ def _db2AssertEmptyDocList(results):
+ _, doclist = results
+ self.assertEqual([], doclist)
+
+ def _getAllDocsFromBothDbs(results):
+ d1 = sol1.get_all_docs()
+ d2 = sol2.get_all_docs()
+ return defer.DeferredList([d1, d2])
+
+ d.addCallback(lambda _: sol2.get_all_docs())
+ d.addCallback(_db2AssertEmptyDocList)
+ d.addCallback(lambda _: sol2.sync())
+ d.addCallback(_getAllDocsFromBothDbs)
+
+ def _assertDocSyncedFromDb1ToDb2(results):
+ r1, r2 = results
+ _, (gen1, doclist1) = r1
+ _, (gen2, doclist2) = r2
+ self.assertEqual(number_of_docs, gen1)
+ self.assertEqual(number_of_docs, gen2)
+ self.assertEqual(number_of_docs, len(doclist1))
+ self.assertEqual(number_of_docs, len(doclist2))
+ self.assertEqual(doclist1[0], doclist2[0])
+
+ d.addCallback(_assertDocSyncedFromDb1ToDb2)
+
+ def _cleanUp(results):
+ db.delete_database()
+ db.close()
+ sol1.close()
+ sol2.close()
+
+ d.addCallback(_cleanUp)
+
+ return d
+
+ def test_encrypted_sym_sync(self):
+ return self._test_encrypted_sym_sync()
def test_encrypted_sym_sync_with_unicode_passphrase(self):
"""
@@ -396,152 +447,20 @@ class EncryptedSyncTestCase(
Soledad server backed by a couch database, using an unicode
passphrase.
"""
- self.startServer()
- # instantiate soledad and create a document
- sol1 = self._soledad_instance(
- # token is verified in test_target.make_token_soledad_app
- auth_token='auth-token',
- passphrase=u'ãáàäéàëíìïóòöõúùüñç',
- )
- _, doclist = sol1.get_all_docs()
- self.assertEqual([], doclist)
- doc1 = sol1.create_doc(json.loads(simple_doc))
- # ensure remote db exists before syncing
- db = CouchDatabase.open_database(
- urljoin(self._couch_url, 'user-user-uuid'),
- create=True,
- ensure_ddocs=True)
- # sync with server
- sol1._server_url = self.getURL()
- sol1.sync()
- # assert doc was sent to couch db
- _, doclist = db.get_all_docs()
- self.assertEqual(1, len(doclist))
- couchdoc = doclist[0]
- # assert document structure in couch server
- self.assertEqual(doc1.doc_id, couchdoc.doc_id)
- self.assertEqual(doc1.rev, couchdoc.rev)
- self.assertEqual(6, len(couchdoc.content))
- self.assertTrue(crypto.ENC_JSON_KEY in couchdoc.content)
- self.assertTrue(crypto.ENC_SCHEME_KEY in couchdoc.content)
- self.assertTrue(crypto.ENC_METHOD_KEY in couchdoc.content)
- self.assertTrue(crypto.ENC_IV_KEY in couchdoc.content)
- self.assertTrue(crypto.MAC_KEY in couchdoc.content)
- self.assertTrue(crypto.MAC_METHOD_KEY in couchdoc.content)
- # instantiate soledad with empty db, but with same secrets path
- sol2 = self._soledad_instance(
- prefix='x',
- auth_token='auth-token',
- passphrase=u'ãáàäéàëíìïóòöõúùüñç',
- )
- _, doclist = sol2.get_all_docs()
- self.assertEqual([], doclist)
- sol2.secrets_path = sol1.secrets_path
- sol2.secrets._load_secrets()
- sol2.set_secret_id(sol1.secret_id)
- # sync the new instance
- sol2._server_url = self.getURL()
- sol2.sync()
- _, doclist = sol2.get_all_docs()
- self.assertEqual(1, len(doclist))
- doc2 = doclist[0]
- # assert incoming doc is equal to the first sent doc
- self.assertEqual(doc1, doc2)
- db.delete_database()
- db.close()
- sol1.close()
- sol2.close()
+ return self._test_encrypted_sym_sync(passphrase=u'ãáàäéàëíìïóòöõúùüñç')
def test_sync_very_large_files(self):
"""
Test if Soledad can sync very large files.
"""
- # define the size of the "very large file"
length = 100*(10**6) # 100 MB
- self.startServer()
- # instantiate soledad and create a document
- sol1 = self._soledad_instance(
- # token is verified in test_target.make_token_soledad_app
- auth_token='auth-token'
- )
- _, doclist = sol1.get_all_docs()
- self.assertEqual([], doclist)
- content = binascii.hexlify(os.urandom(length/2)) # len() == length
- doc1 = sol1.create_doc({'data': content})
- # ensure remote db exists before syncing
- db = CouchDatabase.open_database(
- urljoin(self._couch_url, 'user-user-uuid'),
- create=True,
- ensure_ddocs=True)
- # sync with server
- sol1._server_url = self.getURL()
- sol1.sync()
- # instantiate soledad with empty db, but with same secrets path
- sol2 = self._soledad_instance(prefix='x', auth_token='auth-token')
- _, doclist = sol2.get_all_docs()
- self.assertEqual([], doclist)
- sol2.secrets_path = sol1.secrets_path
- sol2.secrets._load_secrets()
- sol2.set_secret_id(sol1.secret_id)
- # sync the new instance
- sol2._server_url = self.getURL()
- sol2.sync()
- _, doclist = sol2.get_all_docs()
- self.assertEqual(1, len(doclist))
- doc2 = doclist[0]
- # assert incoming doc is equal to the first sent doc
- self.assertEqual(doc1, doc2)
- # delete remote database
- db.delete_database()
- db.close()
- sol1.close()
- sol2.close()
+ return self._test_encrypted_sym_sync(doc_size=length, number_of_docs=1)
def test_sync_many_small_files(self):
"""
Test if Soledad can sync many smallfiles.
"""
- number_of_docs = 100
- self.startServer()
- # instantiate soledad and create a document
- sol1 = self._soledad_instance(
- # token is verified in test_target.make_token_soledad_app
- auth_token='auth-token'
- )
- _, doclist = sol1.get_all_docs()
- self.assertEqual([], doclist)
- # create many small files
- for i in range(0, number_of_docs):
- sol1.create_doc(json.loads(simple_doc))
- # ensure remote db exists before syncing
- db = CouchDatabase.open_database(
- urljoin(self._couch_url, 'user-user-uuid'),
- create=True,
- ensure_ddocs=True)
- # sync with server
- sol1._server_url = self.getURL()
- sol1.sync()
- # instantiate soledad with empty db, but with same secrets path
- sol2 = self._soledad_instance(prefix='x', auth_token='auth-token')
- _, doclist = sol2.get_all_docs()
- self.assertEqual([], doclist)
- sol2.secrets_path = sol1.secrets_path
- sol2.secrets._load_secrets()
- sol2.set_secret_id(sol1.secret_id)
- # sync the new instance
- sol2._server_url = self.getURL()
- sol2.sync()
- _, doclist = sol2.get_all_docs()
- self.assertEqual(number_of_docs, len(doclist))
- # assert incoming docs are equal to sent docs
- for doc in doclist:
- self.assertEqual(sol1.get_doc(doc.doc_id), doc)
- # delete remote database
- db.delete_database()
- db.close()
- sol1.close()
- sol2.close()
-
+ return self._test_encrypted_sym_sync(doc_size=2, number_of_docs=100)
class LockResourceTestCase(
CouchDBTestCase, TestCaseWithServer):
@@ -553,15 +472,18 @@ class LockResourceTestCase(
def make_app_with_state(state):
return make_token_soledad_app(state)
- make_document_for_test = make_leap_document_for_test
+ make_document_for_test = make_soledad_document_for_test
- sync_target = token_leap_sync_target
+ sync_target = token_soledad_sync_target
def setUp(self):
- TestCaseWithServer.setUp(self)
+ # the order of the following initializations is crucial because of
+ # dependencies.
+ # XXX explain better
CouchDBTestCase.setUp(self)
- self.tempdir = tempfile.mkdtemp(prefix="leap_tests-")
self._couch_url = 'http://localhost:' + str(self.wrapper.port)
+ self.tempdir = tempfile.mkdtemp(prefix="leap_tests-")
+ TestCaseWithServer.setUp(self)
# create the databases
CouchDatabase.open_database(
urljoin(self._couch_url, 'shared'),
@@ -575,14 +497,14 @@ class LockResourceTestCase(
self._couch_url, 'shared', 'tokens')
def tearDown(self):
- CouchDBTestCase.tearDown(self)
- TestCaseWithServer.tearDown(self)
# delete remote database
db = CouchDatabase.open_database(
urljoin(self._couch_url, 'shared'),
create=True,
ensure_ddocs=True)
db.delete_database()
+ CouchDBTestCase.tearDown(self)
+ TestCaseWithServer.tearDown(self)
def test__try_obtain_filesystem_lock(self):
responder = mock.Mock()
diff --git a/common/src/leap/soledad/common/tests/test_soledad.py b/common/src/leap/soledad/common/tests/test_soledad.py
index 31c02fc4..0b49d9f5 100644
--- a/common/src/leap/soledad/common/tests/test_soledad.py
+++ b/common/src/leap/soledad/common/tests/test_soledad.py
@@ -20,9 +20,8 @@ Tests for general Soledad functionality.
import os
from mock import Mock
-
from leap.common.events import events_pb2 as proto
-from leap.soledad.common.tests import (
+from leap.soledad.common.tests.util import (
BaseSoledadTest,
ADDRESS,
)
@@ -30,10 +29,9 @@ from leap import soledad
from leap.soledad.common.document import SoledadDocument
from leap.soledad.common.crypto import WrongMacError
from leap.soledad.client import Soledad
-from leap.soledad.client.sqlcipher import SQLCipherDatabase
+from leap.soledad.client.adbapi import U1DBConnectionPool
from leap.soledad.client.secrets import PassphraseTooShort
from leap.soledad.client.shared_db import SoledadSharedDatabase
-from leap.soledad.client.target import SoledadSyncTarget
class AuxMethodsTestCase(BaseSoledadTest):
@@ -41,18 +39,24 @@ class AuxMethodsTestCase(BaseSoledadTest):
def test__init_dirs(self):
sol = self._soledad_instance(prefix='_init_dirs')
local_db_dir = os.path.dirname(sol.local_db_path)
- secrets_path = os.path.dirname(sol.secrets_path)
+ secrets_path = os.path.dirname(sol.secrets.secrets_path)
self.assertTrue(os.path.isdir(local_db_dir))
self.assertTrue(os.path.isdir(secrets_path))
- sol.close()
- def test__init_db(self):
+ def _close_soledad(results):
+ sol.close()
+
+ d = sol.create_doc({})
+ d.addCallback(_close_soledad)
+ return d
+
+ def test__init_u1db_sqlcipher_backend(self):
sol = self._soledad_instance(prefix='_init_db')
- self.assertIsInstance(sol._db, SQLCipherDatabase)
+ self.assertIsInstance(sol._dbpool, U1DBConnectionPool)
self.assertTrue(os.path.isfile(sol.local_db_path))
sol.close()
- def test__init_config_defaults(self):
+ def test__init_config_with_defaults(self):
"""
Test if configuration defaults point to the correct place.
"""
@@ -62,23 +66,16 @@ class AuxMethodsTestCase(BaseSoledadTest):
def __init__(self):
pass
- # instantiate without initializing so we just test _init_config()
+ # instantiate without initializing so we just test
+ # _init_config_with_defaults()
sol = SoledadMock()
sol._passphrase = u''
- sol._secrets_path = None
- sol._local_db_path = None
sol._server_url = ''
- sol._init_config()
- # assert value of secrets_path
- self.assertEquals(
- os.path.join(
- sol.DEFAULT_PREFIX, Soledad.STORAGE_SECRETS_FILE_NAME),
- sol._secrets_path)
+ sol._init_config_with_defaults()
# assert value of local_db_path
self.assertEquals(
- os.path.join(sol.DEFAULT_PREFIX, 'soledad.u1db'),
+ os.path.join(sol.default_prefix, 'soledad.u1db'),
sol.local_db_path)
- sol.close()
def test__init_config_from_params(self):
"""
@@ -93,43 +90,56 @@ class AuxMethodsTestCase(BaseSoledadTest):
cert_file=None)
self.assertEqual(
os.path.join(self.tempdir, 'value_3'),
- sol.secrets_path)
+ sol.secrets.secrets_path)
self.assertEqual(
os.path.join(self.tempdir, 'value_2'),
sol.local_db_path)
- self.assertEqual('value_1', sol.server_url)
+ self.assertEqual('value_1', sol._server_url)
sol.close()
def test_change_passphrase(self):
"""
Test if passphrase can be changed.
"""
+ prefix = '_change_passphrase'
sol = self._soledad_instance(
'leap@leap.se',
passphrase=u'123',
- prefix=self.rand_prefix,
+ prefix=prefix,
)
- doc = sol.create_doc({'simple': 'doc'})
- doc_id = doc.doc_id
-
- # change the passphrase
- sol.change_passphrase(u'654321')
- sol.close()
- self.assertRaises(
- WrongMacError,
- self._soledad_instance, 'leap@leap.se',
- passphrase=u'123',
- prefix=self.rand_prefix)
-
- # use new passphrase and retrieve doc
- sol2 = self._soledad_instance(
- 'leap@leap.se',
- passphrase=u'654321',
- prefix=self.rand_prefix)
- doc2 = sol2.get_doc(doc_id)
- self.assertEqual(doc, doc2)
- sol2.close()
+ def _change_passphrase(doc1):
+ self._doc1 = doc1
+ sol.change_passphrase(u'654321')
+ sol.close()
+
+ def _assert_wrong_password_raises(results):
+ self.assertRaises(
+ WrongMacError,
+ self._soledad_instance, 'leap@leap.se',
+ passphrase=u'123',
+ prefix=prefix)
+
+ def _instantiate_with_new_passphrase(results):
+ sol2 = self._soledad_instance(
+ 'leap@leap.se',
+ passphrase=u'654321',
+ prefix=prefix)
+ self._sol2 = sol2
+ return sol2.get_doc(self._doc1.doc_id)
+
+ def _assert_docs_are_equal(doc2):
+ self.assertEqual(self._doc1, doc2)
+ self._sol2.close()
+
+ d = sol.create_doc({'simple': 'doc'})
+ d.addCallback(_change_passphrase)
+ d.addCallback(_assert_wrong_password_raises)
+ d.addCallback(_instantiate_with_new_passphrase)
+ d.addCallback(_assert_docs_are_equal)
+ d.addCallback(lambda _: sol.close())
+
+ return d
def test_change_passphrase_with_short_passphrase_raises(self):
"""
@@ -150,7 +160,7 @@ class AuxMethodsTestCase(BaseSoledadTest):
Assert passphrase getter works fine.
"""
sol = self._soledad_instance()
- self.assertEqual('123', sol.passphrase)
+ self.assertEqual('123', sol._passphrase)
sol.close()
@@ -175,7 +185,7 @@ class SoledadSharedDBTestCase(BaseSoledadTest):
doc_id = self._soledad.secrets._shared_db_doc_id()
self._soledad.secrets._get_secrets_from_shared_db()
self.assertTrue(
- self._soledad._shared_db().get_doc.assert_called_with(
+ self._soledad.shared_db.get_doc.assert_called_with(
doc_id) is None,
'Wrong doc_id when fetching recovery document.')
@@ -186,11 +196,11 @@ class SoledadSharedDBTestCase(BaseSoledadTest):
doc_id = self._soledad.secrets._shared_db_doc_id()
self._soledad.secrets._put_secrets_in_shared_db()
self.assertTrue(
- self._soledad._shared_db().get_doc.assert_called_with(
+ self._soledad.shared_db.get_doc.assert_called_with(
doc_id) is None,
'Wrong doc_id when fetching recovery document.')
self.assertTrue(
- self._soledad._shared_db.put_doc.assert_called_with(
+ self._soledad.shared_db.put_doc.assert_called_with(
self._doc_put) is None,
'Wrong document when putting recovery document.')
self.assertTrue(
@@ -285,8 +295,8 @@ class SoledadSignalingTestCase(BaseSoledadTest):
ADDRESS,
)
# assert db was locked and unlocked
- sol._shared_db.lock.assert_called_with()
- sol._shared_db.unlock.assert_called_with('atoken')
+ sol.shared_db.lock.assert_called_with()
+ sol.shared_db.unlock.assert_called_with('atoken')
sol.close()
def test_stage2_bootstrap_signals(self):
@@ -299,25 +309,15 @@ class SoledadSignalingTestCase(BaseSoledadTest):
# create a document with secrets
doc = SoledadDocument(doc_id=sol.secrets._shared_db_doc_id())
doc.content = sol.secrets._export_recovery_document()
-
- class Stage2MockSharedDB(object):
-
- get_doc = Mock(return_value=doc)
- put_doc = Mock()
- lock = Mock(return_value=('atoken', 300))
- unlock = Mock()
-
- def __call__(self):
- return self
-
sol.close()
# reset mock
soledad.client.secrets.events.signal.reset_mock()
# get a fresh instance so it emits all bootstrap signals
+ shared_db = self.get_default_shared_mock(get_doc_return_value=doc)
sol = self._soledad_instance(
secrets_path='alternative_stage2.json',
local_db_path='alternative_stage2.u1db',
- shared_db_class=Stage2MockSharedDB)
+ shared_db_class=shared_db)
# reverse call order so we can verify in the order the signals were
# expected
soledad.client.secrets.events.signal.mock_calls.reverse()
@@ -355,33 +355,17 @@ class SoledadSignalingTestCase(BaseSoledadTest):
sol = self._soledad_instance()
# mock the actual db sync so soledad does not try to connect to the
# server
- sol._db.sync = Mock()
- # do the sync
- sol.sync()
- # assert the signal has been emitted
- soledad.client.signal.assert_called_with(
- proto.SOLEDAD_DONE_DATA_SYNC,
- ADDRESS,
- )
- sol.close()
-
- def test_need_sync_signals(self):
- """
- Test Soledad emits SOLEDAD_CREATING_KEYS signal.
- """
- soledad.client.signal.reset_mock()
- sol = self._soledad_instance()
- # mock the sync target
- old_get_sync_info = SoledadSyncTarget.get_sync_info
- SoledadSyncTarget.get_sync_info = Mock(return_value=[0, 0, 0, 0, 2])
- # mock our generation so soledad thinks there's new data to sync
- sol._db._get_generation = Mock(return_value=1)
- # check for new data to sync
- sol.need_sync('http://provider/userdb')
- # assert the signal has been emitted
- soledad.client.signal.assert_called_with(
- proto.SOLEDAD_NEW_DATA_TO_SYNC,
- ADDRESS,
- )
- SoledadSyncTarget.get_sync_info = old_get_sync_info
- sol.close()
+ sol._dbsyncer.sync = Mock()
+
+ def _assert_done_data_sync_signal_emitted(results):
+ # assert the signal has been emitted
+ soledad.client.signal.assert_called_with(
+ proto.SOLEDAD_DONE_DATA_SYNC,
+ ADDRESS,
+ )
+ sol.close()
+
+ # do the sync and assert signal was emitted
+ d = sol.sync()
+ d.addCallback(_assert_done_data_sync_signal_emitted)
+ return d
diff --git a/common/src/leap/soledad/common/tests/test_soledad_app.py b/common/src/leap/soledad/common/tests/test_soledad_app.py
new file mode 100644
index 00000000..6efae1d6
--- /dev/null
+++ b/common/src/leap/soledad/common/tests/test_soledad_app.py
@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+# test_soledad_app.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/>.
+
+
+"""
+Test ObjectStore and Couch backend bits.
+"""
+
+
+from testscenarios import TestWithScenarios
+
+from leap.soledad.common.tests.util import BaseSoledadTest
+from leap.soledad.common.tests.util import make_soledad_document_for_test
+from leap.soledad.common.tests.util import make_soledad_app
+from leap.soledad.common.tests.util import make_token_soledad_app
+from leap.soledad.common.tests.util import make_token_http_database_for_test
+from leap.soledad.common.tests.util import copy_token_http_database_for_test
+from leap.soledad.common.tests.u1db_tests import test_backends
+
+
+#-----------------------------------------------------------------------------
+# The following tests come from `u1db.tests.test_backends`.
+#-----------------------------------------------------------------------------
+
+LEAP_SCENARIOS = [
+ ('http', {
+ 'make_database_for_test': test_backends.make_http_database_for_test,
+ 'copy_database_for_test': test_backends.copy_http_database_for_test,
+ 'make_document_for_test': make_soledad_document_for_test,
+ 'make_app_with_state': make_soledad_app}),
+]
+
+
+class SoledadTests(
+ TestWithScenarios, test_backends.AllDatabaseTests, BaseSoledadTest):
+
+ scenarios = LEAP_SCENARIOS + [
+ ('token_http', {'make_database_for_test':
+ make_token_http_database_for_test,
+ 'copy_database_for_test':
+ copy_token_http_database_for_test,
+ 'make_document_for_test': make_soledad_document_for_test,
+ 'make_app_with_state': make_token_soledad_app,
+ })
+ ]
diff --git a/common/src/leap/soledad/common/tests/test_soledad_doc.py b/common/src/leap/soledad/common/tests/test_soledad_doc.py
index 0952de6d..4a67f80a 100644
--- a/common/src/leap/soledad/common/tests/test_soledad_doc.py
+++ b/common/src/leap/soledad/common/tests/test_soledad_doc.py
@@ -17,28 +17,30 @@
"""
Test Leap backend bits: soledad docs
"""
-from leap.soledad.common.tests import BaseSoledadTest
+from testscenarios import TestWithScenarios
+
from leap.soledad.common.tests.u1db_tests import test_document
-from leap.soledad.common.tests import u1db_tests as tests
-from leap.soledad.common.tests import test_sync_target as st
+from leap.soledad.common.tests.util import BaseSoledadTest
+from leap.soledad.common.tests.util import make_soledad_document_for_test
+
#-----------------------------------------------------------------------------
# The following tests come from `u1db.tests.test_document`.
#-----------------------------------------------------------------------------
-
-class TestSoledadDocument(test_document.TestDocument, BaseSoledadTest):
+class TestSoledadDocument(
+ TestWithScenarios,
+ test_document.TestDocument, BaseSoledadTest):
scenarios = ([(
'leap', {
- 'make_document_for_test': st.make_leap_document_for_test})])
+ 'make_document_for_test': make_soledad_document_for_test})])
-class TestSoledadPyDocument(test_document.TestPyDocument, BaseSoledadTest):
+class TestSoledadPyDocument(
+ TestWithScenarios,
+ test_document.TestPyDocument, BaseSoledadTest):
scenarios = ([(
'leap', {
- 'make_document_for_test': st.make_leap_document_for_test})])
-
-
-load_tests = tests.load_with_scenarios
+ 'make_document_for_test': make_soledad_document_for_test})])
diff --git a/common/src/leap/soledad/common/tests/test_sqlcipher.py b/common/src/leap/soledad/common/tests/test_sqlcipher.py
index 78e2f01b..ceb095b8 100644
--- a/common/src/leap/soledad/common/tests/test_sqlcipher.py
+++ b/common/src/leap/soledad/common/tests/test_sqlcipher.py
@@ -20,10 +20,11 @@ Test sqlcipher backend internals.
import os
import time
import threading
-
+import tempfile
+import shutil
from pysqlcipher import dbapi2
-
+from testscenarios import TestWithScenarios
# u1db stuff.
from u1db import (
@@ -34,17 +35,16 @@ from u1db.backends.sqlite_backend import SQLitePartialExpandDatabase
# soledad stuff.
+from leap.soledad.common import soledad_assert
from leap.soledad.common.document import SoledadDocument
from leap.soledad.client.sqlcipher import (
SQLCipherDatabase,
SQLCipherOptions,
DatabaseIsNotEncrypted,
- initialize_sqlcipher_db,
)
# u1db tests stuff.
-from leap.common.testing.basetest import BaseLeapTest
from leap.soledad.common.tests import u1db_tests as tests
from leap.soledad.common.tests.u1db_tests import test_sqlite_backend
from leap.soledad.common.tests.u1db_tests import test_backends
@@ -53,12 +53,12 @@ from leap.soledad.common.tests.util import (
make_sqlcipher_database_for_test,
copy_sqlcipher_database_for_test,
PASSWORD,
+ BaseSoledadTest,
)
def sqlcipher_open(path, passphrase, create=True, document_factory=None):
return SQLCipherDatabase(
- None,
SQLCipherOptions(path, passphrase, create=create))
@@ -93,30 +93,34 @@ SQLCIPHER_SCENARIOS = [
]
-class SQLCipherTests(test_backends.AllDatabaseTests):
+class SQLCipherTests(TestWithScenarios, test_backends.AllDatabaseTests):
scenarios = SQLCIPHER_SCENARIOS
-class SQLCipherDatabaseTests(test_backends.LocalDatabaseTests):
+class SQLCipherDatabaseTests(TestWithScenarios, test_backends.LocalDatabaseTests):
scenarios = SQLCIPHER_SCENARIOS
class SQLCipherValidateGenNTransIdTests(
+ TestWithScenarios,
test_backends.LocalDatabaseValidateGenNTransIdTests):
scenarios = SQLCIPHER_SCENARIOS
class SQLCipherValidateSourceGenTests(
+ TestWithScenarios,
test_backends.LocalDatabaseValidateSourceGenTests):
scenarios = SQLCIPHER_SCENARIOS
class SQLCipherWithConflictsTests(
+ TestWithScenarios,
test_backends.LocalDatabaseWithConflictsTests):
scenarios = SQLCIPHER_SCENARIOS
-class SQLCipherIndexTests(test_backends.DatabaseIndexTests):
+class SQLCipherIndexTests(
+ TestWithScenarios, test_backends.DatabaseIndexTests):
scenarios = SQLCIPHER_SCENARIOS
@@ -124,7 +128,7 @@ class SQLCipherIndexTests(test_backends.DatabaseIndexTests):
# The following tests come from `u1db.tests.test_sqlite_backend`.
#-----------------------------------------------------------------------------
-class TestSQLCipherDatabase(test_sqlite_backend.TestSQLiteDatabase):
+class TestSQLCipherDatabase(TestWithScenarios, test_sqlite_backend.TestSQLiteDatabase):
def test_atomic_initialize(self):
# This test was modified to ensure that db2.close() is called within
@@ -137,11 +141,11 @@ class TestSQLCipherDatabase(test_sqlite_backend.TestSQLiteDatabase):
class SQLCipherDatabaseTesting(SQLCipherDatabase):
_index_storage_value = "testing"
- def __init__(self, soledad_crypto, dbname, ntry):
+ def __init__(self, dbname, ntry):
self._try = ntry
self._is_initialized_invocations = 0
SQLCipherDatabase.__init__(
- self, soledad_crypto,
+ self,
SQLCipherOptions(dbname, PASSWORD))
def _is_initialized(self, c):
@@ -161,15 +165,14 @@ class TestSQLCipherDatabase(test_sqlite_backend.TestSQLiteDatabase):
def run(self):
try:
- db2 = SQLCipherDatabaseTesting(None, dbname, 2)
+ db2 = SQLCipherDatabaseTesting(dbname, 2)
except Exception, e:
SecondTry.outcome2.append(e)
else:
SecondTry.outcome2.append(db2)
- self.close()
t2 = SecondTry()
- db1 = SQLCipherDatabaseTesting(None, dbname, 1)
+ db1 = SQLCipherDatabaseTesting(dbname, 1)
t2.join()
self.assertIsInstance(SecondTry.outcome2[0], SQLCipherDatabaseTesting)
@@ -368,7 +371,7 @@ class SQLCipherOpen(test_open.TestU1DBOpen):
# Tests for actual encryption of the database
#-----------------------------------------------------------------------------
-class SQLCipherEncryptionTest(BaseLeapTest):
+class SQLCipherEncryptionTest(BaseSoledadTest):
"""
Tests to guarantee SQLCipher is indeed encrypting data when storing.
"""
@@ -379,11 +382,37 @@ class SQLCipherEncryptionTest(BaseLeapTest):
os.unlink(dbfile)
def setUp(self):
+ # the following come from BaseLeapTest.setUpClass, because
+ # twisted.trial doesn't support such class methods for setting up
+ # test classes.
+ 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
+ # this is our own stuff
self.DB_FILE = os.path.join(self.tempdir, 'test.db')
self._delete_dbfiles()
def tearDown(self):
self._delete_dbfiles()
+ # the following come from BaseLeapTest.tearDownClass, because
+ # twisted.trial doesn't support such class methods for tearing down
+ # test classes.
+ os.environ["PATH"] = self.old_path
+ os.environ["HOME"] = self.old_home
+ # safety check! please do not wipe my home...
+ # XXX needs to adapt to non-linuces
+ soledad_assert(
+ self.tempdir.startswith('/tmp/leap_tests-') or
+ self.tempdir.startswith('/var/folder'),
+ "beware! tried to remove a dir which does not "
+ "live in temporal folder!")
+ shutil.rmtree(self.tempdir)
def test_try_to_open_encrypted_db_with_sqlite_backend(self):
"""
@@ -426,6 +455,3 @@ class SQLCipherEncryptionTest(BaseLeapTest):
"dbs.")
except DatabaseIsNotEncrypted:
pass
-
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_sqlcipher_sync.py b/common/src/leap/soledad/common/tests/test_sqlcipher_sync.py
index ad2a06b3..83c3449e 100644
--- a/common/src/leap/soledad/common/tests/test_sqlcipher_sync.py
+++ b/common/src/leap/soledad/common/tests/test_sqlcipher_sync.py
@@ -19,12 +19,14 @@ Test sqlcipher backend sync.
"""
+import os
import simplejson as json
from u1db import (
sync,
vectorclock,
)
+from testscenarios import TestWithScenarios
from leap.soledad.common.crypto import ENC_SCHEME_KEY
from leap.soledad.client.target import SoledadSyncTarget
@@ -33,15 +35,12 @@ from leap.soledad.client.sqlcipher import (
SQLCipherDatabase,
)
-
-from leap.soledad.common.tests import u1db_tests as tests, BaseSoledadTest
+from leap.soledad.common.tests import u1db_tests as tests
from leap.soledad.common.tests.u1db_tests import test_sync
-from leap.soledad.common.tests.test_sqlcipher import (
- SQLCIPHER_SCENARIOS,
- make_document_for_test,
-)
+from leap.soledad.common.tests.test_sqlcipher import SQLCIPHER_SCENARIOS
from leap.soledad.common.tests.util import (
make_soledad_app,
+ BaseSoledadTest,
SoledadWithCouchServerMixin,
)
@@ -50,15 +49,7 @@ from leap.soledad.common.tests.util import (
# The following tests come from `u1db.tests.test_sync`.
#-----------------------------------------------------------------------------
-sync_scenarios = []
-for name, scenario in SQLCIPHER_SCENARIOS:
- scenario = dict(scenario)
- scenario['do_sync'] = test_sync.sync_via_synchronizer
- sync_scenarios.append((name, scenario))
- scenario = dict(scenario)
-
-
-def sync_via_synchronizer_and_leap(test, db_source, db_target,
+def sync_via_synchronizer_and_soledad(test, db_source, db_target,
trace_hook=None, trace_hook_shallow=None):
if trace_hook:
test.skipTest("full trace hook unsupported over http")
@@ -71,17 +62,16 @@ def sync_via_synchronizer_and_leap(test, db_source, db_target,
return sync.Synchronizer(db_source, target).sync()
-sync_scenarios.append(('pyleap', {
- 'make_database_for_test': test_sync.make_database_for_http_test,
- 'copy_database_for_test': test_sync.copy_database_for_http_test,
- 'make_document_for_test': make_document_for_test,
- 'make_app_with_state': tests.test_remote_sync_target.make_http_app,
- 'do_sync': test_sync.sync_via_synchronizer,
-}))
+sync_scenarios = []
+for name, scenario in SQLCIPHER_SCENARIOS:
+ scenario['do_sync'] = test_sync.sync_via_synchronizer
+ sync_scenarios.append((name, scenario))
class SQLCipherDatabaseSyncTests(
- test_sync.DatabaseSyncTests, BaseSoledadTest):
+ TestWithScenarios,
+ test_sync.DatabaseSyncTests,
+ BaseSoledadTest):
"""
Test for succesfull sync between SQLCipher and LeapBackend.
@@ -92,8 +82,8 @@ class SQLCipherDatabaseSyncTests(
scenarios = sync_scenarios
- def setUp(self):
- test_sync.DatabaseSyncTests.setUp(self)
+ #def setUp(self):
+ # test_sync.DatabaseSyncTests.setUp(self)
def tearDown(self):
test_sync.DatabaseSyncTests.tearDown(self)
@@ -112,8 +102,6 @@ class SQLCipherDatabaseSyncTests(
and isinstance(self.db3, SQLCipherDatabase):
self.db3.close()
-
-
def test_sync_autoresolves(self):
"""
Test for sync autoresolve remote.
@@ -321,12 +309,14 @@ target_scenarios = [
'create_db_and_target': _make_local_db_and_token_http_target,
# 'make_app_with_state': tests.test_remote_sync_target.make_http_app,
'make_app_with_state': make_soledad_app,
- 'do_sync': test_sync.sync_via_synchronizer}),
+ 'do_sync': sync_via_synchronizer_and_soledad}),
]
class SQLCipherSyncTargetTests(
- SoledadWithCouchServerMixin, test_sync.DatabaseSyncTargetTests):
+ TestWithScenarios,
+ SoledadWithCouchServerMixin,
+ test_sync.DatabaseSyncTargetTests):
scenarios = (tests.multiply_scenarios(SQLCIPHER_SCENARIOS,
target_scenarios))
@@ -406,4 +396,3 @@ class SQLCipherSyncTargetTests(
self.db._last_exchange_log['return'],
{'last_gen': 2, 'docs':
[(doc.doc_id, doc.rev), (doc2.doc_id, doc2.rev)]})
-
diff --git a/common/src/leap/soledad/common/tests/test_sync.py b/common/src/leap/soledad/common/tests/test_sync.py
index 0433fac9..893df56b 100644
--- a/common/src/leap/soledad/common/tests/test_sync.py
+++ b/common/src/leap/soledad/common/tests/test_sync.py
@@ -16,43 +16,35 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
-import mock
-import os
import json
import tempfile
import threading
import time
+
from urlparse import urljoin
+from twisted.internet import defer
+
+from testscenarios import TestWithScenarios
from leap.soledad.common import couch
+from leap.soledad.client import target
+from leap.soledad.client import sync
+from leap.soledad.server import SoledadApp
-from leap.soledad.common.tests import BaseSoledadTest
-from leap.soledad.common.tests import test_sync_target
from leap.soledad.common.tests import u1db_tests as tests
-from leap.soledad.common.tests.u1db_tests import (
- TestCaseWithServer,
- simple_doc,
- test_backends,
- test_sync
-)
-from leap.soledad.common.tests.test_couch import CouchDBTestCase
-from leap.soledad.common.tests.test_target_soledad import (
- make_token_soledad_app,
- make_leap_document_for_test,
-)
-from leap.soledad.common.tests.test_sync_target import token_leap_sync_target
-from leap.soledad.client import (
- Soledad,
- target,
-)
+from leap.soledad.common.tests.u1db_tests import TestCaseWithServer
+from leap.soledad.common.tests.u1db_tests import simple_doc
+from leap.soledad.common.tests.u1db_tests import test_sync
+from leap.soledad.common.tests.util import make_token_soledad_app
+from leap.soledad.common.tests.util import make_soledad_document_for_test
+from leap.soledad.common.tests.util import token_soledad_sync_target
+from leap.soledad.common.tests.util import BaseSoledadTest
from leap.soledad.common.tests.util import SoledadWithCouchServerMixin
-from leap.soledad.client.sync import SoledadSynchronizer
-from leap.soledad.server import SoledadApp
-
+from leap.soledad.common.tests.test_couch import CouchDBTestCase
class InterruptableSyncTestCase(
- CouchDBTestCase, TestCaseWithServer):
+ BaseSoledadTest, CouchDBTestCase, TestCaseWithServer):
"""
Tests for encrypted sync using Soledad server backed by a couch database.
"""
@@ -61,47 +53,9 @@ class InterruptableSyncTestCase(
def make_app_with_state(state):
return make_token_soledad_app(state)
- make_document_for_test = make_leap_document_for_test
-
- sync_target = token_leap_sync_target
+ make_document_for_test = make_soledad_document_for_test
- def _soledad_instance(self, user='user-uuid', passphrase=u'123',
- prefix='',
- secrets_path=Soledad.STORAGE_SECRETS_FILE_NAME,
- local_db_path='soledad.u1db', server_url='',
- cert_file=None, auth_token=None, secret_id=None):
- """
- Instantiate Soledad.
- """
-
- # this callback ensures we save a document which is sent to the shared
- # db.
- def _put_doc_side_effect(doc):
- self._doc_put = doc
-
- # we need a mocked shared db or else Soledad will try to access the
- # network to find if there are uploaded secrets.
- class MockSharedDB(object):
-
- get_doc = mock.Mock(return_value=None)
- put_doc = mock.Mock(side_effect=_put_doc_side_effect)
- lock = mock.Mock(return_value=('atoken', 300))
- unlock = mock.Mock()
-
- def __call__(self):
- return self
-
- Soledad._shared_db = MockSharedDB()
- return Soledad(
- user,
- passphrase,
- secrets_path=os.path.join(self.tempdir, prefix, secrets_path),
- local_db_path=os.path.join(
- self.tempdir, prefix, local_db_path),
- server_url=server_url,
- cert_file=cert_file,
- auth_token=auth_token,
- secret_id=secret_id)
+ sync_target = token_soledad_sync_target
def make_app(self):
self.request_state = couch.CouchServerState(
@@ -135,7 +89,8 @@ class InterruptableSyncTestCase(
def run(self):
while db._get_generation() < 2:
- time.sleep(1)
+ #print "WAITING %d" % db._get_generation()
+ time.sleep(0.1)
self._soledad.stop_sync()
time.sleep(1)
@@ -143,16 +98,7 @@ class InterruptableSyncTestCase(
self.startServer()
# instantiate soledad and create a document
- sol = self._soledad_instance(
- # token is verified in test_target.make_token_soledad_app
- auth_token='auth-token'
- )
- _, doclist = sol.get_all_docs()
- self.assertEqual([], doclist)
-
- # create many small files
- for i in range(0, number_of_docs):
- sol.create_doc(json.loads(simple_doc))
+ sol = self._soledad_instance(user='user-uuid', server_url=self.getURL())
# ensure remote db exists before syncing
db = couch.CouchDatabase.open_database(
@@ -164,21 +110,35 @@ class InterruptableSyncTestCase(
t = _SyncInterruptor(sol, db)
t.start()
- # sync with server
- sol._server_url = self.getURL()
- sol.sync() # this will be interrupted when couch db gen >= 2
- t.join()
+ d = sol.get_all_docs()
+ d.addCallback(lambda results: self.assertEqual([], results[1]))
- # recover the sync process
- sol.sync()
+ def _create_docs(results):
+ # create many small files
+ deferreds = []
+ for i in range(0, number_of_docs):
+ deferreds.append(sol.create_doc(json.loads(simple_doc)))
+ return defer.DeferredList(deferreds)
- gen, doclist = db.get_all_docs()
- self.assertEqual(number_of_docs, len(doclist))
-
- # delete remote database
- db.delete_database()
- db.close()
- sol.close()
+ # sync with server
+ d.addCallback(_create_docs)
+ d.addCallback(lambda _: sol.get_all_docs())
+ d.addCallback(lambda results: self.assertEqual(number_of_docs, len(results[1])))
+ d.addCallback(lambda _: sol.sync())
+ d.addCallback(lambda _: t.join())
+ d.addCallback(lambda _: db.get_all_docs())
+ d.addCallback(lambda results: self.assertNotEqual(number_of_docs, len(results[1])))
+ d.addCallback(lambda _: sol.sync())
+ d.addCallback(lambda _: db.get_all_docs())
+ d.addCallback(lambda results: self.assertEqual(number_of_docs, len(results[1])))
+
+ def _tear_down(results):
+ db.delete_database()
+ db.close()
+ sol.close()
+
+ d.addCallback(_tear_down)
+ return d
def make_soledad_app(state):
@@ -186,6 +146,7 @@ def make_soledad_app(state):
class TestSoledadDbSync(
+ TestWithScenarios,
SoledadWithCouchServerMixin,
test_sync.TestDbSync):
"""
@@ -198,7 +159,7 @@ class TestSoledadDbSync(
'make_database_for_test': tests.make_memory_database_for_test,
}),
('py-token-http', {
- 'make_app_with_state': test_sync_target.make_token_soledad_app,
+ 'make_app_with_state': make_token_soledad_app,
'make_database_for_test': tests.make_memory_database_for_test,
'token': True
}),
@@ -211,10 +172,11 @@ class TestSoledadDbSync(
"""
Need to explicitely invoke inicialization on all bases.
"""
- tests.TestCaseWithServer.setUp(self)
- self.main_test_class = test_sync.TestDbSync
+ #tests.TestCaseWithServer.setUp(self)
+ #self.main_test_class = test_sync.TestDbSync
SoledadWithCouchServerMixin.setUp(self)
self.startServer()
+ self.db = self.make_database_for_test(self, 'test1')
self.db2 = couch.CouchDatabase.open_database(
urljoin(
'http://localhost:' + str(self.wrapper.port), 'test'),
@@ -227,7 +189,7 @@ class TestSoledadDbSync(
"""
self.db2.delete_database()
SoledadWithCouchServerMixin.tearDown(self)
- tests.TestCaseWithServer.tearDown(self)
+ #tests.TestCaseWithServer.tearDown(self)
def do_sync(self, target_name):
"""
@@ -240,7 +202,7 @@ class TestSoledadDbSync(
'token': 'auth-token',
}})
target_url = self.getURL(target_name)
- return SoledadSynchronizer(
+ return sync.SoledadSynchronizer(
self.db,
target.SoledadSyncTarget(
target_url,
@@ -254,8 +216,10 @@ class TestSoledadDbSync(
Adapted to check for encrypted content.
"""
+
doc1 = self.db.create_doc_from_json(tests.simple_doc)
doc2 = self.db2.create_doc_from_json(tests.nested_doc)
+
local_gen_before_sync = self.do_sync('test')
gen, _, changes = self.db.whats_changed(local_gen_before_sync)
self.assertEqual(1, len(changes))
@@ -287,6 +251,3 @@ class TestSoledadDbSync(
s_gen, _ = db3._get_replica_gen_and_trans_id('test1')
self.assertEqual(1, t_gen)
self.assertEqual(1, s_gen)
-
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_sync_deferred.py b/common/src/leap/soledad/common/tests/test_sync_deferred.py
index 07a9742b..26889aff 100644
--- a/common/src/leap/soledad/common/tests/test_sync_deferred.py
+++ b/common/src/leap/soledad/common/tests/test_sync_deferred.py
@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
# test_sync_deferred.py
# Copyright (C) 2014 LEAP
#
@@ -21,28 +20,31 @@ import time
import os
import random
import string
+import shutil
+
from urlparse import urljoin
-from leap.soledad.common.tests import u1db_tests as tests, ADDRESS
+from leap.soledad.common import couch
+from leap.soledad.client.sqlcipher import (
+ SQLCipherOptions,
+ SQLCipherDatabase,
+ SQLCipherU1DBSync,
+)
+
+from testscenarios import TestWithScenarios
+
+from leap.soledad.common.tests import u1db_tests as tests
from leap.soledad.common.tests.u1db_tests import test_sync
+from leap.soledad.common.tests.util import ADDRESS
+from leap.soledad.common.tests.util import SoledadWithCouchServerMixin
+from leap.soledad.common.tests.util import make_soledad_app
-from leap.soledad.common.document import SoledadDocument
-from leap.soledad.common import couch
-from leap.soledad.client import target
-from leap.soledad.client.sync import SoledadSynchronizer
# Just to make clear how this test is different... :)
DEFER_DECRYPTION = True
WAIT_STEP = 1
MAX_WAIT = 10
-
-
-from leap.soledad.client.sqlcipher import open as open_sqlcipher
-from leap.soledad.common.tests.util import SoledadWithCouchServerMixin
-from leap.soledad.common.tests.util import make_soledad_app
-
-
DBPASS = "pass"
@@ -54,8 +56,10 @@ class BaseSoledadDeferredEncTest(SoledadWithCouchServerMixin):
defer_sync_encryption = True
def setUp(self):
+ SoledadWithCouchServerMixin.setUp(self)
# config info
self.db1_file = os.path.join(self.tempdir, "db1.u1db")
+ os.unlink(self.db1_file)
self.db_pass = DBPASS
self.email = ADDRESS
@@ -64,17 +68,21 @@ class BaseSoledadDeferredEncTest(SoledadWithCouchServerMixin):
# each local db.
self.rand_prefix = ''.join(
map(lambda x: random.choice(string.ascii_letters), range(6)))
- # initialize soledad by hand so we can control keys
- self._soledad = self._soledad_instance(
- prefix=self.rand_prefix, user=self.email)
-
- # open test dbs: db1 will be the local sqlcipher db
- # (which instantiates a syncdb)
- self.db1 = open_sqlcipher(self.db1_file, DBPASS, create=True,
- document_factory=SoledadDocument,
- crypto=self._soledad._crypto,
- defer_encryption=True,
- sync_db_key=DBPASS)
+
+ # open test dbs: db1 will be the local sqlcipher db (which
+ # instantiates a syncdb). We use the self._soledad instance that was
+ # already created on some setUp method.
+ import binascii
+ tohex = binascii.b2a_hex
+ key = tohex(self._soledad.secrets.get_local_storage_key())
+ sync_db_key = tohex(self._soledad.secrets.get_sync_db_key())
+ dbpath = self._soledad._local_db_path
+
+ self.opts = SQLCipherOptions(
+ dbpath, key, is_raw_key=True, create=False,
+ defer_encryption=True, sync_db_key=sync_db_key)
+ self.db1 = SQLCipherDatabase(self.opts)
+
self.db2 = couch.CouchDatabase.open_database(
urljoin(
'http://localhost:' + str(self.wrapper.port), 'test'),
@@ -87,20 +95,8 @@ class BaseSoledadDeferredEncTest(SoledadWithCouchServerMixin):
self._soledad.close()
# XXX should not access "private" attrs
- import shutil
shutil.rmtree(os.path.dirname(self._soledad._local_db_path))
-
-
-#SQLCIPHER_SCENARIOS = [
-# ('http', {
-# #'make_app_with_state': test_sync_target.make_token_soledad_app,
-# 'make_app_with_state': make_soledad_app,
-# 'make_database_for_test': ts.make_sqlcipher_database_for_test,
-# 'copy_database_for_test': ts.copy_sqlcipher_database_for_test,
-# 'make_document_for_test': ts.make_document_for_test,
-# 'token': True
-# }),
-#]
+ SoledadWithCouchServerMixin.tearDown(self)
class SyncTimeoutError(Exception):
@@ -111,8 +107,9 @@ class SyncTimeoutError(Exception):
class TestSoledadDbSyncDeferredEncDecr(
- BaseSoledadDeferredEncTest,
- test_sync.TestDbSync):
+ TestWithScenarios,
+ test_sync.TestDbSync,
+ BaseSoledadDeferredEncTest):
"""
Test db.sync remote sync shortcut.
Case with deferred encryption and decryption: using the intermediate
@@ -129,13 +126,17 @@ class TestSoledadDbSyncDeferredEncDecr(
oauth = False
token = True
+ def make_app(self):
+ self.request_state = couch.CouchServerState(
+ self._couch_url, 'shared', 'tokens')
+ return self.make_app_with_state(self.request_state)
+
def setUp(self):
"""
Need to explicitely invoke inicialization on all bases.
"""
- tests.TestCaseWithServer.setUp(self)
- self.main_test_class = test_sync.TestDbSync
BaseSoledadDeferredEncTest.setUp(self)
+ self.server = self.server_thread = None
self.startServer()
self.syncer = None
@@ -143,8 +144,10 @@ class TestSoledadDbSyncDeferredEncDecr(
"""
Need to explicitely invoke destruction on all bases.
"""
+ dbsyncer = getattr(self, 'dbsyncer', None)
+ if dbsyncer:
+ dbsyncer.close()
BaseSoledadDeferredEncTest.tearDown(self)
- tests.TestCaseWithServer.tearDown(self)
def do_sync(self, target_name):
"""
@@ -152,25 +155,20 @@ class TestSoledadDbSyncDeferredEncDecr(
and Token auth.
"""
if self.token:
- extra = dict(creds={'token': {
+ creds={'token': {
'uuid': 'user-uuid',
'token': 'auth-token',
- }})
+ }}
target_url = self.getURL(target_name)
- syncdb = getattr(self.db1, "_sync_db", None)
-
- syncer = SoledadSynchronizer(
- self.db1,
- target.SoledadSyncTarget(
- target_url,
- crypto=self._soledad._crypto,
- sync_db=syncdb,
- **extra))
- # Keep a reference to be able to know when the sync
- # has finished.
- self.syncer = syncer
- return syncer.sync(
- autocreate=True, defer_decryption=DEFER_DECRYPTION)
+
+ # get a u1db syncer
+ crypto = self._soledad._crypto
+ replica_uid = self.db1._replica_uid
+ dbsyncer = SQLCipherU1DBSync(self.opts, crypto, replica_uid,
+ defer_encryption=True)
+ self.dbsyncer = dbsyncer
+ return dbsyncer.sync(target_url, creds=creds,
+ autocreate=True,defer_decryption=DEFER_DECRYPTION)
else:
return test_sync.TestDbSync.do_sync(self, target_name)
@@ -195,28 +193,30 @@ class TestSoledadDbSyncDeferredEncDecr(
"""
doc1 = self.db1.create_doc_from_json(tests.simple_doc)
doc2 = self.db2.create_doc_from_json(tests.nested_doc)
+ d = self.do_sync('test')
- import time
- # need to give time to the encryption to proceed
- # TODO should implement a defer list to subscribe to the all-decrypted
- # event
- time.sleep(2)
+ def _assert_successful_sync(results):
+ import time
+ # need to give time to the encryption to proceed
+ # TODO should implement a defer list to subscribe to the all-decrypted
+ # event
+ time.sleep(2)
+ local_gen_before_sync = results
+ self.wait_for_sync()
- local_gen_before_sync = self.do_sync('test')
- self.wait_for_sync()
+ gen, _, changes = self.db1.whats_changed(local_gen_before_sync)
+ self.assertEqual(1, len(changes))
- gen, _, changes = self.db1.whats_changed(local_gen_before_sync)
- self.assertEqual(1, len(changes))
+ self.assertEqual(doc2.doc_id, changes[0][0])
+ self.assertEqual(1, gen - local_gen_before_sync)
- self.assertEqual(doc2.doc_id, changes[0][0])
- self.assertEqual(1, gen - local_gen_before_sync)
+ self.assertGetEncryptedDoc(
+ self.db2, doc1.doc_id, doc1.rev, tests.simple_doc, False)
+ self.assertGetEncryptedDoc(
+ self.db1, doc2.doc_id, doc2.rev, tests.nested_doc, False)
- self.assertGetEncryptedDoc(
- self.db2, doc1.doc_id, doc1.rev, tests.simple_doc, False)
- self.assertGetEncryptedDoc(
- self.db1, doc2.doc_id, doc2.rev, tests.nested_doc, False)
+ d.addCallback(_assert_successful_sync)
+ return d
def test_db_sync_autocreate(self):
pass
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_sync_target.py b/common/src/leap/soledad/common/tests/test_sync_target.py
index 45009f4e..3792159a 100644
--- a/common/src/leap/soledad/common/tests/test_sync_target.py
+++ b/common/src/leap/soledad/common/tests/test_sync_target.py
@@ -19,82 +19,38 @@ Test Leap backend bits: sync target
"""
import cStringIO
import os
-
+import time
import simplejson as json
import u1db
+import random
+import string
+import shutil
+
+from testscenarios import TestWithScenarios
+from urlparse import urljoin
-from u1db.remote import http_database
+from leap.soledad.client import target
+from leap.soledad.client import crypto
+from leap.soledad.client.sqlcipher import SQLCipherU1DBSync
+from leap.soledad.client.sqlcipher import SQLCipherOptions
+from leap.soledad.client.sqlcipher import SQLCipherDatabase
-from leap.soledad.client import (
- target,
- auth,
- crypto,
- sync,
-)
+from leap.soledad.common import couch
from leap.soledad.common.document import SoledadDocument
from leap.soledad.common.tests import u1db_tests as tests
-from leap.soledad.common.tests import BaseSoledadTest
-from leap.soledad.common.tests.util import (
- make_sqlcipher_database_for_test,
- make_soledad_app,
- make_token_soledad_app,
- SoledadWithCouchServerMixin,
-)
-from leap.soledad.common.tests.u1db_tests import test_backends
+from leap.soledad.common.tests.util import make_sqlcipher_database_for_test
+from leap.soledad.common.tests.util import make_soledad_app
+from leap.soledad.common.tests.util import make_token_soledad_app
+from leap.soledad.common.tests.util import make_soledad_document_for_test
+from leap.soledad.common.tests.util import token_soledad_sync_target
+from leap.soledad.common.tests.util import BaseSoledadTest
+from leap.soledad.common.tests.util import SoledadWithCouchServerMixin
+from leap.soledad.common.tests.util import ADDRESS
from leap.soledad.common.tests.u1db_tests import test_remote_sync_target
from leap.soledad.common.tests.u1db_tests import test_sync
-#-----------------------------------------------------------------------------
-# The following tests come from `u1db.tests.test_backends`.
-#-----------------------------------------------------------------------------
-
-def make_leap_document_for_test(test, doc_id, rev, content,
- has_conflicts=False):
- return SoledadDocument(
- doc_id, rev, content, has_conflicts=has_conflicts)
-
-
-LEAP_SCENARIOS = [
- ('http', {
- 'make_database_for_test': test_backends.make_http_database_for_test,
- 'copy_database_for_test': test_backends.copy_http_database_for_test,
- 'make_document_for_test': make_leap_document_for_test,
- 'make_app_with_state': make_soledad_app}),
-]
-
-
-def make_token_http_database_for_test(test, replica_uid):
- test.startServer()
- test.request_state._create_database(replica_uid)
-
- class _HTTPDatabaseWithToken(
- http_database.HTTPDatabase, auth.TokenBasedAuth):
-
- def set_token_credentials(self, uuid, token):
- auth.TokenBasedAuth.set_token_credentials(self, uuid, token)
-
- def _sign_request(self, method, url_query, params):
- return auth.TokenBasedAuth._sign_request(
- self, method, url_query, params)
-
- http_db = _HTTPDatabaseWithToken(test.getURL('test'))
- http_db.set_token_credentials('user-uuid', 'auth-token')
- return http_db
-
-
-def copy_token_http_database_for_test(test, db):
- # DO NOT COPY OR REUSE THIS CODE OUTSIDE TESTS: COPYING U1DB DATABASES IS
- # THE WRONG THING TO DO, THE ONLY REASON WE DO SO HERE IS TO TEST THAT WE
- # CORRECTLY DETECT IT HAPPENING SO THAT WE CAN RAISE ERRORS RATHER THAN
- # CORRUPT USER DATA. USE SYNC INSTEAD, OR WE WILL SEND NINJA TO YOUR
- # HOUSE.
- http_db = test.request_state._copy_database(db)
- http_db.set_token_credentials(http_db, 'user-uuid', 'auth-token')
- return http_db
-
-
#-----------------------------------------------------------------------------
# The following tests come from `u1db.tests.test_remote_sync_target`.
#-----------------------------------------------------------------------------
@@ -122,12 +78,6 @@ class TestSoledadParsingSyncStream(
target.
"""
- def setUp(self):
- test_remote_sync_target.TestParsingSyncStream.setUp(self)
-
- def tearDown(self):
- test_remote_sync_target.TestParsingSyncStream.tearDown(self)
-
def test_extra_comma(self):
"""
Test adapted to use encrypted content.
@@ -209,17 +159,6 @@ class TestSoledadParsingSyncStream(
# functions for TestRemoteSyncTargets
#
-def leap_sync_target(test, path):
- return target.SoledadSyncTarget(
- test.getURL(path), crypto=test._soledad._crypto)
-
-
-def token_leap_sync_target(test, path):
- st = leap_sync_target(test, path)
- st.set_token_credentials('user-uuid', 'auth-token')
- return st
-
-
def make_local_db_and_soledad_target(test, path='test'):
test.startServer()
db = test.request_state._create_database(os.path.basename(path))
@@ -235,32 +174,32 @@ def make_local_db_and_token_soledad_target(test):
class TestSoledadSyncTarget(
+ TestWithScenarios,
SoledadWithCouchServerMixin,
test_remote_sync_target.TestRemoteSyncTargets):
scenarios = [
('token_soledad',
{'make_app_with_state': make_token_soledad_app,
- 'make_document_for_test': make_leap_document_for_test,
+ 'make_document_for_test': make_soledad_document_for_test,
'create_db_and_target': make_local_db_and_token_soledad_target,
'make_database_for_test': make_sqlcipher_database_for_test,
- 'sync_target': token_leap_sync_target}),
+ 'sync_target': token_soledad_sync_target}),
]
def setUp(self):
- tests.TestCaseWithServer.setUp(self)
- self.main_test_class = test_remote_sync_target.TestRemoteSyncTargets
+ TestWithScenarios.setUp(self)
SoledadWithCouchServerMixin.setUp(self)
self.startServer()
self.db1 = make_sqlcipher_database_for_test(self, 'test1')
self.db2 = self.request_state._create_database('test2')
def tearDown(self):
- SoledadWithCouchServerMixin.tearDown(self)
- tests.TestCaseWithServer.tearDown(self)
- db2, _ = self.request_state.ensure_database('test2')
- db2.delete_database()
+ #db2, _ = self.request_state.ensure_database('test2')
+ self.db2.delete_database()
self.db1.close()
+ SoledadWithCouchServerMixin.tearDown(self)
+ TestWithScenarios.tearDown(self)
def test_sync_exchange_send(self):
"""
@@ -268,7 +207,6 @@ class TestSoledadSyncTarget(
This test was adapted to decrypt remote content before assert.
"""
- self.startServer()
db = self.request_state._create_database('test')
remote_target = self.getSyncTarget('test')
other_docs = []
@@ -289,14 +227,9 @@ class TestSoledadSyncTarget(
"""
Test for sync exchange failure and retry.
- This test was adapted to:
- - decrypt remote content before assert.
- - not expect a bounced document because soledad has stateful
- recoverable sync.
+ This test was adapted to decrypt remote content before assert.
"""
- self.startServer()
-
def blackhole_getstderr(inst):
return cStringIO.StringIO()
@@ -332,8 +265,9 @@ class TestSoledadSyncTarget(
doc2 = self.make_document('doc-here2', 'replica:1',
'{"value": "here2"}')
- # we do not expect an HTTPError because soledad sync fails gracefully
- remote_target.sync_exchange(
+ self.assertRaises(
+ u1db.errors.HTTPError,
+ remote_target.sync_exchange,
[(doc1, 10, 'T-sid'), (doc2, 11, 'T-sud')],
'replica', last_known_generation=0, last_known_trans_id=None,
return_doc_cb=receive_doc)
@@ -364,7 +298,6 @@ class TestSoledadSyncTarget(
This test was adapted to decrypt remote content before assert.
"""
- self.startServer()
remote_target = self.getSyncTarget('test')
other_docs = []
replica_uid_box = []
@@ -405,7 +338,9 @@ target_scenarios = [
class SoledadDatabaseSyncTargetTests(
- SoledadWithCouchServerMixin, test_sync.DatabaseSyncTargetTests):
+ TestWithScenarios,
+ SoledadWithCouchServerMixin,
+ test_sync.DatabaseSyncTargetTests):
scenarios = (
tests.multiply_scenarios(
@@ -500,8 +435,25 @@ class SoledadDatabaseSyncTargetTests(
[(doc.doc_id, doc.rev), (doc2.doc_id, doc2.rev)]})
+# Just to make clear how this test is different... :)
+DEFER_DECRYPTION = False
+
+WAIT_STEP = 1
+MAX_WAIT = 10
+DBPASS = "pass"
+
+
+class SyncTimeoutError(Exception):
+ """
+ Dummy exception to notify timeout during sync.
+ """
+ pass
+
+
class TestSoledadDbSync(
- SoledadWithCouchServerMixin, test_sync.TestDbSync):
+ TestWithScenarios,
+ SoledadWithCouchServerMixin,
+ test_sync.TestDbSync):
"""Test db.sync remote sync shortcut"""
scenarios = [
@@ -516,13 +468,67 @@ class TestSoledadDbSync(
oauth = False
token = False
+
+ def make_app(self):
+ self.request_state = couch.CouchServerState(
+ self._couch_url, 'shared', 'tokens')
+ return self.make_app_with_state(self.request_state)
+
def setUp(self):
- self.main_test_class = test_sync.TestDbSync
+ """
+ Need to explicitely invoke inicialization on all bases.
+ """
SoledadWithCouchServerMixin.setUp(self)
+ self.server = self.server_thread = None
+ self.startServer()
+ self.syncer = None
+
+ # config info
+ self.db1_file = os.path.join(self.tempdir, "db1.u1db")
+ os.unlink(self.db1_file)
+ self.db_pass = DBPASS
+ self.email = ADDRESS
+
+ # get a random prefix for each test, so we do not mess with
+ # concurrency during initialization and shutting down of
+ # each local db.
+ self.rand_prefix = ''.join(
+ map(lambda x: random.choice(string.ascii_letters), range(6)))
+
+ # open test dbs: db1 will be the local sqlcipher db (which
+ # instantiates a syncdb). We use the self._soledad instance that was
+ # already created on some setUp method.
+ import binascii
+ tohex = binascii.b2a_hex
+ key = tohex(self._soledad.secrets.get_local_storage_key())
+ sync_db_key = tohex(self._soledad.secrets.get_sync_db_key())
+ dbpath = self._soledad._local_db_path
+
+ self.opts = SQLCipherOptions(
+ dbpath, key, is_raw_key=True, create=False,
+ defer_encryption=True, sync_db_key=sync_db_key)
+ self.db1 = SQLCipherDatabase(self.opts)
+
+ self.db2 = couch.CouchDatabase.open_database(
+ urljoin(
+ 'http://localhost:' + str(self.wrapper.port), 'test'),
+ create=True,
+ ensure_ddocs=True)
def tearDown(self):
+ """
+ Need to explicitely invoke destruction on all bases.
+ """
+ dbsyncer = getattr(self, 'dbsyncer', None)
+ if dbsyncer:
+ dbsyncer.close()
+ self.db1.close()
+ self.db2.close()
+ self._soledad.close()
+
+ # XXX should not access "private" attrs
+ shutil.rmtree(os.path.dirname(self._soledad._local_db_path))
SoledadWithCouchServerMixin.tearDown(self)
- self.db.close()
def do_sync(self, target_name):
"""
@@ -530,44 +536,71 @@ class TestSoledadDbSync(
and Token auth.
"""
if self.token:
- extra = dict(creds={'token': {
+ creds={'token': {
'uuid': 'user-uuid',
'token': 'auth-token',
- }})
+ }}
target_url = self.getURL(target_name)
- return sync.SoledadSynchronizer(
- self.db,
- target.SoledadSyncTarget(
- target_url,
- crypto=self._soledad._crypto,
- **extra)).sync(autocreate=True,
- defer_decryption=False)
+
+ # get a u1db syncer
+ crypto = self._soledad._crypto
+ replica_uid = self.db1._replica_uid
+ dbsyncer = SQLCipherU1DBSync(self.opts, crypto, replica_uid,
+ defer_encryption=True)
+ self.dbsyncer = dbsyncer
+ return dbsyncer.sync(target_url, creds=creds,
+ autocreate=True,defer_decryption=DEFER_DECRYPTION)
else:
return test_sync.TestDbSync.do_sync(self, target_name)
+ def wait_for_sync(self):
+ """
+ Wait for sync to finish.
+ """
+ wait = 0
+ syncer = self.syncer
+ if syncer is not None:
+ while syncer.syncing:
+ time.sleep(WAIT_STEP)
+ wait += WAIT_STEP
+ if wait >= MAX_WAIT:
+ raise SyncTimeoutError
+
def test_db_sync(self):
"""
Test sync.
Adapted to check for encrypted content.
"""
- doc1 = self.db.create_doc_from_json(tests.simple_doc)
+ doc1 = self.db1.create_doc_from_json(tests.simple_doc)
doc2 = self.db2.create_doc_from_json(tests.nested_doc)
- local_gen_before_sync = self.do_sync('test2')
- gen, _, changes = self.db.whats_changed(local_gen_before_sync)
- self.assertEqual(1, len(changes))
- self.assertEqual(doc2.doc_id, changes[0][0])
- self.assertEqual(1, gen - local_gen_before_sync)
- self.assertGetEncryptedDoc(
- self.db2, doc1.doc_id, doc1.rev, tests.simple_doc, False)
- self.assertGetEncryptedDoc(
- self.db, doc2.doc_id, doc2.rev, tests.nested_doc, False)
+ d = self.do_sync('test')
+
+ def _assert_successful_sync(results):
+ import time
+ # need to give time to the encryption to proceed
+ # TODO should implement a defer list to subscribe to the all-decrypted
+ # event
+ time.sleep(2)
+ local_gen_before_sync = results
+ self.wait_for_sync()
+
+ gen, _, changes = self.db1.whats_changed(local_gen_before_sync)
+ self.assertEqual(1, len(changes))
+
+ self.assertEqual(doc2.doc_id, changes[0][0])
+ self.assertEqual(1, gen - local_gen_before_sync)
+
+ self.assertGetEncryptedDoc(
+ self.db2, doc1.doc_id, doc1.rev, tests.simple_doc, False)
+ self.assertGetEncryptedDoc(
+ self.db1, doc2.doc_id, doc2.rev, tests.nested_doc, False)
+
+ d.addCallback(_assert_successful_sync)
+ return d
def test_db_sync_autocreate(self):
"""
We bypass this test because we never need to autocreate databases.
"""
pass
-
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_target.py b/common/src/leap/soledad/common/tests/test_target.py
deleted file mode 100644
index eb5e2874..00000000
--- a/common/src/leap/soledad/common/tests/test_target.py
+++ /dev/null
@@ -1,797 +0,0 @@
-# -*- coding: utf-8 -*-
-# test_target.py
-# Copyright (C) 2013 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/>.
-
-
-"""
-Test Leap backend bits.
-"""
-
-import u1db
-import os
-import simplejson as json
-import cStringIO
-
-
-from u1db.sync import Synchronizer
-from u1db.remote import (
- http_client,
- http_database,
-)
-
-from leap.soledad import client
-from leap.soledad.client import (
- target,
- auth,
- VerifiedHTTPSConnection,
-)
-from leap.soledad.common.document import SoledadDocument
-from leap.soledad.server.auth import SoledadTokenAuthMiddleware
-
-
-from leap.soledad.common.tests import u1db_tests as tests
-from leap.soledad.common.tests import BaseSoledadTest
-from leap.soledad.common.tests.util import (
- make_sqlcipher_database_for_test,
- make_soledad_app,
- make_token_soledad_app,
- SoledadWithCouchServerMixin,
-)
-from leap.soledad.common.tests.u1db_tests import test_backends
-from leap.soledad.common.tests.u1db_tests import test_http_database
-from leap.soledad.common.tests.u1db_tests import test_http_client
-from leap.soledad.common.tests.u1db_tests import test_document
-from leap.soledad.common.tests.u1db_tests import test_remote_sync_target
-from leap.soledad.common.tests.u1db_tests import test_https
-from leap.soledad.common.tests.u1db_tests import test_sync
-
-
-#-----------------------------------------------------------------------------
-# The following tests come from `u1db.tests.test_backends`.
-#-----------------------------------------------------------------------------
-
-def make_leap_document_for_test(test, doc_id, rev, content,
- has_conflicts=False):
- return SoledadDocument(
- doc_id, rev, content, has_conflicts=has_conflicts)
-
-
-LEAP_SCENARIOS = [
- ('http', {
- 'make_database_for_test': test_backends.make_http_database_for_test,
- 'copy_database_for_test': test_backends.copy_http_database_for_test,
- 'make_document_for_test': make_leap_document_for_test,
- 'make_app_with_state': make_soledad_app}),
-]
-
-
-def make_token_http_database_for_test(test, replica_uid):
- test.startServer()
- test.request_state._create_database(replica_uid)
-
- class _HTTPDatabaseWithToken(
- http_database.HTTPDatabase, auth.TokenBasedAuth):
-
- def set_token_credentials(self, uuid, token):
- auth.TokenBasedAuth.set_token_credentials(self, uuid, token)
-
- def _sign_request(self, method, url_query, params):
- return auth.TokenBasedAuth._sign_request(
- self, method, url_query, params)
-
- http_db = _HTTPDatabaseWithToken(test.getURL('test'))
- http_db.set_token_credentials('user-uuid', 'auth-token')
- return http_db
-
-
-def copy_token_http_database_for_test(test, db):
- # DO NOT COPY OR REUSE THIS CODE OUTSIDE TESTS: COPYING U1DB DATABASES IS
- # THE WRONG THING TO DO, THE ONLY REASON WE DO SO HERE IS TO TEST THAT WE
- # CORRECTLY DETECT IT HAPPENING SO THAT WE CAN RAISE ERRORS RATHER THAN
- # CORRUPT USER DATA. USE SYNC INSTEAD, OR WE WILL SEND NINJA TO YOUR
- # HOUSE.
- http_db = test.request_state._copy_database(db)
- http_db.set_token_credentials(http_db, 'user-uuid', 'auth-token')
- return http_db
-
-
-class SoledadTests(test_backends.AllDatabaseTests, BaseSoledadTest):
-
- scenarios = LEAP_SCENARIOS + [
- ('token_http', {'make_database_for_test':
- make_token_http_database_for_test,
- 'copy_database_for_test':
- copy_token_http_database_for_test,
- 'make_document_for_test': make_leap_document_for_test,
- 'make_app_with_state': make_token_soledad_app,
- })
- ]
-
-
-#-----------------------------------------------------------------------------
-# The following tests come from `u1db.tests.test_http_client`.
-#-----------------------------------------------------------------------------
-
-class TestSoledadClientBase(test_http_client.TestHTTPClientBase):
- """
- This class should be used to test Token auth.
- """
-
- def getClientWithToken(self, **kwds):
- self.startServer()
-
- class _HTTPClientWithToken(
- http_client.HTTPClientBase, auth.TokenBasedAuth):
-
- def set_token_credentials(self, uuid, token):
- auth.TokenBasedAuth.set_token_credentials(self, uuid, token)
-
- def _sign_request(self, method, url_query, params):
- return auth.TokenBasedAuth._sign_request(
- self, method, url_query, params)
-
- return _HTTPClientWithToken(self.getURL('dbase'), **kwds)
-
- def test_oauth(self):
- """
- Suppress oauth test (we test for token auth here).
- """
- pass
-
- def test_oauth_ctr_creds(self):
- """
- Suppress oauth test (we test for token auth here).
- """
- pass
-
- def test_oauth_Unauthorized(self):
- """
- Suppress oauth test (we test for token auth here).
- """
- pass
-
- def app(self, environ, start_response):
- res = test_http_client.TestHTTPClientBase.app(
- self, environ, start_response)
- if res is not None:
- return res
- # mime solead application here.
- if '/token' in environ['PATH_INFO']:
- auth = environ.get(SoledadTokenAuthMiddleware.HTTP_AUTH_KEY)
- if not auth:
- start_response("401 Unauthorized",
- [('Content-Type', 'application/json')])
- return [json.dumps({"error": "unauthorized",
- "message": e.message})]
- scheme, encoded = auth.split(None, 1)
- if scheme.lower() != 'token':
- start_response("401 Unauthorized",
- [('Content-Type', 'application/json')])
- return [json.dumps({"error": "unauthorized",
- "message": e.message})]
- uuid, token = encoded.decode('base64').split(':', 1)
- if uuid != 'user-uuid' and token != 'auth-token':
- return unauth_err("Incorrect address or token.")
- start_response("200 OK", [('Content-Type', 'application/json')])
- return [json.dumps([environ['PATH_INFO'], uuid, token])]
-
- def test_token(self):
- """
- Test if token is sent correctly.
- """
- cli = self.getClientWithToken()
- cli.set_token_credentials('user-uuid', 'auth-token')
- res, headers = cli._request('GET', ['doc', 'token'])
- self.assertEqual(
- ['/dbase/doc/token', 'user-uuid', 'auth-token'], json.loads(res))
-
- def test_token_ctr_creds(self):
- cli = self.getClientWithToken(creds={'token': {
- 'uuid': 'user-uuid',
- 'token': 'auth-token',
- }})
- res, headers = cli._request('GET', ['doc', 'token'])
- self.assertEqual(
- ['/dbase/doc/token', 'user-uuid', 'auth-token'], json.loads(res))
-
-
-#-----------------------------------------------------------------------------
-# The following tests come from `u1db.tests.test_document`.
-#-----------------------------------------------------------------------------
-
-class TestSoledadDocument(test_document.TestDocument, BaseSoledadTest):
-
- scenarios = ([(
- 'leap', {'make_document_for_test': make_leap_document_for_test})])
-
-
-class TestSoledadPyDocument(test_document.TestPyDocument, BaseSoledadTest):
-
- scenarios = ([(
- 'leap', {'make_document_for_test': make_leap_document_for_test})])
-
-
-#-----------------------------------------------------------------------------
-# The following tests come from `u1db.tests.test_remote_sync_target`.
-#-----------------------------------------------------------------------------
-
-class TestSoledadSyncTargetBasics(
- test_remote_sync_target.TestHTTPSyncTargetBasics):
- """
- Some tests had to be copied to this class so we can instantiate our own
- target.
- """
-
- def test_parse_url(self):
- remote_target = target.SoledadSyncTarget('http://127.0.0.1:12345/')
- self.assertEqual('http', remote_target._url.scheme)
- self.assertEqual('127.0.0.1', remote_target._url.hostname)
- self.assertEqual(12345, remote_target._url.port)
- self.assertEqual('/', remote_target._url.path)
-
-
-class TestSoledadParsingSyncStream(
- test_remote_sync_target.TestParsingSyncStream,
- BaseSoledadTest):
- """
- Some tests had to be copied to this class so we can instantiate our own
- target.
- """
-
- def setUp(self):
- test_remote_sync_target.TestParsingSyncStream.setUp(self)
-
- def tearDown(self):
- test_remote_sync_target.TestParsingSyncStream.tearDown(self)
-
- def test_extra_comma(self):
- """
- Test adapted to use encrypted content.
- """
- doc = SoledadDocument('i', rev='r')
- doc.content = {}
- enc_json = target.encrypt_doc(self._soledad._crypto, doc)
- tgt = target.SoledadSyncTarget(
- "http://foo/foo", crypto=self._soledad._crypto)
-
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream, "[\r\n{},\r\n]", None)
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream,
- '[\r\n{},\r\n{"id": "i", "rev": "r", '
- '"content": %s, "gen": 3, "trans_id": "T-sid"}'
- ',\r\n]' % json.dumps(enc_json),
- lambda doc, gen, trans_id: None)
-
- def test_wrong_start(self):
- tgt = target.SoledadSyncTarget("http://foo/foo")
-
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream, "{}\r\n]", None)
-
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream, "\r\n{}\r\n]", None)
-
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream, "", None)
-
- def test_wrong_end(self):
- tgt = target.SoledadSyncTarget("http://foo/foo")
-
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream, "[\r\n{}", None)
-
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream, "[\r\n", None)
-
- def test_missing_comma(self):
- tgt = target.SoledadSyncTarget("http://foo/foo")
-
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream,
- '[\r\n{}\r\n{"id": "i", "rev": "r", '
- '"content": "c", "gen": 3}\r\n]', None)
-
- def test_no_entries(self):
- tgt = target.SoledadSyncTarget("http://foo/foo")
-
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream, "[\r\n]", None)
-
- def test_error_in_stream(self):
- tgt = target.SoledadSyncTarget("http://foo/foo")
-
- self.assertRaises(u1db.errors.Unavailable,
- tgt._parse_sync_stream,
- '[\r\n{"new_generation": 0},'
- '\r\n{"error": "unavailable"}\r\n', None)
-
- self.assertRaises(u1db.errors.Unavailable,
- tgt._parse_sync_stream,
- '[\r\n{"error": "unavailable"}\r\n', None)
-
- self.assertRaises(u1db.errors.BrokenSyncStream,
- tgt._parse_sync_stream,
- '[\r\n{"error": "?"}\r\n', None)
-
-
-#
-# functions for TestRemoteSyncTargets
-#
-
-def leap_sync_target(test, path):
- return target.SoledadSyncTarget(
- test.getURL(path), crypto=test._soledad._crypto)
-
-
-def token_leap_sync_target(test, path):
- st = leap_sync_target(test, path)
- st.set_token_credentials('user-uuid', 'auth-token')
- return st
-
-
-def make_local_db_and_soledad_target(test, path='test'):
- test.startServer()
- db = test.request_state._create_database(os.path.basename(path))
- st = target.SoledadSyncTarget.connect(
- test.getURL(path), crypto=test._soledad._crypto)
- return db, st
-
-
-def make_local_db_and_token_soledad_target(test):
- db, st = make_local_db_and_soledad_target(test, 'test')
- st.set_token_credentials('user-uuid', 'auth-token')
- return db, st
-
-
-class TestSoledadSyncTarget(
- SoledadWithCouchServerMixin,
- test_remote_sync_target.TestRemoteSyncTargets):
-
- scenarios = [
- ('token_soledad',
- {'make_app_with_state': make_token_soledad_app,
- 'make_document_for_test': make_leap_document_for_test,
- 'create_db_and_target': make_local_db_and_token_soledad_target,
- 'make_database_for_test': make_sqlcipher_database_for_test,
- 'sync_target': token_leap_sync_target}),
- ]
-
- def setUp(self):
- tests.TestCaseWithServer.setUp(self)
- self.main_test_class = test_remote_sync_target.TestRemoteSyncTargets
- SoledadWithCouchServerMixin.setUp(self)
- self.startServer()
- self.db1 = make_sqlcipher_database_for_test(self, 'test1')
- self.db2 = self.request_state._create_database('test2')
-
- def tearDown(self):
- SoledadWithCouchServerMixin.tearDown(self)
- tests.TestCaseWithServer.tearDown(self)
- db, _ = self.request_state.ensure_database('test2')
- db.delete_database()
- for i in ['db1', 'db2']:
- if hasattr(self, i):
- db = getattr(self, i)
- db.close()
-
- def test_sync_exchange_send(self):
- """
- Test for sync exchanging send of document.
-
- This test was adapted to decrypt remote content before assert.
- """
- self.startServer()
- db = self.request_state._create_database('test')
- remote_target = self.getSyncTarget('test')
- other_docs = []
-
- def receive_doc(doc, gen, trans_id):
- other_docs.append((doc.doc_id, doc.rev, doc.get_json()))
-
- doc = self.make_document('doc-here', 'replica:1', '{"value": "here"}')
- new_gen, trans_id = remote_target.sync_exchange(
- [(doc, 10, 'T-sid')], 'replica', last_known_generation=0,
- last_known_trans_id=None, return_doc_cb=receive_doc)
- self.assertEqual(1, new_gen)
- self.assertGetEncryptedDoc(
- db, 'doc-here', 'replica:1', '{"value": "here"}', False)
- db.close()
-
- def test_sync_exchange_send_failure_and_retry_scenario(self):
- """
- Test for sync exchange failure and retry.
-
- This test was adapted to:
- - decrypt remote content before assert.
- - not expect a bounced document because soledad has stateful
- recoverable sync.
- """
-
- self.startServer()
-
- def blackhole_getstderr(inst):
- return cStringIO.StringIO()
-
- self.patch(self.server.RequestHandlerClass, 'get_stderr',
- blackhole_getstderr)
- db = self.request_state._create_database('test')
- _put_doc_if_newer = db._put_doc_if_newer
- trigger_ids = ['doc-here2']
-
- def bomb_put_doc_if_newer(self, doc, save_conflict,
- replica_uid=None, replica_gen=None,
- replica_trans_id=None, number_of_docs=None,
- doc_idx=None, sync_id=None):
- if doc.doc_id in trigger_ids:
- raise Exception
- return _put_doc_if_newer(doc, save_conflict=save_conflict,
- replica_uid=replica_uid,
- replica_gen=replica_gen,
- replica_trans_id=replica_trans_id,
- number_of_docs=number_of_docs,
- doc_idx=doc_idx,
- sync_id=sync_id)
- from leap.soledad.common.tests.test_couch import IndexedCouchDatabase
- self.patch(
- IndexedCouchDatabase, '_put_doc_if_newer', bomb_put_doc_if_newer)
- remote_target = self.getSyncTarget('test')
- other_changes = []
-
- def receive_doc(doc, gen, trans_id):
- other_changes.append(
- (doc.doc_id, doc.rev, doc.get_json(), gen, trans_id))
-
- doc1 = self.make_document('doc-here', 'replica:1', '{"value": "here"}')
- doc2 = self.make_document('doc-here2', 'replica:1',
- '{"value": "here2"}')
- # We do not expect an exception here because the sync fails gracefully
- remote_target.sync_exchange(
- [(doc1, 10, 'T-sid'), (doc2, 11, 'T-sud')],
- 'replica', last_known_generation=0, last_known_trans_id=None,
- return_doc_cb=receive_doc)
- self.assertGetEncryptedDoc(
- db, 'doc-here', 'replica:1', '{"value": "here"}',
- False)
- self.assertEqual(
- (10, 'T-sid'), db._get_replica_gen_and_trans_id('replica'))
- self.assertEqual([], other_changes)
- # retry
- trigger_ids = []
- new_gen, trans_id = remote_target.sync_exchange(
- [(doc2, 11, 'T-sud')], 'replica', last_known_generation=0,
- last_known_trans_id=None, return_doc_cb=receive_doc)
- self.assertGetEncryptedDoc(
- db, 'doc-here2', 'replica:1', '{"value": "here2"}',
- False)
- self.assertEqual(
- (11, 'T-sud'), db._get_replica_gen_and_trans_id('replica'))
- self.assertEqual(2, new_gen)
- self.assertEqual(
- ('doc-here', 'replica:1', '{"value": "here"}', 1),
- other_changes[0][:-1])
- db.close()
-
- def test_sync_exchange_send_ensure_callback(self):
- """
- Test for sync exchange failure and retry.
-
- This test was adapted to decrypt remote content before assert.
- """
- self.startServer()
- remote_target = self.getSyncTarget('test')
- other_docs = []
- replica_uid_box = []
-
- def receive_doc(doc, gen, trans_id):
- other_docs.append((doc.doc_id, doc.rev, doc.get_json()))
-
- def ensure_cb(replica_uid):
- replica_uid_box.append(replica_uid)
-
- doc = self.make_document('doc-here', 'replica:1', '{"value": "here"}')
- new_gen, trans_id = remote_target.sync_exchange(
- [(doc, 10, 'T-sid')], 'replica', last_known_generation=0,
- last_known_trans_id=None, return_doc_cb=receive_doc,
- ensure_callback=ensure_cb)
- self.assertEqual(1, new_gen)
- db = self.request_state.open_database('test')
- self.assertEqual(1, len(replica_uid_box))
- self.assertEqual(db._replica_uid, replica_uid_box[0])
- self.assertGetEncryptedDoc(
- db, 'doc-here', 'replica:1', '{"value": "here"}', False)
- db.close()
-
- def test_sync_exchange_in_stream_error(self):
- # we bypass this test because our sync_exchange process does not
- # return u1db error 503 "unavailable" for now.
- pass
-
-
-#-----------------------------------------------------------------------------
-# The following tests come from `u1db.tests.test_https`.
-#-----------------------------------------------------------------------------
-
-def token_leap_https_sync_target(test, host, path):
- _, port = test.server.server_address
- st = target.SoledadSyncTarget(
- 'https://%s:%d/%s' % (host, port, path),
- crypto=test._soledad._crypto)
- st.set_token_credentials('user-uuid', 'auth-token')
- return st
-
-
-class TestSoledadSyncTargetHttpsSupport(
- test_https.TestHttpSyncTargetHttpsSupport,
- BaseSoledadTest):
-
- scenarios = [
- ('token_soledad_https',
- {'server_def': test_https.https_server_def,
- 'make_app_with_state': make_token_soledad_app,
- 'make_document_for_test': make_leap_document_for_test,
- 'sync_target': token_leap_https_sync_target}),
- ]
-
- def setUp(self):
- # the parent constructor undoes our SSL monkey patch to ensure tests
- # run smoothly with standard u1db.
- test_https.TestHttpSyncTargetHttpsSupport.setUp(self)
- # so here monkey patch again to test our functionality.
- http_client._VerifiedHTTPSConnection = VerifiedHTTPSConnection
- client.SOLEDAD_CERT = http_client.CA_CERTS
-
- def test_working(self):
- """
- Test that SSL connections work well.
-
- This test was adapted to patch Soledad's HTTPS connection custom class
- with the intended CA certificates.
- """
- self.startServer()
- db = self.request_state._create_database('test')
- self.patch(client, 'SOLEDAD_CERT', self.cacert_pem)
- remote_target = self.getSyncTarget('localhost', 'test')
- remote_target.record_sync_info('other-id', 2, 'T-id')
- self.assertEqual(
- (2, 'T-id'), db._get_replica_gen_and_trans_id('other-id'))
-
- def test_host_mismatch(self):
- """
- Test that SSL connections to a hostname different than the one in the
- certificate raise CertificateError.
-
- This test was adapted to patch Soledad's HTTPS connection custom class
- with the intended CA certificates.
- """
- self.startServer()
- self.request_state._create_database('test')
- self.patch(client, 'SOLEDAD_CERT', self.cacert_pem)
- remote_target = self.getSyncTarget('127.0.0.1', 'test')
- self.assertRaises(
- http_client.CertificateError, remote_target.record_sync_info,
- 'other-id', 2, 'T-id')
-
-
-#-----------------------------------------------------------------------------
-# The following tests come from `u1db.tests.test_http_database`.
-#-----------------------------------------------------------------------------
-
-class _HTTPDatabase(http_database.HTTPDatabase, auth.TokenBasedAuth):
- """
- Wraps our token auth implementation.
- """
-
- def set_token_credentials(self, uuid, token):
- auth.TokenBasedAuth.set_token_credentials(self, uuid, token)
-
- def _sign_request(self, method, url_query, params):
- return auth.TokenBasedAuth._sign_request(
- self, method, url_query, params)
-
-
-class TestHTTPDatabaseWithCreds(
- test_http_database.TestHTTPDatabaseCtrWithCreds):
-
- def test_get_sync_target_inherits_token_credentials(self):
- # this test was from TestDatabaseSimpleOperations but we put it here
- # for convenience.
- self.db = _HTTPDatabase('dbase')
- self.db.set_token_credentials('user-uuid', 'auth-token')
- st = self.db.get_sync_target()
- self.assertEqual(self.db._creds, st._creds)
-
- def test_ctr_with_creds(self):
- db1 = _HTTPDatabase('http://dbs/db', creds={'token': {
- 'uuid': 'user-uuid',
- 'token': 'auth-token',
- }})
- self.assertIn('token', db1._creds)
-
-
-#-----------------------------------------------------------------------------
-# The following tests come from `u1db.tests.test_sync`.
-#-----------------------------------------------------------------------------
-
-target_scenarios = [
- ('token_leap', {'create_db_and_target':
- make_local_db_and_token_soledad_target,
- 'make_app_with_state': make_soledad_app}),
-]
-
-
-class SoledadDatabaseSyncTargetTests(
- SoledadWithCouchServerMixin, test_sync.DatabaseSyncTargetTests):
-
- scenarios = (
- tests.multiply_scenarios(
- tests.DatabaseBaseTests.scenarios,
- target_scenarios))
-
- whitebox = False
-
- def setUp(self):
- self.main_test_class = test_sync.DatabaseSyncTargetTests
- SoledadWithCouchServerMixin.setUp(self)
-
- def test_sync_exchange(self):
- """
- Test sync exchange.
-
- This test was adapted to decrypt remote content before assert.
- """
- sol, _ = make_local_db_and_soledad_target(self)
- docs_by_gen = [
- (self.make_document('doc-id', 'replica:1', tests.simple_doc), 10,
- 'T-sid')]
- new_gen, trans_id = self.st.sync_exchange(
- docs_by_gen, 'replica', last_known_generation=0,
- last_known_trans_id=None, return_doc_cb=self.receive_doc)
- self.assertGetEncryptedDoc(
- self.db, 'doc-id', 'replica:1', tests.simple_doc, False)
- self.assertTransactionLog(['doc-id'], self.db)
- last_trans_id = self.getLastTransId(self.db)
- self.assertEqual(([], 1, last_trans_id),
- (self.other_changes, new_gen, last_trans_id))
- self.assertEqual(10, self.st.get_sync_info('replica')[3])
- sol.close()
-
- def test_sync_exchange_push_many(self):
- """
- Test sync exchange.
-
- This test was adapted to decrypt remote content before assert.
- """
- docs_by_gen = [
- (self.make_document(
- 'doc-id', 'replica:1', tests.simple_doc), 10, 'T-1'),
- (self.make_document(
- 'doc-id2', 'replica:1', tests.nested_doc), 11, 'T-2')]
- new_gen, trans_id = self.st.sync_exchange(
- docs_by_gen, 'replica', last_known_generation=0,
- last_known_trans_id=None, return_doc_cb=self.receive_doc)
- self.assertGetEncryptedDoc(
- self.db, 'doc-id', 'replica:1', tests.simple_doc, False)
- self.assertGetEncryptedDoc(
- self.db, 'doc-id2', 'replica:1', tests.nested_doc, False)
- self.assertTransactionLog(['doc-id', 'doc-id2'], self.db)
- last_trans_id = self.getLastTransId(self.db)
- self.assertEqual(([], 2, last_trans_id),
- (self.other_changes, new_gen, trans_id))
- self.assertEqual(11, self.st.get_sync_info('replica')[3])
-
- def test_sync_exchange_returns_many_new_docs(self):
- """
- Test sync exchange.
-
- This test was adapted to avoid JSON serialization comparison as local
- and remote representations might differ. It looks directly at the
- doc's contents instead.
- """
- doc = self.db.create_doc_from_json(tests.simple_doc)
- doc2 = self.db.create_doc_from_json(tests.nested_doc)
- self.assertTransactionLog([doc.doc_id, doc2.doc_id], self.db)
- new_gen, _ = self.st.sync_exchange(
- [], 'other-replica', last_known_generation=0,
- last_known_trans_id=None, return_doc_cb=self.receive_doc)
- self.assertTransactionLog([doc.doc_id, doc2.doc_id], self.db)
- self.assertEqual(2, new_gen)
- self.assertEqual(
- [(doc.doc_id, doc.rev, 1),
- (doc2.doc_id, doc2.rev, 2)],
- [c[:-3] + c[-2:-1] for c in self.other_changes])
- self.assertEqual(
- json.loads(tests.simple_doc),
- json.loads(self.other_changes[0][2]))
- self.assertEqual(
- json.loads(tests.nested_doc),
- json.loads(self.other_changes[1][2]))
- if self.whitebox:
- self.assertEqual(
- self.db._last_exchange_log['return'],
- {'last_gen': 2, 'docs':
- [(doc.doc_id, doc.rev), (doc2.doc_id, doc2.rev)]})
-
-
-class TestSoledadDbSync(
- SoledadWithCouchServerMixin, test_sync.TestDbSync):
- """Test db.sync remote sync shortcut"""
-
- scenarios = [
- ('py-token-http', {
- 'create_db_and_target': make_local_db_and_token_soledad_target,
- 'make_app_with_state': make_token_soledad_app,
- 'make_database_for_test': make_sqlcipher_database_for_test,
- 'token': True
- }),
- ]
-
- oauth = False
- token = False
-
- def setUp(self):
- self.main_test_class = test_sync.TestDbSync
- SoledadWithCouchServerMixin.setUp(self)
-
- def tearDown(self):
- SoledadWithCouchServerMixin.tearDown(self)
- self.db.close()
-
- def do_sync(self, target_name):
- """
- Perform sync using SoledadSyncTarget and Token auth.
- """
- if self.token:
- extra = dict(creds={'token': {
- 'uuid': 'user-uuid',
- 'token': 'auth-token',
- }})
- target_url = self.getURL(target_name)
- return Synchronizer(
- self.db,
- target.SoledadSyncTarget(
- target_url,
- crypto=self._soledad._crypto,
- **extra)).sync(autocreate=True)
- else:
- return test_sync.TestDbSync.do_sync(self, target_name)
-
- def test_db_sync(self):
- """
- Test sync.
-
- Adapted to check for encrypted content.
- """
- doc1 = self.db.create_doc_from_json(tests.simple_doc)
- doc2 = self.db2.create_doc_from_json(tests.nested_doc)
- local_gen_before_sync = self.do_sync('test2')
- gen, _, changes = self.db.whats_changed(local_gen_before_sync)
- self.assertEqual(1, len(changes))
- self.assertEqual(doc2.doc_id, changes[0][0])
- self.assertEqual(1, gen - local_gen_before_sync)
- self.assertGetEncryptedDoc(
- self.db2, doc1.doc_id, doc1.rev, tests.simple_doc, False)
- self.assertGetEncryptedDoc(
- self.db, doc2.doc_id, doc2.rev, tests.nested_doc, False)
-
- def test_db_sync_autocreate(self):
- """
- We bypass this test because we never need to autocreate databases.
- """
- pass
-
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_target_soledad.py b/common/src/leap/soledad/common/tests/test_target_soledad.py
deleted file mode 100644
index 899203b8..00000000
--- a/common/src/leap/soledad/common/tests/test_target_soledad.py
+++ /dev/null
@@ -1,102 +0,0 @@
-from u1db.remote import (
- http_database,
-)
-
-from leap.soledad.client import (
- auth,
- VerifiedHTTPSConnection,
-)
-from leap.soledad.common.document import SoledadDocument
-from leap.soledad.server import SoledadApp
-from leap.soledad.server.auth import SoledadTokenAuthMiddleware
-
-
-from leap.soledad.common.tests import u1db_tests as tests
-from leap.soledad.common.tests import BaseSoledadTest
-from leap.soledad.common.tests.u1db_tests import test_backends
-
-
-#-----------------------------------------------------------------------------
-# The following tests come from `u1db.tests.test_backends`.
-#-----------------------------------------------------------------------------
-
-def make_leap_document_for_test(test, doc_id, rev, content,
- has_conflicts=False):
- return SoledadDocument(
- doc_id, rev, content, has_conflicts=has_conflicts)
-
-
-def make_soledad_app(state):
- return SoledadApp(state)
-
-
-def make_token_soledad_app(state):
- app = SoledadApp(state)
-
- def _verify_authentication_data(uuid, auth_data):
- if uuid == 'user-uuid' and auth_data == 'auth-token':
- return True
- return False
-
- # we test for action authorization in leap.soledad.common.tests.test_server
- def _verify_authorization(uuid, environ):
- return True
-
- application = SoledadTokenAuthMiddleware(app)
- application._verify_authentication_data = _verify_authentication_data
- application._verify_authorization = _verify_authorization
- return application
-
-
-LEAP_SCENARIOS = [
- ('http', {
- 'make_database_for_test': test_backends.make_http_database_for_test,
- 'copy_database_for_test': test_backends.copy_http_database_for_test,
- 'make_document_for_test': make_leap_document_for_test,
- 'make_app_with_state': make_soledad_app}),
-]
-
-
-def make_token_http_database_for_test(test, replica_uid):
- test.startServer()
- test.request_state._create_database(replica_uid)
-
- class _HTTPDatabaseWithToken(
- http_database.HTTPDatabase, auth.TokenBasedAuth):
-
- def set_token_credentials(self, uuid, token):
- auth.TokenBasedAuth.set_token_credentials(self, uuid, token)
-
- def _sign_request(self, method, url_query, params):
- return auth.TokenBasedAuth._sign_request(
- self, method, url_query, params)
-
- http_db = _HTTPDatabaseWithToken(test.getURL('test'))
- http_db.set_token_credentials('user-uuid', 'auth-token')
- return http_db
-
-
-def copy_token_http_database_for_test(test, db):
- # DO NOT COPY OR REUSE THIS CODE OUTSIDE TESTS: COPYING U1DB DATABASES IS
- # THE WRONG THING TO DO, THE ONLY REASON WE DO SO HERE IS TO TEST THAT WE
- # CORRECTLY DETECT IT HAPPENING SO THAT WE CAN RAISE ERRORS RATHER THAN
- # CORRUPT USER DATA. USE SYNC INSTEAD, OR WE WILL SEND NINJA TO YOUR
- # HOUSE.
- http_db = test.request_state._copy_database(db)
- http_db.set_token_credentials(http_db, 'user-uuid', 'auth-token')
- return http_db
-
-
-class SoledadTests(test_backends.AllDatabaseTests, BaseSoledadTest):
-
- scenarios = LEAP_SCENARIOS + [
- ('token_http', {'make_database_for_test':
- make_token_http_database_for_test,
- 'copy_database_for_test':
- copy_token_http_database_for_test,
- 'make_document_for_test': make_leap_document_for_test,
- 'make_app_with_state': make_token_soledad_app,
- })
- ]
-
-load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/u1db_tests/__init__.py b/common/src/leap/soledad/common/tests/u1db_tests/__init__.py
index ad66fb06..6efeb87f 100644
--- a/common/src/leap/soledad/common/tests/u1db_tests/__init__.py
+++ b/common/src/leap/soledad/common/tests/u1db_tests/__init__.py
@@ -35,7 +35,7 @@ from pysqlcipher import dbapi2
from StringIO import StringIO
import testscenarios
-import testtools
+from twisted.trial import unittest
from u1db import (
errors,
@@ -50,7 +50,7 @@ from u1db.remote import (
)
-class TestCase(testtools.TestCase):
+class TestCase(unittest.TestCase):
def createTempDir(self, prefix='u1db-tmp-'):
"""Create a temporary directory to do some work in.
diff --git a/common/src/leap/soledad/common/tests/u1db_tests/test_https.py b/common/src/leap/soledad/common/tests/u1db_tests/test_https.py
index c086fbc0..c5b316ed 100644
--- a/common/src/leap/soledad/common/tests/u1db_tests/test_https.py
+++ b/common/src/leap/soledad/common/tests/u1db_tests/test_https.py
@@ -75,7 +75,7 @@ class TestHttpSyncTargetHttpsSupport(tests.TestCaseWithServer):
# order to maintain the compatibility with u1db default tests, we undo
# that replacement here.
http_client._VerifiedHTTPSConnection = \
- soledad.client.old__VerifiedHTTPSConnection
+ soledad.client.api.old__VerifiedHTTPSConnection
super(TestHttpSyncTargetHttpsSupport, self).setUp()
def getSyncTarget(self, host, path=None):
diff --git a/common/src/leap/soledad/common/tests/util.py b/common/src/leap/soledad/common/tests/util.py
index b27e6356..d4439ef4 100644
--- a/common/src/leap/soledad/common/tests/util.py
+++ b/common/src/leap/soledad/common/tests/util.py
@@ -21,24 +21,40 @@ Utilities used by multiple test suites.
"""
+import os
import tempfile
import shutil
+import random
+import string
+import u1db
+import subprocess
+import time
+import re
+
+from mock import Mock
from urlparse import urljoin
-
from StringIO import StringIO
from pysqlcipher import dbapi2
+
from u1db.errors import DatabaseDoesNotExist
+from u1db.remote import http_database
+
+from twisted.trial import unittest
+from leap.common.files import mkdir_p
from leap.soledad.common import soledad_assert
+from leap.soledad.common.document import SoledadDocument
from leap.soledad.common.couch import CouchDatabase, CouchServerState
-from leap.soledad.server import SoledadApp
-from leap.soledad.server.auth import SoledadTokenAuthMiddleware
+from leap.soledad.common.crypto import ENC_SCHEME_KEY
+from leap.soledad.client import Soledad
+from leap.soledad.client import target
+from leap.soledad.client import auth
+from leap.soledad.client.crypto import decrypt_doc_dict
-from leap.soledad.common.tests import BaseSoledadTest
-from leap.soledad.common.tests.test_couch import CouchDBWrapper, CouchDBTestCase
-
+from leap.soledad.server import SoledadApp
+from leap.soledad.server.auth import SoledadTokenAuthMiddleware
from leap.soledad.client.sqlcipher import (
SQLCipherDatabase,
@@ -47,6 +63,7 @@ from leap.soledad.client.sqlcipher import (
PASSWORD = '123456'
+ADDRESS = 'leap@leap.se'
def make_sqlcipher_database_for_test(test, replica_uid):
@@ -62,7 +79,7 @@ def copy_sqlcipher_database_for_test(test, db):
# CORRECTLY DETECT IT HAPPENING SO THAT WE CAN RAISE ERRORS RATHER THAN
# CORRUPT USER DATA. USE SYNC INSTEAD, OR WE WILL SEND NINJA TO YOUR
# HOUSE.
- new_db = SQLCipherDatabase(':memory:', PASSWORD)
+ new_db = make_sqlcipher_database_for_test(test, None)
tmpfile = StringIO()
for line in db._db_handle.iterdump():
if not 'sqlite_sequence' in line: # work around bug in iterdump
@@ -98,6 +115,295 @@ def make_token_soledad_app(state):
return application
+def make_soledad_document_for_test(test, doc_id, rev, content,
+ has_conflicts=False):
+ return SoledadDocument(
+ doc_id, rev, content, has_conflicts=has_conflicts)
+
+
+def make_token_http_database_for_test(test, replica_uid):
+ test.startServer()
+ test.request_state._create_database(replica_uid)
+
+ class _HTTPDatabaseWithToken(
+ http_database.HTTPDatabase, auth.TokenBasedAuth):
+
+ def set_token_credentials(self, uuid, token):
+ auth.TokenBasedAuth.set_token_credentials(self, uuid, token)
+
+ def _sign_request(self, method, url_query, params):
+ return auth.TokenBasedAuth._sign_request(
+ self, method, url_query, params)
+
+ http_db = _HTTPDatabaseWithToken(test.getURL('test'))
+ http_db.set_token_credentials('user-uuid', 'auth-token')
+ return http_db
+
+
+def copy_token_http_database_for_test(test, db):
+ # DO NOT COPY OR REUSE THIS CODE OUTSIDE TESTS: COPYING U1DB DATABASES IS
+ # THE WRONG THING TO DO, THE ONLY REASON WE DO SO HERE IS TO TEST THAT WE
+ # CORRECTLY DETECT IT HAPPENING SO THAT WE CAN RAISE ERRORS RATHER THAN
+ # CORRUPT USER DATA. USE SYNC INSTEAD, OR WE WILL SEND NINJA TO YOUR
+ # HOUSE.
+ http_db = test.request_state._copy_database(db)
+ http_db.set_token_credentials(http_db, 'user-uuid', 'auth-token')
+ return http_db
+
+
+class MockedSharedDBTest(object):
+
+ def get_default_shared_mock(self, put_doc_side_effect=None,
+ get_doc_return_value=None):
+ """
+ Get a default class for mocking the shared DB
+ """
+ class defaultMockSharedDB(object):
+ get_doc = Mock(return_value=get_doc_return_value)
+ put_doc = Mock(side_effect=put_doc_side_effect)
+ lock = Mock(return_value=('atoken', 300))
+ unlock = Mock(return_value=True)
+ open = Mock(return_value=None)
+ syncable = True
+
+ def __call__(self):
+ return self
+ return defaultMockSharedDB
+
+
+def soledad_sync_target(test, path):
+ return target.SoledadSyncTarget(
+ test.getURL(path), crypto=test._soledad._crypto)
+
+
+def token_soledad_sync_target(test, path):
+ st = soledad_sync_target(test, path)
+ st.set_token_credentials('user-uuid', 'auth-token')
+ return st
+
+
+class BaseSoledadTest(unittest.TestCase, MockedSharedDBTest):
+ """
+ Instantiates Soledad for usage in tests.
+ """
+ defer_sync_encryption = False
+
+ def setUp(self):
+ # The following snippet comes from BaseLeapTest.setUpClass, but we
+ # repeat it here because twisted.trial does not work with
+ # setUpClass/tearDownClass.
+ 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
+
+ # config info
+ self.db1_file = os.path.join(self.tempdir, "db1.u1db")
+ self.db2_file = os.path.join(self.tempdir, "db2.u1db")
+ self.email = ADDRESS
+ # open test dbs
+ self._db1 = u1db.open(self.db1_file, create=True,
+ document_factory=SoledadDocument)
+ self._db2 = u1db.open(self.db2_file, create=True,
+ document_factory=SoledadDocument)
+ # get a random prefix for each test, so we do not mess with
+ # concurrency during initialization and shutting down of
+ # each local db.
+ self.rand_prefix = ''.join(
+ map(lambda x: random.choice(string.ascii_letters), range(6)))
+ # initialize soledad by hand so we can control keys
+ # XXX check if this soledad is actually used
+ self._soledad = self._soledad_instance(
+ prefix=self.rand_prefix, user=self.email)
+
+ def tearDown(self):
+ self._db1.close()
+ self._db2.close()
+ self._soledad.close()
+
+ # restore paths
+ os.environ["PATH"] = self.old_path
+ os.environ["HOME"] = self.old_home
+
+ def _delete_temporary_dirs():
+ # XXX should not access "private" attrs
+ for f in [self._soledad.local_db_path,
+ self._soledad.secrets.secrets_path]:
+ if os.path.isfile(f):
+ os.unlink(f)
+ # The following snippet comes from BaseLeapTest.setUpClass, but we
+ # repeat it here because twisted.trial does not work with
+ # setUpClass/tearDownClass.
+ soledad_assert(
+ self.tempdir.startswith('/tmp/leap_tests-'),
+ "beware! tried to remove a dir which does not "
+ "live in temporal folder!")
+ shutil.rmtree(self.tempdir)
+
+ from twisted.internet import reactor
+ reactor.addSystemEventTrigger(
+ "after", "shutdown", _delete_temporary_dirs)
+
+
+ def _soledad_instance(self, user=ADDRESS, passphrase=u'123',
+ prefix='',
+ secrets_path='secrets.json',
+ local_db_path='soledad.u1db',
+ server_url='https://127.0.0.1/',
+ cert_file=None,
+ shared_db_class=None,
+ auth_token='auth-token'):
+
+ def _put_doc_side_effect(doc):
+ self._doc_put = doc
+
+ if shared_db_class is not None:
+ MockSharedDB = shared_db_class
+ else:
+ MockSharedDB = self.get_default_shared_mock(
+ _put_doc_side_effect)
+
+ return Soledad(
+ user,
+ passphrase,
+ secrets_path=os.path.join(
+ self.tempdir, prefix, secrets_path),
+ local_db_path=os.path.join(
+ self.tempdir, prefix, local_db_path),
+ server_url=server_url, # Soledad will fail if not given an url.
+ cert_file=cert_file,
+ defer_encryption=self.defer_sync_encryption,
+ shared_db=MockSharedDB(),
+ auth_token=auth_token)
+
+ def assertGetEncryptedDoc(
+ self, db, doc_id, doc_rev, content, has_conflicts):
+ """
+ Assert that the document in the database looks correct.
+ """
+ exp_doc = self.make_document(doc_id, doc_rev, content,
+ has_conflicts=has_conflicts)
+ doc = db.get_doc(doc_id)
+
+ if ENC_SCHEME_KEY in doc.content:
+ # XXX check for SYM_KEY too
+ key = self._soledad._crypto.doc_passphrase(doc.doc_id)
+ secret = self._soledad._crypto.secret
+ decrypted = decrypt_doc_dict(
+ doc.content, doc.doc_id, doc.rev,
+ key, secret)
+ doc.set_json(decrypted)
+ self.assertEqual(exp_doc.doc_id, doc.doc_id)
+ self.assertEqual(exp_doc.rev, doc.rev)
+ self.assertEqual(exp_doc.has_conflicts, doc.has_conflicts)
+ self.assertEqual(exp_doc.content, doc.content)
+
+
+#-----------------------------------------------------------------------------
+# A wrapper for running couchdb locally.
+#-----------------------------------------------------------------------------
+
+# from: https://github.com/smcq/paisley/blob/master/paisley/test/util.py
+# TODO: include license of above project.
+class CouchDBWrapper(object):
+ """
+ Wrapper for external CouchDB instance which is started and stopped for
+ testing.
+ """
+
+ def start(self):
+ """
+ Start a CouchDB instance for a test.
+ """
+ self.tempdir = tempfile.mkdtemp(suffix='.couch.test')
+
+ path = os.path.join(os.path.dirname(__file__),
+ 'couchdb.ini.template')
+ handle = open(path)
+ conf = handle.read() % {
+ 'tempdir': self.tempdir,
+ }
+ handle.close()
+
+ confPath = os.path.join(self.tempdir, 'test.ini')
+ handle = open(confPath, 'w')
+ handle.write(conf)
+ handle.close()
+
+ # create the dirs from the template
+ mkdir_p(os.path.join(self.tempdir, 'lib'))
+ mkdir_p(os.path.join(self.tempdir, 'log'))
+ args = ['/usr/bin/couchdb', '-n', '-a', confPath]
+ null = open('/dev/null', 'w')
+
+ self.process = subprocess.Popen(
+ args, env=None, stdout=null.fileno(), stderr=null.fileno(),
+ close_fds=True)
+ # find port
+ logPath = os.path.join(self.tempdir, 'log', 'couch.log')
+ while not os.path.exists(logPath):
+ if self.process.poll() is not None:
+ got_stdout, got_stderr = "", ""
+ if self.process.stdout is not None:
+ got_stdout = self.process.stdout.read()
+
+ if self.process.stderr is not None:
+ got_stderr = self.process.stderr.read()
+ raise Exception("""
+couchdb exited with code %d.
+stdout:
+%s
+stderr:
+%s""" % (
+ self.process.returncode, got_stdout, got_stderr))
+ time.sleep(0.01)
+ while os.stat(logPath).st_size == 0:
+ time.sleep(0.01)
+ PORT_RE = re.compile(
+ 'Apache CouchDB has started on http://127.0.0.1:(?P<port>\d+)')
+
+ handle = open(logPath)
+ line = handle.read()
+ handle.close()
+ m = PORT_RE.search(line)
+ if not m:
+ self.stop()
+ raise Exception("Cannot find port in line %s" % line)
+ self.port = int(m.group('port'))
+
+ def stop(self):
+ """
+ Terminate the CouchDB instance.
+ """
+ self.process.terminate()
+ self.process.communicate()
+ shutil.rmtree(self.tempdir)
+
+
+class CouchDBTestCase(unittest.TestCase, MockedSharedDBTest):
+ """
+ TestCase base class for tests against a real CouchDB server.
+ """
+
+ def setUp(self):
+ """
+ Make sure we have a CouchDB instance for a test.
+ """
+ self.wrapper = CouchDBWrapper()
+ self.wrapper.start()
+ #self.db = self.wrapper.db
+
+ def tearDown(self):
+ """
+ Stop CouchDB instance for test.
+ """
+ self.wrapper.stop()
+
class CouchServerStateForTests(CouchServerState):
"""
This is a slightly modified CouchDB server state that allows for creating
@@ -126,43 +432,15 @@ class SoledadWithCouchServerMixin(
BaseSoledadTest,
CouchDBTestCase):
- @classmethod
- def setUpClass(cls):
- """
- Make sure we have a CouchDB instance for a test.
- """
- # from BaseLeapTest
- cls.tempdir = tempfile.mkdtemp(prefix="leap_tests-")
- # from CouchDBTestCase
- cls.wrapper = CouchDBWrapper()
- cls.wrapper.start()
- #self.db = self.wrapper.db
-
- @classmethod
- def tearDownClass(cls):
- """
- Stop CouchDB instance for test.
- """
- # from BaseLeapTest
- soledad_assert(
- cls.tempdir.startswith('/tmp/leap_tests-'),
- "beware! tried to remove a dir which does not "
- "live in temporal folder!")
- shutil.rmtree(cls.tempdir)
- # from CouchDBTestCase
- cls.wrapper.stop()
-
def setUp(self):
- BaseSoledadTest.setUp(self)
CouchDBTestCase.setUp(self)
+ BaseSoledadTest.setUp(self)
main_test_class = getattr(self, 'main_test_class', None)
if main_test_class is not None:
main_test_class.setUp(self)
self._couch_url = 'http://localhost:%d' % self.wrapper.port
def tearDown(self):
- BaseSoledadTest.tearDown(self)
- CouchDBTestCase.tearDown(self)
main_test_class = getattr(self, 'main_test_class', None)
if main_test_class is not None:
main_test_class.tearDown(self)
@@ -172,6 +450,8 @@ class SoledadWithCouchServerMixin(
db.delete_database()
except DatabaseDoesNotExist:
pass
+ BaseSoledadTest.tearDown(self)
+ CouchDBTestCase.tearDown(self)
def make_app(self):
couch_url = urljoin(
--
cgit v1.2.3