Produits Entr'ouvert » OLAP / Business Intelligence pour Publik

#!/bin/sh -xue

test -d wcs || git clone http://git.entrouvert.org/wcs.git

(cd wcs && git pull)

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

import sys

import subprocess

import time

import os

import shutil

import random

import socket

from contextlib import closing

from collections import namedtuple

import psycopg2

import pytest

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)

@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='field_string'),

    fields.ItemField(id='2', label='2nd field', type='item',

                     items=['foo', 'bar', 'baz'], varname='field_item'),

]

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'

    elif i%4 == 1:

        formdata.data['2'] = 'bar'

        formdata.data['2_display'] = 'bar'

    else:

        formdata.data['2'] = 'baz'

        formdata.data['2_display'] = 'baz'

    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(scope='session')

def wcs(tmp_path_factory):

    '''Session scoped wcs fixture, so read-only.'''

    if 'WCSCTL' not in os.environ or not os.path.exists(os.environ['WCSCTL']):

        pytest.skip('WCSCTL not defined in environment')

    WCSCTL = os.environ.get('WCSCTL')

    WCS_DIR = tmp_path_factory.mktemp('wcs')

    HOSTNAME = '127.0.0.1'

    PORT = 8899

    ADDRESS = '0.0.0.0'

    WCS_PID = None

    def run_wcs_script(script, hostname):

        '''Run python script inside w.c.s. environment'''

        script_path = WCS_DIR / (script + '.py')

        with script_path.open('w') as fd:

            fd.write(WCS_SCRIPTS[script])

        subprocess.check_call(

            [WCSCTL, 'runscript', '--app-dir', str(WCS_DIR), '--vhost', hostname,

             str(script_path)])

    tenant_dir = WCS_DIR / HOSTNAME

    tenant_dir.mkdir()

    run_wcs_script('setup-auth', HOSTNAME)

    run_wcs_script('create-user', HOSTNAME)

    run_wcs_script('create-data', HOSTNAME)

    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, HOSTNAME))

    # launch a Django worker for running w.c.s.

    WCS_PID = os.fork()

    if not WCS_PID:

        os.chdir(os.path.dirname(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/' % (HOSTNAME, PORT), appdir=WCS_DIR, pid=WCS_PID)

    os.kill(WCS_PID, 9)

    shutil.rmtree(str(WCS_DIR))

import subprocess

def test_wcs_fixture(wcs, postgres_db, tmpdir, caplog):

    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 = public

'''.format(wcs=wcs, model_dir=model_dir, dsn=postgres_db.dsn))

    from wcs_olap import cmd

    import sys

    sys.argv = ['', '--no-log-errors', str(config_ini)]

    cmd.main2()

    expected_schema = [

        ('agent', 'id'),

        ('agent', 'label'),

        ('category', 'id'),

        ('category', 'label'),

        ('channel', 'id'),

        ('channel', 'label'),

        ('evolution', 'id'),

        ('evolution', 'generic_status_id'),

        ('evolution', 'formdata_id'),

        ('evolution', 'time'),

        ('evolution', 'date'),

        ('evolution', 'hour_id'),

        ('evolution_demande', 'id'),

        ('evolution_demande', 'status_id'),

        ('evolution_demande', 'formdata_id'),

        ('evolution_demande', 'time'),

        ('evolution_demande', 'date'),

        ('evolution_demande', 'hour_id'),

        ('formdata', 'id'),

        ('formdata', 'formdef_id'),

        ('formdata', 'receipt_time'),

        ('formdata', 'hour_id'),

        ('formdata', 'channel_id'),

        ('formdata', 'backoffice'),

        ('formdata', 'generic_status_id'),

        ('formdata', 'endpoint_delay'),

        ('formdata', 'first_agent_id'),

        ('formdata', 'geolocation_base'),

        ('formdata', 'json_data'),

        ('formdata_demande', 'id'),

        ('formdata_demande', 'formdef_id'),

        ('formdata_demande', 'receipt_time'),

        ('formdata_demande', 'hour_id'),

        ('formdata_demande', 'channel_id'),

        ('formdata_demande', 'backoffice'),

        ('formdata_demande', 'generic_status_id'),

        ('formdata_demande', 'endpoint_delay'),

        ('formdata_demande', 'first_agent_id'),

        ('formdata_demande', 'geolocation_base'),

        ('formdata_demande', 'json_data'),

        ('formdata_demande', 'status_id'),

        ('formdata_demande', 'field_field_item'),

        ('formdata_demande', 'function__receiver'),

        ('formdata_demande_field_field_item', 'id'),

        ('formdata_demande_field_field_item', 'label'),

        ('formdef', 'id'),

        ('formdef', 'category_id'),

        ('formdef', 'label'),

        ('hour', 'id'),

        ('hour', 'label'),

        ('role', 'id'),

        ('role', 'label'),

        ('status', 'id'),

        ('status', 'label'),

        ('status_demande', 'id'),

        ('status_demande', 'label')

    ]

    with postgres_db.conn() as conn:

        with conn.cursor() as c:

            c.execute('SELECT table_name, column_name '

                      'FROM information_schema.columns '

                      'WHERE table_schema = \'public\' ORDER BY table_name, ordinal_position')

            assert list(c.fetchall()) == expected_schema

	WCSCTL=wcs/wcsctl.py

	quixote<3.0

	psycopg2-binary

	vobject

	gadjo

	django>=1.11,<1.12

	./get_wcs.sh

	py.test {env:COVERAGE:} {posargs:--random-group tests}

    parser.add_argument('--no-log-errors', dest='no_log_errors',

                        action='store_true', default=False)

                if args.no_log_errors:

                    raise