diff options
Diffstat (limited to 'common/src/leap/soledad/common/tests/util.py')
| -rw-r--r-- | common/src/leap/soledad/common/tests/util.py | 352 |
1 files changed, 316 insertions, 36 deletions
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( |