Produits Entr'ouvert » OLAP / Business Intelligence pour Publik

import utils

    fields.ItemField(id='4', label='4rth field', type='item', varname='item_open'),

        formdata.data['4'] = 'open_one'

        formdata.data['4_display'] = 'open_one'

        formdata.data['4'] = 'open_two'

        formdata.data['4_display'] = 'open_two'

        formdata.data['4'] = 'open_three'

        formdata.data['4_display'] = 'open_three'

@pytest.fixture

def wcs_dir(tmp_path_factory):

    return tmp_path_factory.mktemp('wcs')

@pytest.fixture

def wcs(tmp_path_factory, wcs_dir):

    tenant_dir = wcs_dir / utils.HOSTNAME

    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 (wcs_dir / 'wcs.cfg').open('w') as fd:

app_dir = %s\n''' % wcs_dir)

    with (wcs_dir / 'local_settings.py').open('w') as fd:

''' % (wcs_dir, utils.HOSTNAME))

        os.chdir(os.path.dirname(utils.WCSCTL))

        os.environ['WCS_SETTINGS_FILE'] = str(wcs_dir / 'local_settings.py')

    yield Wcs(url='http://%s:%s/' % (utils.HOSTNAME, PORT), appdir=wcs_dir, pid=WCS_PID)

    shutil.rmtree(str(wcs_dir))

            },

            {

               "filter" : true,

               "join" : [

                  "item_open"

               ],

               "label" : "4rth field",

               "name" : "item_open",

               "type" : "integer",

               "value" : "\"item_open\".id",

               "value_label" : "\"item_open\".label"

            },

            {

               "detail" : "id",

               "master" : "field_item_open",

               "name" : "item_open",

               "table" : "formdata_demande_field_item_open"

import utils

        ('formdata_demande', 'field_item_open'),

        ('formdata_demande_field_item_open', 'id'),

        ('formdata_demande_field_item_open', 'label'),

def test_dimension_stability(wcs, wcs_dir, postgres_db, tmpdir, olap_cmd, caplog):

    olap_cmd()

    with postgres_db.conn() as conn:

        with conn.cursor() as c:

            c.execute('SET search_path TO \'olap\'')

            c.execute('SELECT * FROM formdata_demande_field_item ORDER BY id')

            refs = c.fetchall()

            assert len(refs) == 3

            c.execute('SELECT * FROM formdata_demande_field_item_open ORDER BY id')

            open_refs = c.fetchall()

            assert len(open_refs) == 3

    # Change an item of the field

    script = u"""

import datetime

import random

from quixote import get_publisher

from wcs.formdef import FormDef

formdef = FormDef.get_by_urlname('demande')

for field in formdef.fields:

    if field.label == '2nd field':

        ref_field = field

        break

ref_field.items = ['foo', 'bar', 'bazouka']

formdef.store()

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

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 1'}

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

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

formdata.data['4'] = 'open_new_value'

formdata.data['4_display'] = 'open_new_value'

formdata.jump_status('new')

formdata.store()

"""

    utils.run_wcs_script(wcs_dir, script, 'toto')

    olap_cmd()

    # We expect to find in the new dimension table

    # the same  records as before (including the one of the item that disappeared)

    # plus the new item

    with postgres_db.conn() as conn:

        with conn.cursor() as c:

            c.execute('SET search_path TO \'olap\'')

            c.execute('SELECT * FROM formdata_demande_field_item ORDER BY id')

            new_refs = c.fetchall()

            assert len(new_refs) == 4

            for ref in refs:

                assert ref in new_refs

            assert new_refs[-1][1] == 'bazouka'

            bazouka_id = new_refs[-1][0]

            c.execute('SELECT * FROM formdata_demande_field_item_open ORDER BY id')

            new_open_refs = c.fetchall()

            assert len(new_open_refs) == 4

            for ref in open_refs:

                assert ref in new_open_refs

            assert new_open_refs[-1][1] == 'open_new_value'

            open_new_id = new_open_refs[-1][0]

            c.execute('''SELECT field_item, field_item_open

            FROM formdata_demande ORDER BY id''')

            formdata = c.fetchone()

            assert formdata[0] == bazouka_id

            assert formdata[1] == open_new_id

import os

import subprocess

HOSTNAME = '127.0.0.1'

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

def run_wcs_script(wcs_dir, script, script_name):

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

    script_path = wcs_dir / (script_name + '.py')

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

        fd.write(script)

    subprocess.check_call(

        [WCSCTL, 'runscript', '--app-dir', str(wcs_dir), '--vhost', HOSTNAME,

         str(script_path)])

    channels = [

        [1, 'web', u'web'],

        [2, 'mail', u'courrier'],

        [3, 'phone', u'téléphone'],

        [4, 'counter', u'guichet'],

        [5, 'backoffice', u'backoffice'],

        [6, 'email', u'email'],

        [7, 'fax', u'fax'],

    ]

    channel_to_id = dict((c[1], c[0]) for c in channels)

    id_to_channel = dict((c[0], c[1]) for c in channels)

    status = [

        [1, 'Nouveau'],

        [2, 'En cours'],

        [3, 'Terminé'],

    ]

    status_to_id = dict((c[1], c[0]) for c in channels)

    id_to_status = dict((c[0], c[1]) for c in channels)

        psycopg2.extensions.register_type(psycopg2.extensions.UNICODE, self.cur)

    def prev_table_exists(self, name):

        query = """SELECT EXISTS (SELECT 1 FROM information_schema.tables

        WHERE  table_schema = '{schema}' AND table_name = '%s')""" % name

        self.ex(query)

        return self.cur.fetchone()[0]

    def create_labeled_table_serial(self, name, comment):

        self.create_table(

            name, [['id', 'serial primary key'], ['label', 'varchar']], comment=comment)

        if self.prev_table_exists(name):

            # Insert data from previous table

            self.ex(

                'INSERT INTO {schema_temp}.%(name)s select * FROM {schema}.%(name)s'

                % {'name': name}

            )

            # Update sequence

            self.ex("""SELECT setval(pg_get_serial_sequence('%(name)s', 'id'),

            (SELECT MAX(id) FROM %(name)s))""" % {'name': name})

    def create_labeled_table(self, name, labels, comment=None):

        self.create_table(

            name,

            [

                ['id', 'smallint primary key'],

                ['label', 'varchar']

            ], comment=comment)

        if self.prev_table_exists(name):

            # Insert data from previous table

            self.ex(

                'INSERT INTO {schema_temp}.%(name)s select * FROM {schema}.%(name)s'

                % {'name': name}

            )

            # Find what is missing

            to_insert = []

            for _id, _label in labels:

                self.ex("SELECT * FROM %s WHERE label = '%s'" % (name, _label))

                if self.cur.fetchone() is None:

                    to_insert.append(_label)

            labels = None

            if to_insert:

                self.ex('SELECT MAX(id) FROM %s' % name)

                next_id = self.cur.fetchone()[0] + 1

                ids = range(next_id, next_id + len(to_insert))

                labels = zip(ids, to_insert)

        if labels:

            values = ', '.join(self.cur.mogrify('(%s, %s)', [_id, _label]) for _id, _label in labels)

            self.ex('INSERT INTO %s (id, label) VALUES %s' % (str(name), values))

        res = {}

        self.ex("SELECT id, label FROM %s" % str(name))

        for id_, label in self.cur.fetchall():

            res[label] = id_

        return res

        tmp_role_map = self.create_labeled_table('{role_table}', roles.items(), comment=u'role')

        self.role_mapping = dict(

            (role.id, tmp_role_map[role.name]) for role in self.roles)

        tmp_cat_map = self.create_labeled_table(

            '{category_table}', enumerate(c.name for c in self.categories), comment=u'catégorie')

        self.categories_mapping = dict((c.id, tmp_cat_map[c.name]) for c in self.categories)

        self.create_labeled_table_serial('{agent_table}', comment=u'agents')

        self.ex("SELECT id FROM {agent_table} WHERE label = '%s'" % name)

        res = self.cur.fetchone()

        if res:

            return res[0]

        tmp_status_map = self.olap_feeder.create_labeled_table(

            self.status_table_name, enumerate([s.name for s in statuses]),

            comment=u'statuts du formulaire « %s »' % self.formdef.schema.name)

        self.status_mapping = dict((s.id, tmp_status_map[s.name]) for s in statuses)

                    self.items_mappings[field.varname] = self.create_labeled_table(

                        table_name, enumerate(field.items), comment=comment)

                    tmp_options_map = self.create_labeled_table(

                        table_name, [(i, o['label']) for i, o in options], comment=comment)

                    self.items_mappings[field.varname] = dict(

                        (o['value'], tmp_options_map[o['value']]) for i, o in options)

                    self.create_labeled_table_serial(table_name, comment=comment)

        self.ex("SELECT id FROM {item_table} WHERE label = '%s'" % value,

                ctx={'item_table': table_name})

        res = self.cur.fetchone()

        if res:

            return res[0]