diff options
Diffstat (limited to 'testing/test_soledad/util.py')
| -rw-r--r-- | testing/test_soledad/util.py | 399 |
1 files changed, 0 insertions, 399 deletions
diff --git a/testing/test_soledad/util.py b/testing/test_soledad/util.py deleted file mode 100644 index ca8d098d..00000000 --- a/testing/test_soledad/util.py +++ /dev/null @@ -1,399 +0,0 @@ -# -*- CODING: UTF-8 -*- -# util.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/>. - -""" -Utilities used by multiple test suites. -""" - -import os -import random -import string -import couchdb -import pytest -import sys - -from six.moves.urllib.parse import urljoin -from six import StringIO -from uuid import uuid4 -from mock import Mock - -from twisted.trial import unittest - -from leap.common.testing.basetest import BaseLeapTest - -from leap.soledad.common import l2db -from leap.soledad.common.l2db import sync -from leap.soledad.common.l2db.remote import http_database - -from leap.soledad.common.document import SoledadDocument -from leap.soledad.common.couch import CouchDatabase -from leap.soledad.common.couch.state import CouchServerState - -from leap.soledad.client import Soledad -from leap.soledad.client import http_target -from leap.soledad.client import auth -from leap.soledad.client._crypto import is_symmetrically_encrypted -from leap.soledad.client._db.sqlcipher import SQLCipherDatabase -from leap.soledad.client._db.sqlcipher import SQLCipherOptions - -from leap.soledad.server import SoledadApp - -if sys.version_info[0] < 3: - from pysqlcipher import dbapi2 -else: - from pysqlcipher3 import dbapi2 - - -PASSWORD = '123456' -ADDRESS = 'user-1234' - - -def make_local_db_and_target(test): - db = test.create_database('test') - st = db.get_sync_target() - return db, st - - -def make_document_for_test(test, doc_id, rev, content, has_conflicts=False): - return SoledadDocument(doc_id, rev, content, has_conflicts=has_conflicts) - - -def make_sqlcipher_database_for_test(test, replica_uid): - db = SQLCipherDatabase( - SQLCipherOptions(':memory:', PASSWORD)) - db._set_replica_uid(replica_uid) - return db - - -def copy_sqlcipher_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. - new_db = make_sqlcipher_database_for_test(test, None) - tmpfile = StringIO() - for line in db._db_handle.iterdump(): - if 'sqlite_sequence' not in line: # work around bug in iterdump - tmpfile.write('%s\n' % line) - tmpfile.seek(0) - new_db._db_handle = dbapi2.connect(':memory:') - new_db._db_handle.cursor().executescript(tmpfile.read()) - new_db._db_handle.commit() - new_db._set_replica_uid(db._replica_uid) - new_db._factory = db._factory - return new_db - - -SQLCIPHER_SCENARIOS = [ - ('sqlcipher', {'make_database_for_test': make_sqlcipher_database_for_test, - 'copy_database_for_test': copy_sqlcipher_database_for_test, - 'make_document_for_test': make_document_for_test, }), -] - - -def make_soledad_app(state): - return SoledadApp(state) - - -def make_token_soledad_app(state): - application = SoledadApp(state) - - def _verify_authentication_data(uuid, auth_data): - if uuid.startswith('user-') 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._verify_authentication_data = _verify_authentication_data - application._verify_authorization = _verify_authorization - 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 - - -def sync_via_synchronizer(test, db_source, db_target, trace_hook=None, - trace_hook_shallow=None): - target = db_target.get_sync_target() - trace_hook = trace_hook or trace_hook_shallow - if trace_hook: - target._set_trace_hook(trace_hook) - return sync.Synchronizer(db_source, target).sync() - - -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) - open = Mock(return_value=None) - close = Mock(return_value=None) - - def __call__(self): - return self - return defaultMockSharedDB - - -def soledad_sync_target( - test, path, source_replica_uid=uuid4().hex): - creds = {'token': { - 'uuid': 'user-uuid', - 'token': 'auth-token', - }} - return http_target.SoledadHTTPSyncTarget( - test.getURL(path), - source_replica_uid, - creds, - test._soledad._crypto, - None) # cert_file - - -# redefine the base leap test class so it inherits from twisted trial's -# TestCase. This is needed so trial knows that it has to manage a reactor and -# wait for deferreds returned by tests to be fired. - -BaseLeapTest = type( - 'BaseLeapTest', (unittest.TestCase,), dict(BaseLeapTest.__dict__)) - - -class BaseSoledadTest(BaseLeapTest, MockedSharedDBTest): - - """ - Instantiates Soledad for usage in tests. - """ - - @pytest.mark.usefixtures("method_tmpdir") - 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.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 = l2db.open(self.db1_file, create=True, - document_factory=SoledadDocument) - self._db2 = l2db.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() - - 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) - - 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) - - soledad = 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, - shared_db=MockSharedDB(), - auth_token=auth_token, - with_blobs=True) - self.addCleanup(soledad.close) - return soledad - - @pytest.inlineCallbacks - 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 is_symmetrically_encrypted(doc.content['raw']): - crypt = self._soledad._crypto - decrypted = yield crypt.decrypt_doc(doc) - 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) - - -@pytest.mark.usefixtures("couch_url") -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.couch_server = couchdb.Server(self.couch_url) - - def delete_db(self, name): - try: - self.couch_server.delete(name) - except: - # ignore if already missing - pass - - -class CouchServerStateForTests(CouchServerState): - - """ - This is a slightly modified CouchDB server state that allows for creating - a database. - - Ordinarily, the CouchDB server state does not allow some operations, - because for security purposes the Soledad Server should not even have - enough permissions to perform them. For tests, we allow database creation, - otherwise we'd have to create those databases in setUp/tearDown methods, - which is less pleasant than allowing the db to be automatically created. - """ - - def __init__(self, *args, **kwargs): - self.dbs = [] - super(CouchServerStateForTests, self).__init__(*args, **kwargs) - - def _create_database(self, replica_uid=None, dbname=None): - """ - Create db and append to a list, allowing test to close it later - """ - dbname = dbname or ('test-%s' % uuid4().hex) - db = CouchDatabase.open_database( - urljoin(self.couch_url, dbname), - True, - replica_uid=replica_uid or 'test') - self.dbs.append(db) - return db - - def ensure_database(self, dbname): - db = self._create_database(dbname=dbname) - return db, db.replica_uid - - -class SoledadWithCouchServerMixin( - BaseSoledadTest, - CouchDBTestCase): - - def 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) - - def tearDown(self): - main_test_class = getattr(self, 'main_test_class', None) - if main_test_class is not None: - main_test_class.tearDown(self) - # delete the test database - BaseSoledadTest.tearDown(self) - CouchDBTestCase.tearDown(self) - - def make_app(self): - self.request_state = CouchServerStateForTests(self.couch_url) - self.addCleanup(self.delete_dbs) - return self.make_app_with_state(self.request_state) - - def delete_dbs(self): - for db in self.request_state.dbs: - self.delete_db(db._dbname) |