diff options
| -rw-r--r-- | testing/tests/conftest.py | 189 | ||||
| -rw-r--r-- | testing/tests/perf/conftest.py | 188 |
2 files changed, 189 insertions, 188 deletions
diff --git a/testing/tests/conftest.py b/testing/tests/conftest.py index 9e4319ac..5698c8a9 100644 --- a/testing/tests/conftest.py +++ b/testing/tests/conftest.py @@ -1,4 +1,29 @@ +import json +import os import pytest +import requests +import signal +import time + +from hashlib import sha512 +from subprocess import call +from urlparse import urljoin +from uuid import uuid4 + +from leap.soledad.common.couch import CouchDatabase +from leap.soledad.client import Soledad + + +# +# default options for all tests +# + +DEFAULT_PASSPHRASE = '123' + +DEFAULT_URL = 'http://127.0.0.1:2424' +DEFAULT_PRIVKEY = 'soledad_privkey.pem' +DEFAULT_CERTKEY = 'soledad_certkey.pem' +DEFAULT_TOKEN = 'an-auth-token' def pytest_addoption(parser): @@ -16,3 +41,167 @@ def couch_url(request): @pytest.fixture def method_tmpdir(request, tmpdir): request.instance.tempdir = tmpdir.strpath + + +# +# remote_db fixture: provides an empty database for a given user in a per +# function scope. +# + +class UserDatabase(object): + + def __init__(self, url, uuid): + self._remote_db_url = urljoin(url, 'user-%s' % uuid) + + def setup(self): + return CouchDatabase.open_database( + url=self._remote_db_url, create=True, replica_uid=None) + + def teardown(self): + requests.delete(self._remote_db_url) + + +@pytest.fixture() +def remote_db(request): + couch_url = request.config.option.couch_url + + def create(uuid): + db = UserDatabase(couch_url, uuid) + request.addfinalizer(db.teardown) + return db.setup() + return create + + +def get_pid(pidfile): + if not os.path.isfile(pidfile): + return 0 + try: + with open(pidfile) as f: + return int(f.read()) + except IOError: + return 0 + + +# +# soledad_server fixture: provides a running soledad server in a per module +# context (same soledad server for all tests in this module). +# + +class SoledadServer(object): + + def __init__(self, tmpdir_factory, couch_url): + tmpdir = tmpdir_factory.mktemp('soledad-server') + self._pidfile = os.path.join(tmpdir.strpath, 'soledad-server.pid') + self._logfile = os.path.join(tmpdir.strpath, 'soledad-server.log') + self._couch_url = couch_url + + def start(self): + self._create_conf_file() + # start the server + call([ + 'twistd', + '--logfile=%s' % self._logfile, + '--pidfile=%s' % self._pidfile, + 'web', + '--wsgi=leap.soledad.server.application', + '--port=2424' + ]) + + def _create_conf_file(self): + if not os.access('/etc', os.W_OK): + return + if not os.path.isdir('/etc/soledad'): + os.mkdir('/etc/soledad') + with open('/etc/soledad/soledad-server.conf', 'w') as f: + content = '[soledad-server]\ncouch_url = %s' % self._couch_url + f.write(content) + + def stop(self): + pid = get_pid(self._pidfile) + os.kill(pid, signal.SIGKILL) + + +@pytest.fixture(scope='module') +def soledad_server(tmpdir_factory, request): + couch_url = request.config.option.couch_url + server = SoledadServer(tmpdir_factory, couch_url) + server.start() + request.addfinalizer(server.stop) + return server + + +# +# soledad_dbs fixture: provides all databases needed by soledad server in a per +# module scope (same databases for all tests in this module). +# + +def _token_dbname(): + dbname = 'tokens_' + \ + str(int(time.time() / (30 * 24 * 3600))) + return dbname + + +class SoledadDatabases(object): + + def __init__(self, url): + self._token_db_url = urljoin(url, _token_dbname()) + self._shared_db_url = urljoin(url, 'shared') + + def setup(self, uuid): + self._create_dbs() + self._add_token(uuid) + + def _create_dbs(self): + requests.put(self._token_db_url) + requests.put(self._shared_db_url) + + def _add_token(self, uuid): + token = sha512(DEFAULT_TOKEN).hexdigest() + content = {'type': 'Token', 'user_id': uuid} + requests.put( + self._token_db_url + '/' + token, data=json.dumps(content)) + + def teardown(self): + requests.delete(self._token_db_url) + requests.delete(self._shared_db_url) + + +@pytest.fixture() +def soledad_dbs(request): + couch_url = request.config.option.couch_url + + def create(uuid): + db = SoledadDatabases(couch_url) + request.addfinalizer(db.teardown) + return db.setup(uuid) + return create + + +# +# soledad_client fixture: provides a clean soledad client for a test function. +# + +@pytest.fixture() +def soledad_client(tmpdir, soledad_server, remote_db, soledad_dbs, request): + passphrase = DEFAULT_PASSPHRASE + server_url = DEFAULT_URL + token = DEFAULT_TOKEN + default_uuid = uuid4().hex + remote_db(default_uuid) + soledad_dbs(default_uuid) + + # get a soledad instance + def create(): + secrets_path = os.path.join(tmpdir.strpath, '%s.secret' % default_uuid) + local_db_path = os.path.join(tmpdir.strpath, '%s.db' % default_uuid) + soledad_client = Soledad( + default_uuid, + unicode(passphrase), + secrets_path=secrets_path, + local_db_path=local_db_path, + server_url=server_url, + cert_file=None, + auth_token=token) + request.addfinalizer(soledad_client.close) + return soledad_client + return create diff --git a/testing/tests/perf/conftest.py b/testing/tests/perf/conftest.py index 0700f1e2..a9cc3464 100644 --- a/testing/tests/perf/conftest.py +++ b/testing/tests/perf/conftest.py @@ -1,21 +1,9 @@ -import json -import os import pytest -import requests import random import base64 -import signal -import time -from hashlib import sha512 -from uuid import uuid4 -from subprocess import call -from urlparse import urljoin from twisted.internet import threads, reactor -from leap.soledad.client import Soledad -from leap.soledad.common.couch import CouchDatabase - # we have to manually setup the events server in order to be able to signal # events. This is usually done by the enclosing application using soledad @@ -30,18 +18,6 @@ def pytest_addoption(parser): help="the number of documents to use in performance tests") -# -# default options for all tests -# - -DEFAULT_PASSPHRASE = '123' - -DEFAULT_URL = 'http://127.0.0.1:2424' -DEFAULT_PRIVKEY = 'soledad_privkey.pem' -DEFAULT_CERTKEY = 'soledad_certkey.pem' -DEFAULT_TOKEN = 'an-auth-token' - - @pytest.fixture() def payload(): def generate(size): @@ -52,140 +28,6 @@ def payload(): return generate -# -# soledad_dbs fixture: provides all databases needed by soledad server in a per -# module scope (same databases for all tests in this module). -# - -def _token_dbname(): - dbname = 'tokens_' + \ - str(int(time.time() / (30 * 24 * 3600))) - return dbname - - -class SoledadDatabases(object): - - def __init__(self, url): - self._token_db_url = urljoin(url, _token_dbname()) - self._shared_db_url = urljoin(url, 'shared') - - def setup(self, uuid): - self._create_dbs() - self._add_token(uuid) - - def _create_dbs(self): - requests.put(self._token_db_url) - requests.put(self._shared_db_url) - - def _add_token(self, uuid): - token = sha512(DEFAULT_TOKEN).hexdigest() - content = {'type': 'Token', 'user_id': uuid} - requests.put( - self._token_db_url + '/' + token, data=json.dumps(content)) - - def teardown(self): - requests.delete(self._token_db_url) - requests.delete(self._shared_db_url) - - -@pytest.fixture() -def soledad_dbs(request): - couch_url = request.config.option.couch_url - - def create(uuid): - db = SoledadDatabases(couch_url) - request.addfinalizer(db.teardown) - return db.setup(uuid) - return create - - -# -# remote_db fixture: provides an empty database for a given user in a per -# function scope. -# - -class UserDatabase(object): - - def __init__(self, url, uuid): - self._remote_db_url = urljoin(url, 'user-%s' % uuid) - - def setup(self): - return CouchDatabase.open_database( - url=self._remote_db_url, create=True, replica_uid=None) - - def teardown(self): - requests.delete(self._remote_db_url) - - -@pytest.fixture() -def remote_db(request): - couch_url = request.config.option.couch_url - - def create(uuid): - db = UserDatabase(couch_url, uuid) - request.addfinalizer(db.teardown) - return db.setup() - return create - - -def get_pid(pidfile): - if not os.path.isfile(pidfile): - return 0 - try: - with open(pidfile) as f: - return int(f.read()) - except IOError: - return 0 - - -# -# soledad_server fixture: provides a running soledad server in a per module -# context (same soledad server for all tests in this module). -# - -class SoledadServer(object): - - def __init__(self, tmpdir_factory, couch_url): - tmpdir = tmpdir_factory.mktemp('soledad-server') - self._pidfile = os.path.join(tmpdir.strpath, 'soledad-server.pid') - self._logfile = os.path.join(tmpdir.strpath, 'soledad-server.log') - self._couch_url = couch_url - - def start(self): - self._create_conf_file() - # start the server - call([ - 'twistd', - '--logfile=%s' % self._logfile, - '--pidfile=%s' % self._pidfile, - 'web', - '--wsgi=leap.soledad.server.application', - '--port=2424' - ]) - - def _create_conf_file(self): - if not os.access('/etc', os.W_OK): - return - if not os.path.isdir('/etc/soledad'): - os.mkdir('/etc/soledad') - with open('/etc/soledad/soledad-server.conf', 'w') as f: - content = '[soledad-server]\ncouch_url = %s' % self._couch_url - f.write(content) - - def stop(self): - pid = get_pid(self._pidfile) - os.kill(pid, signal.SIGKILL) - - -@pytest.fixture(scope='module') -def soledad_server(tmpdir_factory, request): - couch_url = request.config.option.couch_url - server = SoledadServer(tmpdir_factory, couch_url) - server.start() - request.addfinalizer(server.stop) - return server - - @pytest.fixture() def txbenchmark(benchmark): def blockOnThread(*args, **kwargs): @@ -213,33 +55,3 @@ def txbenchmark_with_setup(benchmark): rounds=4, warmup_rounds=1) return threads.deferToThread(bench) return blockOnThreadWithSetup - - -# -# soledad_client fixture: provides a clean soledad client for a test function. -# - -@pytest.fixture() -def soledad_client(tmpdir, soledad_server, remote_db, soledad_dbs, request): - passphrase = DEFAULT_PASSPHRASE - server_url = DEFAULT_URL - token = DEFAULT_TOKEN - default_uuid = uuid4().hex - remote_db(default_uuid) - soledad_dbs(default_uuid) - - # get a soledad instance - def create(): - secrets_path = os.path.join(tmpdir.strpath, '%s.secret' % uuid4().hex) - local_db_path = os.path.join(tmpdir.strpath, '%s.db' % uuid4().hex) - soledad_client = Soledad( - default_uuid, - unicode(passphrase), - secrets_path=secrets_path, - local_db_path=local_db_path, - server_url=server_url, - cert_file=None, - auth_token=token) - request.addfinalizer(soledad_client.close) - return soledad_client - return create |