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

        # 47: store LoggedErrors in SQL

        # 48: remove acked attribute from LoggedError

        do_loggederrors_table()

    if sql_level < 50:

        # 49: store Role in SQL

        # 50: switch role uuid column to varchar

        do_role_table()

        migrate_legacy_roles()

    cur.execute('''UPDATE wcs_meta SET value = %s WHERE key = %s''', (str(SQL_LEVEL[0]), 'sql_level'))

    conn.commit()

    cur.close()

@guard_postgres