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

cats = []

for i in range(1, 10):

   cat = Category()

   cat.name = 'Test %d' % i

   cat.description = 'Hello world'

   cat.store()

   cats.append(cat)

formdef = FormDef()

formdef.name = 'form title'

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()

# another formdef in same category

formdef = FormDef()

formdef.name = 'a second form title'

formdef.category_id = cat.id

formdef.fields = []

formdef.store()

# a formdef in another category

formdef = FormDef()

formdef.name = 'third form title'

formdef.category_id = cats[2].id

formdef.fields = []

formdef.enable_tracking_codes = True

formdef.store()

# a draft of that formdef

formdata = formdef.data_class()()

formdata.data = {}

formdata.page_no = 1

formdata.status = 'draft'

formdata.receipt_time = datetime.datetime(2015, 1, 1).timetuple()

formdata.user_id = user.id

formdata.store()

if '127.0.0.2' in get_publisher().get_frontoffice_url():

    # create a tracking code on second website only

    code = get_publisher().tracking_code_class()

    code.id = 'CNPHNTFB'

    code.formdata = formdata

# a private formdef

role = Role(name='Blah')

role.store()

formdef = FormDef()

formdef.name = 'a private form'

formdef.category_id = cats[2].id

formdef.roles = [role.id]

formdef.backoffice_submission_roles = [role.id]

formdef.fields = []

formdef.store()

user2 = get_publisher().user_class()  # agent

user2.name = 'foo2 bar2'

user2.email = 'foo2@example.net'

user2.roles = [role.id]

user2.store()

""",

}

@pytest.mark.skipif(

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

    reason='WCSCTL not defined in environment')

@pytest.fixture(scope='session')

def wcs(tmp_path_factory):

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

    WCS_DIR = tmp_path_factory.mktemp('wcs')

    WCS_PID = None

    def run_wcs_script(script, hostname):

        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)])

    HOSTNAME = '127.0.0.1'

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

    PORT = 8899

    ADDRESS = '0.0.0.0'

    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 django 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'

    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):

    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}

[loggers]

keys = root

[formatters]

keys = console

[formatter_console]

format = %(asctime)s %(levelname)s %(message)s

[handlers]

keys = console

[handler_console]

class = StreamHandler

level = NOTSET

args = (sys.stderr,)

formatter = console

[logger_root]

level = DEBUG

handlers = console

[{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()

    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')

            for row in c.fetchall():

                print row

            import pdb

            pdb.set_trace()

	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