Produits Entr'ouvert » w.c.s.

import time

import psycopg2

import pytest

from quixote import cleanup

import wcs.qommon.storage as st

from wcs import fields, sql

from wcs.formdata import Evolution

from wcs.formdef import FormDef

from wcs.qommon import force_str

from wcs.wf.jump import JumpWorkflowStatusItem

from wcs.wf.register_comment import RegisterCommenterWorkflowStatusItem

from wcs.workflows import CommentableWorkflowStatusItem, Workflow, WorkflowCriticalityLevel

from .utilities import clean_temporary_pub, create_temporary_pub

def setup_module(module):

    global pub, formdef

    cleanup()

    pub = create_temporary_pub(sql_mode=True)

    formdef = FormDef()

    formdef.name = 'tests'

    formdef.fields = [

        fields.StringField(id='0', label='string'),

        fields.EmailField(id='1', label='email'),

        fields.TextField(id='2', label='text'),

        fields.BoolField(id='3', label='bool'),

        fields.ItemField(id='4', label='item', items=('apple', 'pear', 'peach', 'apricot')),

        fields.DateField(id='5', label='date'),

        fields.ItemsField(id='6', label='items', items=('apple', 'pear', 'peach', 'apricot')),

    ]

    formdef.store()

def teardown_module(module):

    clean_temporary_pub()

def test_sql_field_items():

    check_sql_field('6', ['apricot'], display=True)

    check_sql_field('6', ['apricot', 'pear'], display=True)

    check_sql_field('6', ['pomme', 'poire', 'pêche'], display=True)

def test_sql_geoloc():

    test_formdef = FormDef()

    test_formdef.name = 'geoloc'

    test_formdef.fields = []

    test_formdef.geolocations = {'base': 'Plop'}

    test_formdef.store()

    data_class = test_formdef.data_class(mode='sql')

    formdata = data_class()

    formdata.store()

    id2 = formdata.id

    formdata = data_class()

    formdata.data = {'2': 'you would think other ideas of text would emerge'}

    formdata.store()

    id3 = formdata.id

    ids = data_class.get_ids_from_query('text')

    assert set(ids) == set([id1, id3])

    ids = data_class.get_ids_from_query('classical')

    assert set(ids) == set([id2])

def test_sql_rollback_on_error():

    data_class = formdef.data_class(mode='sql')

    data_class.wipe()

    with pytest.raises(psycopg2.Error):

        # this will raise a psycopg2.ProgrammingError as there's no FOOBAR

        # column in the table.

        data_class.get_ids_with_indexed_value('FOOBAR', '2')

                    sql_dict['parts'] = None

                cur.execute(sql_statement, sql_dict)

                evo._sql_id = cur.fetchone()[0]

        fts_strings = [str(self.id), self.get_display_id()]

        fts_strings.append(self._formdef.name)

        if self.tracking_code:

            fts_strings.append(self.tracking_code)

                continue

            value = None

            if field.key in ('string', 'text', 'email'):

            elif field.key in ('item', 'items'):

            if value:

                if isinstance(value, str):

                    fts_strings.append(value)

                elif type(value) in (tuple, list):

                    fts_strings.extend(value)

        if self._evolution:

            for evo in self._evolution:

                if evo.comment:

    cur.close()

    return result

# latest migration, number + description (description is not used

# programmaticaly but will make sure git conflicts if two migrations are

# separately added with the same number)

def migrate_global_views(conn, cur):

    cur.execute(

        '''SELECT COUNT(*) FROM information_schema.tables

                    WHERE table_schema = 'public'

                      AND table_name = %s''',

        ('wcs_all_forms',),

        migrate_views(conn, cur)

        for formdef in FormDef.select():

            formdef.data_class().rebuild_security()

    if sql_level < 23:

        # 12: (second part), store fts in existing rows

        # 21: (second part), store ascii_name of users

        # 23: (first part), use misc.simplify() over full text queries

        set_reindex('user', 'needed', conn=conn, cur=cur)

        # 17: store last_update_time in tables

        # 18: add user name to full-text search index

        # 21: (third part), add user ascii_names to full-text index

        # 23: (second part) use misc.simplify() over full text queries

        # 28: add display id and formdef name to full-text index

        # 29: add evolution parts to full-text index

        # 31: add user_label to formdata

        # 38: extract submission_agent_id to its own column

        # 41: update full text normalization

        set_reindex('formdata', 'needed', conn=conn, cur=cur)

    if sql_level < 46:

        from wcs.carddef import CardDef

        from wcs.formdef import FormDef

        # 24: add index on evolution(formdata_id)

        # 35: add indexes on formdata(receipt_time) and formdata(anonymised)

        # 36: add index on formdata(user_id)