path: root/testing/tests/conftest.py

import json
import os
import pytest
import requests
import signal
import socket
import sys
import time

from hashlib import sha512
from subprocess import check_call
from six.moves.urllib.parse import urljoin
from uuid import uuid4

from leap.soledad.common.couch import CouchDatabase
from leap.soledad.client import Soledad


# mark tests that depend on couchdb
def pytest_collection_modifyitems(items):
    marker = getattr(pytest.mark, 'needs_couch')
    for item in items:
        if 'soledad/testing/tests/couch/' in item.module.__file__:
            item.add_marker(marker)


#
# 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 option is 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!")


@pytest.fixture
def couch_url(request):
    url = request.config.getoption('--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):
        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.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')
        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):
    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(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.
        db_file = '%s.db' % default_uuid
        if force_fresh_db:
            prefix = uuid4().hex
            db_file = prefix + '-' + db_file
        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=None,
            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