wcs-olap/tests/conftest.py

# -*- coding: utf-8 -*-

import sys
import time
import os
import shutil
import random
import socket
from contextlib import closing
from collections import namedtuple

import psycopg2
import pytest

import utils


Wcs = namedtuple('Wcs', ['url', 'appdir', 'pid'])


class Database(object):
    def __init__(self):
        self.db_name = 'db%s' % random.getrandbits(20)
        self.dsn = 'dbname=%s' % self.db_name
        with closing(psycopg2.connect('')) as conn:
            conn.set_isolation_level(0)
            with conn.cursor() as cursor:
                cursor.execute('CREATE DATABASE %s' % self.db_name)

    def conn(self):
        return closing(psycopg2.connect(self.dsn))

    def delete(self):
        with closing(psycopg2.connect('')) as conn:
            conn.set_isolation_level(0)
            with conn.cursor() as cursor:
                cursor.execute('DROP DATABASE IF EXISTS %s' % self.db_name)

    def __repr__(self):
        return '<Postgres Database %r>' % self.db_name


@pytest.fixture
def postgres_db():
    db = Database()
    try:
        yield db
    finally:
        db.delete()


WCS_SCRIPTS = {
    'setup-auth': u"""
from quixote import get_publisher

get_publisher().cfg['identification'] = {'methods': ['password']}
get_publisher().cfg['debug'] = {'display_exceptions': 'text'}
get_publisher().write_cfg()
""",
    'create-user': u"""
from quixote import get_publisher
from qommon.ident.password_accounts import PasswordAccount

user = get_publisher().user_class()
user.name = 'foo bar'
user.email = 'foo@example.net'
user.store()
account = PasswordAccount(id='user')
account.set_password('user')
account.user_id = user.id
account.store()
""",
    'create-data': u"""
import datetime
import random
from quixote import get_publisher

from wcs.categories import Category
from wcs.formdef import FormDef
from wcs.roles import Role
from wcs import fields

cat = Category()
cat.name = 'Catégorie'
cat.description = ''
cat.store()

formdef = FormDef()
formdef.name = 'Demande'
formdef.category_id = cat.id
formdef.fields = [
    fields.StringField(id='1', label='1st field', type='string', anonymise=False, varname='string'),
    fields.ItemField(id='2', label='2nd field', type='item',
                     items=['foo', 'bar', 'baz'], varname='item'),
    fields.BoolField(id='3', label='3rd field', type='bool', varname='bool'),
    fields.ItemField(id='4', label='4rth field', type='item', varname='item_open'),
]
formdef.store()

user = get_publisher().user_class.select()[0]

for i in range(50):
    formdata = formdef.data_class()()
    formdata.just_created()
    formdata.receipt_time = datetime.datetime(2018, random.randrange(1, 13), random.randrange(1, 29)).timetuple()
    formdata.data = {'1': 'FOO BAR %d' % i}
    if i%4 == 0:
        formdata.data['2'] = 'foo'
        formdata.data['2_display'] = 'foo'
        formdata.data['4'] = 'open_one'
        formdata.data['4_display'] = 'open_one'
    elif i%4 == 1:
        formdata.data['2'] = 'bar'
        formdata.data['2_display'] = 'bar'
        formdata.data['4'] = 'open_two'
        formdata.data['4_display'] = 'open_two'
    else:
        formdata.data['2'] = 'baz'
        formdata.data['2_display'] = 'baz'
        formdata.data['4'] = "open'three"
        formdata.data['4_display'] = "open'three"

    formdata.data['3'] = bool(i % 2)
    if i%3 == 0:
        formdata.jump_status('new')
    else:
        formdata.jump_status('finished')
    if i%7 == 0:
        formdata.user_id = user.id
    formdata.store()
""",
}


@pytest.fixture
def wcs_dir(tmp_path_factory):
    return tmp_path_factory.mktemp('wcs')


@pytest.fixture
def wcs(tmp_path_factory, wcs_dir):
    '''Session scoped wcs fixture, so read-only.'''
    PORT = 8899
    ADDRESS = '0.0.0.0'
    WCS_PID = None

    tenant_dir = wcs_dir / utils.HOSTNAME
    tenant_dir.mkdir()

    utils.run_wcs_script(wcs_dir, WCS_SCRIPTS['setup-auth'], 'setup-auth')
    utils.run_wcs_script(wcs_dir, WCS_SCRIPTS['create-user'], 'create-user')
    utils.run_wcs_script(wcs_dir, WCS_SCRIPTS['create-data'], 'create-data')

    with (tenant_dir / 'site-options.cfg').open('w') as fd:
        fd.write(u'''[api-secrets]
olap = olap
''')

    with (wcs_dir / 'wcs.cfg').open('w') as fd:
        fd.write(u'''[main]
app_dir = %s\n''' % wcs_dir)

    with (wcs_dir / 'local_settings.py').open('w') as fd:
        fd.write(u'''
WCS_LEGACY_CONFIG_FILE = '%s/wcs.cfg'
THEMES_DIRECTORY = '/'
ALLOWED_HOSTS = ['%s']
''' % (wcs_dir, utils.HOSTNAME))

    # launch a Django worker for running w.c.s.
    WCS_PID = os.fork()
    if not WCS_PID:
        os.chdir(os.path.dirname(utils.WCSCTL))
        os.environ['DJANGO_SETTINGS_MODULE'] = 'wcs.settings'
        os.environ['WCS_SETTINGS_FILE'] = str(wcs_dir / 'local_settings.py')
        os.execvp('python', ['python', 'manage.py', 'runserver', '--noreload', '%s:%s' % (ADDRESS, PORT)])
        sys.exit(0)

    # verify w.c.s. is launched
    s = socket.socket()
    i = 0
    while True:
        i += 1
        try:
            s.connect((ADDRESS, PORT))
        except Exception:
            time.sleep(0.1)
        else:
            s.close()
            break
        assert i < 50, 'no connection found after 5 seconds'

    # verify w.c.s. is still running
    pid, exit_code = os.waitpid(WCS_PID, os.WNOHANG)
    if pid:
        assert False, 'w.c.s. stopped with exit-code %s' % exit_code

    yield Wcs(url='http://%s:%s/' % (utils.HOSTNAME, PORT), appdir=wcs_dir, pid=WCS_PID)
    os.kill(WCS_PID, 9)
    shutil.rmtree(str(wcs_dir))


@pytest.fixture
def olap_cmd(wcs, tmpdir, postgres_db):
    config_ini = tmpdir / 'config.ini'
    model_dir = tmpdir / 'model_dir'
    model_dir.mkdir()
    with config_ini.open('w') as fd:
        fd.write(u'''
[wcs-olap]
cubes_model_dirs = {model_dir}
pg_dsn = {dsn}

[{wcs.url}]
orig = olap
key = olap
schema = olap
'''.format(wcs=wcs, model_dir=model_dir, dsn=postgres_db.dsn))

    from wcs_olap import cmd
    import sys

    def f(no_log_errors=True):
        old_argv = sys.argv
        try:
            sys.argv = ['', str(config_ini)]
            if no_log_errors:
                sys.argv.insert(1, '--no-log-errors')
            cmd.main2()
        finally:
            sys.argv = old_argv
    f.model_dir = model_dir
    return f