summaryrefslogtreecommitdiff
path: root/tests/conftest.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/conftest.py')
-rw-r--r--tests/conftest.py398
1 files changed, 398 insertions, 0 deletions
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 00000000..d3a39289
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,398 @@
+import glob
+import base64
+import json
+import os
+import pytest
+import re
+import random
+import requests
+import signal
+import socket
+import subprocess
+import sys
+import time
+import urlparse
+
+from hashlib import sha512
+from six.moves.urllib.parse import urljoin
+from uuid import uuid4
+
+from leap.soledad.common.couch import CouchDatabase
+from leap.soledad.client import Soledad
+
+
+def _select_subdir(subdir, blacklist, items):
+
+ # allow blacklisted subdir if explicited in command line
+ if subdir and subdir in blacklist:
+ blacklist.remove(subdir)
+
+ # determine blacklisted subdirs
+ dirname = os.path.dirname(__file__)
+ blacklisted_subdirs = map(lambda s: os.path.join(dirname, s), blacklist)
+
+ # determine base path for selected tests
+ path = dirname
+ if subdir:
+ path = os.path.join(dirname, subdir)
+
+ # remove tests from blacklisted subdirs
+ selected = []
+ deselected = []
+ for item in items:
+ filename = item.module.__file__
+ blacklisted = any(
+ map(lambda s: filename.startswith(s), blacklisted_subdirs))
+ if blacklisted or not filename.startswith(path):
+ deselected.append(item)
+ else:
+ selected.append(item)
+
+ return selected, deselected
+
+
+def pytest_collection_modifyitems(items, config):
+
+ # mark tests that depend on couchdb
+ marker = getattr(pytest.mark, 'needs_couch')
+ for item in items:
+ if 'soledad/testing/tests/couch/' in item.module.__file__:
+ item.add_marker(marker)
+
+ # select/deselect tests based on a blacklist and the subdir option given in
+ # command line
+ blacklist = ['benchmarks', 'responsiveness']
+ subdir = config.getoption('subdir')
+ selected, deselected = _select_subdir(subdir, blacklist, items)
+ config.hook.pytest_deselected(items=deselected)
+ items[:] = selected
+
+
+#
+# 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):
+ parser.addoption(
+ "--couch-url", type="string", default="http://127.0.0.1:5984",
+ help="the url for the couch server to be used during tests")
+
+ # the following options are only used in benchmarks, but has to be defined
+ # here due to how pytest discovers plugins during startup.
+ parser.addoption(
+ "--watch-memory", default=False, action="store_true",
+ help="whether to monitor memory percentages during test run. "
+ "**Warning**: enabling this will impact the time taken and the "
+ "CPU used by the benchmarked code, so use with caution!")
+
+ parser.addoption(
+ "--soledad-server-url", type="string", default=None,
+ help="Soledad Server URL. A local server will be started if and only "
+ "if no URL is passed.")
+
+ # the following option is only used in responsiveness tests, but has to be
+ # defined here due to how pytest discovers plugins during startup.
+ parser.addoption(
+ "--elasticsearch-url", type="string", default=None,
+ help="the url for posting responsiveness results to elasticsearch")
+
+ parser.addoption(
+ "--subdir", type="string", default=None,
+ help="select only tests from a certain subdirectory of ./tests/")
+
+
+def _request(method, url, data=None, do=True):
+ if do:
+ method = getattr(requests, method)
+ method(url, data=data)
+ else:
+ cmd = 'curl --netrc -X %s %s' % (method.upper(), url)
+ if data:
+ cmd += ' -d "%s"' % json.dumps(data)
+ print cmd
+
+
+@pytest.fixture
+def couch_url(request):
+ url = request.config.option.couch_url
+ request.cls.couch_url = url
+
+
+@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, create=True):
+ self._remote_db_url = urljoin(url, 'user-%s' % uuid)
+ self._create = create
+
+ def setup(self):
+ if self._create:
+ return CouchDatabase.open_database(
+ url=self._remote_db_url, create=True, replica_uid=None)
+ else:
+ _request('put', self._remote_db_url, do=False)
+
+ def teardown(self):
+ _request('delete', self._remote_db_url, do=self._create)
+
+
+@pytest.fixture()
+def remote_db(request):
+ couch_url = request.config.option.couch_url
+
+ def create(uuid, create=True):
+ db = UserDatabase(couch_url, uuid, create=create)
+ 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.tmpdir = tmpdir
+ 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
+ executable = 'twistd'
+ if 'VIRTUAL_ENV' not in os.environ:
+ executable = os.path.join(
+ os.path.dirname(os.environ['_']), 'twistd')
+ subprocess.check_call([
+ executable,
+ '--logfile=%s' % self._logfile,
+ '--pidfile=%s' % self._pidfile,
+ 'web',
+ '--class=leap.soledad.server.entrypoints.SoledadEntrypoint',
+ '--port=tcp:2424'
+ ])
+
+ def _create_conf_file(self):
+
+ # come up with name of the configuration file
+ fname = '/etc/soledad/soledad-server.conf'
+ if not os.access('/etc', os.W_OK):
+ fname = os.path.join(self.tmpdir.strpath, 'soledad-server.conf')
+
+ # create the configuration file
+ dirname = os.path.dirname(fname)
+ if not os.path.isdir(dirname):
+ os.mkdir(dirname)
+ with open(fname, 'w') as f:
+ blobs_path = os.path.join(str(self.tmpdir), 'blobs')
+ content = '''[soledad-server]
+couch_url = %s
+blobs = true
+blobs_path = %s''' % (self._couch_url, blobs_path)
+ f.write(content)
+
+ # update the environment to use that file
+ os.environ.update({'SOLEDAD_SERVER_CONFIG_FILE': fname})
+
+ def stop(self):
+ pid = get_pid(self._pidfile)
+ os.kill(pid, signal.SIGTERM)
+
+
+@pytest.fixture(scope='module')
+def soledad_server(tmpdir_factory, request):
+
+ # avoid starting a server if the url is remote
+ soledad_url = request.config.option.soledad_server_url
+ if soledad_url is not None:
+ return None
+
+ # start a soledad server
+ 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, create=True):
+ self._token_db_url = urljoin(url, _token_dbname())
+ self._shared_db_url = urljoin(url, 'shared')
+ self._create = create
+
+ def setup(self, uuid):
+ self._create_dbs()
+ self._add_token(uuid)
+
+ def _create_dbs(self):
+ _request('put', self._token_db_url, do=self._create)
+ _request('put', self._shared_db_url, do=self._create)
+
+ def _add_token(self, uuid):
+ token = sha512(DEFAULT_TOKEN).hexdigest()
+ content = {'type': 'Token', 'user_id': uuid}
+ _request('put', self._token_db_url + '/' + token,
+ data=json.dumps(content), do=self._create)
+
+ def teardown(self):
+ _request('delete', self._token_db_url, do=self._create)
+ _request('delete', self._shared_db_url, do=self._create)
+
+
+@pytest.fixture()
+def soledad_dbs(request):
+ couch_url = request.config.option.couch_url
+
+ def create(uuid, create=True):
+ db = SoledadDatabases(couch_url, create=create)
+ request.addfinalizer(db.teardown)
+ return db.setup(uuid)
+ return create
+
+
+#
+# soledad_client fixture: provides a clean soledad client for a test function.
+#
+
+def _get_certfile(url, tmpdir):
+
+ # download the certificate
+ parsed = urlparse.urlsplit(url)
+ netloc = re.sub('^[^\.]+\.', '', parsed.netloc)
+ host, _ = netloc.split(':')
+ response = requests.get('https://%s/ca.crt' % host, verify=False)
+
+ # store it in a temporary file
+ cert_file = os.path.join(tmpdir.strpath, 'cert.pem')
+ with open(cert_file, 'w') as f:
+ f.write(response.text)
+
+ return cert_file
+
+
+@pytest.fixture()
+def soledad_client(tmpdir, soledad_server, remote_db, soledad_dbs, request):
+ passphrase = DEFAULT_PASSPHRASE
+ token = DEFAULT_TOKEN
+
+ # default values for local server
+ server_url = DEFAULT_URL
+ default_uuid = uuid4().hex
+ create = True
+ cert_file = None
+
+ # use values for remote server if server url is passed
+ url_arg = request.config.option.soledad_server_url
+ if url_arg:
+ server_url = url_arg
+ default_uuid = 'test-user'
+ create = False
+ cert_file = _get_certfile(server_url, tmpdir)
+
+ remote_db(default_uuid, create=create)
+ soledad_dbs(default_uuid, create=create)
+
+ # get a soledad instance
+ def create(force_fresh_db=False):
+ secrets_file = '%s.secret' % default_uuid
+ secrets_path = os.path.join(tmpdir.strpath, secrets_file)
+
+ # in some tests we might want to use the same user and remote database
+ # but with a clean/empty local database (i.e. download benchmarks), so
+ # here we provide a way to do that.
+ idx = 1
+ if force_fresh_db:
+ # find the next index for this user
+ idx = len(glob.glob('%s/*-*.db' % tmpdir.strpath)) + 1
+ db_file = '%s-%d.db' % (default_uuid, idx)
+ local_db_path = os.path.join(tmpdir.strpath, db_file)
+
+ soledad_client = Soledad(
+ default_uuid,
+ unicode(passphrase),
+ secrets_path=secrets_path,
+ local_db_path=local_db_path,
+ server_url=server_url,
+ cert_file=cert_file,
+ auth_token=token,
+ with_blobs=True)
+ request.addfinalizer(soledad_client.close)
+ return soledad_client
+ return create
+
+
+#
+# pytest-benchmark customizations
+#
+
+# avoid hooking if this is not a benchmarking environment
+if 'pytest_benchmark' in sys.modules:
+
+ def pytest_benchmark_update_machine_info(config, machine_info):
+ """
+ Add the host's hostname information to machine_info.
+
+ Get the value from the HOST_HOSTNAME environment variable if it is set,
+ or from the actual system's hostname otherwise.
+ """
+ hostname = os.environ.get('HOST_HOSTNAME', socket.gethostname())
+ machine_info['host'] = hostname
+
+
+#
+# benchmark/responsiveness fixtures
+#
+
+@pytest.fixture()
+def payload():
+ def generate(size):
+ random.seed(1337) # same seed to avoid different bench results
+ payload_bytes = bytearray(random.getrandbits(8) for _ in xrange(size))
+ # encode as base64 to avoid ascii encode/decode errors
+ return base64.b64encode(payload_bytes)[:size] # remove b64 overhead
+ return generate