entrouvert/wcs
entrouvert
/
wcs
14
1
Fork 0
Code Issues Pull Requests 39 Releases Activity
wcs/tests/test_sql.py

1858 lines
57 KiB
Python
Raw Permalink Blame History

# -*- coding: utf-8 -*-
from __future__ import print_function
import datetime
import os
import random
import shutil
import string
import sys
import time
from django.core.management import call_command
from quixote import cleanup
from wcs.qommon import force_str
from wcs import formdef, publisher, fields
from wcs.formdef import FormDef
from wcs.formdata import Evolution
from wcs.roles import Role
from wcs.workflows import Workflow, CommentableWorkflowStatusItem, WorkflowCriticalityLevel
from wcs.wf.jump import JumpWorkflowStatusItem
from wcs.wf.register_comment import RegisterCommenterWorkflowStatusItem
from wcs import sql
import wcs.qommon.storage as st
from utilities import create_temporary_pub, clean_temporary_pub
import pytest
postgresql = pytest.mark.postgresql
try:
import psycopg2
except ImportError:
pass
def setup_module(module):
global pub, formdef
cleanup()
pub = create_temporary_pub(sql_mode=True)
formdef = 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()
@postgresql
def test_sql_table_name_invalid_chars():
test_formdef = FormDef()
test_formdef.name = 'test-some|char;are better/ignored'
test_formdef.fields = []
test_formdef.store()
assert test_formdef.table_name is not None
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
@postgresql
def test_sql_data_class():
data_class = formdef.data_class(mode='sql')
@postgresql
def test_sql_count():
data_class = formdef.data_class(mode='sql')
assert data_class.count() == 0
@postgresql
def test_sql_store():
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.status = 'wf-0'
formdata.user_id = '5'
formdata.store()
assert formdata.id
@postgresql
def test_sql_get():
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.status = 'wf-0'
formdata.user_id = '5'
formdata.store()
id = formdata.id
formdata = data_class.get(id)
assert formdata.user_id == '5'
assert formdata.status == 'wf-0'
@postgresql
def test_sql_store_channel():
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.status = 'wf-0'
formdata.user_id = '5'
formdata.submission_channel = 'mail'
formdata.store()
assert data_class.get(formdata.id).submission_channel == 'mail'
formdata.submission_channel = None
formdata.store()
assert data_class.get(formdata.id).submission_channel is None
@postgresql
def test_sql_get_missing():
data_class = formdef.data_class(mode='sql')
with pytest.raises(KeyError):
data_class.get(123456)
@postgresql
def test_sql_get_missing_ignore_errors():
data_class = formdef.data_class(mode='sql')
assert data_class.get(123456, ignore_errors=True) is None
def check_sql_field(no, value):
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.data = {no: value}
formdata.store()
id = formdata.id
formdata = data_class.get(id)
assert formdata.data.get(no) == value
@postgresql
def test_sql_field_string():
check_sql_field('0', 'hello world')
check_sql_field('0', 'élo world')
check_sql_field('0', None)
@postgresql
def test_sql_field_email():
check_sql_field('1', 'fred@example.com')
@postgresql
def test_sql_field_text():
check_sql_field('2', 'long text')
check_sql_field('2', 'long tèxt')
@postgresql
def test_sql_field_bool():
check_sql_field('3', False)
check_sql_field('3', True)
@postgresql
def test_sql_field_item():
check_sql_field('4', 'apricot')
@postgresql
def test_sql_field_date():
check_sql_field('5', datetime.date.today().timetuple())
@postgresql
def test_sql_field_items():
check_sql_field('6', ['apricot'])
check_sql_field('6', ['apricot', 'pear'])
check_sql_field('6', ['pomme', 'poire', 'pêche'])
@postgresql
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.data = {}
formdata.store() # NULL geolocation
formdata2 = data_class.get(formdata.id)
assert not formdata2.geolocations
formdata.geolocations = {'base': {'lat': 12, 'lon': 21}}
formdata.store()
formdata2 = data_class.get(formdata.id)
assert formdata2.geolocations == formdata.geolocations
formdata.geolocations = {}
formdata.store()
formdata2 = data_class.get(formdata.id)
assert formdata2.geolocations == formdata.geolocations
@postgresql
def test_sql_multi_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.data = {}
formdata.geolocations = {'base': {'lat': 12, 'lon': 21}}
formdata.store()
formdata2 = data_class.get(formdata.id)
assert formdata2.geolocations == formdata.geolocations
test_formdef.geolocations = {'base': 'Plop', '2nd': 'XXX'}
test_formdef.store()
formdata.geolocations = {'base': {'lat': 12, 'lon': 21}, '2nd': {'lat': -34, 'lon': -12}}
formdata.store()
formdata2 = data_class.get(formdata.id)
assert formdata2.geolocations == formdata.geolocations
test_formdef.geolocations = {'base': 'Plop'}
test_formdef.store()
formdata2 = data_class.get(formdata.id)
assert formdata2.geolocations == {'base': {'lat': 12, 'lon': 21}}
@postgresql
def test_sql_change():
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.data = {'0': 'test'}
formdata.store()
id = formdata.id
formdata = data_class.get(id)
assert formdata.data.get('0') == 'test'
formdata.data = {'0': 'test2'}
formdata.store()
formdata = data_class.get(id)
assert formdata.data.get('0') == 'test2'
@postgresql
def test_sql_remove():
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.data = {'0': 'test'}
formdata.store()
id = formdata.id
formdata = data_class.get(id)
assert formdata.data.get('0') == 'test'
formdata.remove_self()
with pytest.raises(KeyError):
data_class.get(id)
@postgresql
def test_sql_wipe():
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.store()
assert data_class.count() != 0
data_class.wipe()
assert data_class.count() == 0
@postgresql
def test_sql_evolution():
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.just_created()
formdata.store()
id = formdata.id
formdata = data_class.get(id)
assert len(formdata.evolution) == 1
evo = Evolution()
evo.time = time.localtime()
evo.status = formdata.status
evo.comment = 'hello world'
formdata.evolution.append(evo)
formdata.store()
formdata = data_class.get(id)
assert len(formdata.evolution) == 2
assert formdata.evolution[-1].comment == 'hello world'
@postgresql
def test_sql_evolution_change():
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.just_created()
formdata.store()
id = formdata.id
formdata = data_class.get(id)
assert len(formdata.evolution) == 1
evo = Evolution()
evo.time = time.localtime()
evo.status = formdata.status
evo.comment = 'hello world'
formdata.evolution.append(evo)
formdata.store()
formdata = data_class.get(id)
assert len(formdata.evolution) == 2
assert formdata.evolution[-1].comment == 'hello world'
formdata.evolution[-1].comment = 'foobar'
formdata.store()
formdata = data_class.get(id)
assert len(formdata.evolution) == 2
assert formdata.evolution[-1].comment == 'foobar'
@postgresql
def test_sql_multiple_evolutions():
data_class = formdef.data_class(mode='sql')
for i in range(20):
formdata = data_class()
formdata.just_created()
formdata.store()
id = formdata.id
formdata = data_class.get(id)
evo = Evolution()
evo.time = time.localtime()
evo.status = formdata.status
evo.comment = 'hello world %d' % i
formdata.evolution.append(evo)
formdata.store()
values = data_class.select()
data_class.load_all_evolutions(values)
assert [x._evolution for x in values]
@postgresql
def test_sql_get_ids_with_indexed_value():
data_class = formdef.data_class(mode='sql')
data_class.wipe()
formdata = data_class()
formdata.store()
id1 = formdata.id
formdata = data_class()
formdata.user_id = '2'
formdata.store()
id2 = formdata.id
formdata = data_class()
formdata.user_id = '2'
formdata.store()
id3 = formdata.id
ids = data_class.get_ids_with_indexed_value('user_id', '2')
assert set(ids) == set([id2, id3])
@postgresql
def test_sql_get_ids_from_query():
data_class = formdef.data_class(mode='sql')
data_class.wipe()
formdata = data_class()
formdata.data = {'2': 'this is some reasonably long text'}
formdata.store()
id1 = formdata.id
formdata = data_class()
formdata.data = {'2': 'hello world is still a classical example'}
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])
@postgresql
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')
data_class.wipe()
@postgresql
def test_sql_get_ids_with_indexed_value_dict():
data_class = formdef.data_class(mode='sql')
data_class.wipe()
formdata = data_class()
formdata.store()
id1 = formdata.id
formdata = data_class()
formdata.workflow_roles = {'plop': '2'}
formdata.store()
id2 = formdata.id
formdata = data_class()
formdata.workflow_roles = {'plop': '2'}
formdata.store()
id3 = formdata.id
ids = data_class.get_ids_with_indexed_value('workflow_roles', '2')
assert set(ids) == set([id2, id3])
@postgresql
def test_create_user():
sql.SqlUser.wipe()
user = sql.SqlUser()
user.name = 'Pierre'
user.store()
@postgresql
def test_get_user():
sql.SqlUser.wipe()
user = sql.SqlUser()
user.name = 'Pierre'
user.store()
assert sql.SqlUser.get(user.id) is not None
@postgresql
def test_get_missing_user():
sql.SqlUser.wipe()
with pytest.raises(KeyError):
sql.SqlUser.get(12345)
@postgresql
def test_get_missing_user_ignore_errors():
sql.SqlUser.wipe()
assert sql.SqlUser.get(12345, ignore_errors=True) is None
@postgresql
def test_user_formdef():
sql.SqlUser.wipe()
from wcs.admin.settings import UserFieldsFormDef
formdef = UserFieldsFormDef(pub)
formdef.fields = [fields.StringField(id='3', label='test', type='string')]
formdef.store()
user = sql.SqlUser()
user.name = 'Pierre'
user.form_data = {'3': 'Papier'}
user.store()
assert sql.SqlUser.get(user.id, ignore_errors=True).form_data['3'] == 'Papier'
del pub.cfg['users']['formdef']
pub.write_cfg()
@postgresql
def test_get_users_with_role():
sql.SqlUser.wipe()
user = sql.SqlUser()
user.name = 'Pierre'
user.roles = [1]
user.store()
user_id = user.id
user = sql.SqlUser()
user.name = 'Papier'
user.store()
assert len(sql.SqlUser.get_users_with_role(1)) == 1
@postgresql
def test_get_users_with_name_identifier():
sql.SqlUser.wipe()
user = sql.SqlUser()
user.name = 'Pierre'
user.name_identifiers = ['foo']
user.store()
user_id = user.id
user = sql.SqlUser()
user.name = 'Papier'
user.store()
assert len(sql.SqlUser.get_users_with_name_identifier('foo')) == 1
assert sql.SqlUser.get_users_with_name_identifier('foo')[0].name == 'Pierre'
@postgresql
def test_get_users_fts():
sql.SqlUser.wipe()
user = sql.SqlUser()
user.name = 'Pierre'
user.name_identifiers = ['foo']
user.store()
user_id = user.id
user = sql.SqlUser()
user.name = 'Papier'
user.store()
assert len(sql.SqlUser.get_ids_from_query('pierre')) == 1
assert sql.SqlUser.get(sql.SqlUser.get_ids_from_query('pierre')[0]).id == user_id
@postgresql
def test_get_users_formdef_fts():
sql.SqlUser.wipe()
from wcs.admin.settings import UserFieldsFormDef
formdef = UserFieldsFormDef(pub)
formdef.fields = [fields.StringField(id='3', label='test', type='string')]
formdef.store()
user = sql.SqlUser()
user.name = 'Pierre'
user.form_data = {'3': 'Papier'}
user.store()
user_id = user.id
assert len(sql.SqlUser.get_ids_from_query('pierre papier')) == 1
assert sql.SqlUser.get(sql.SqlUser.get_ids_from_query('pierre papier')[0]).id == user_id
assert len(sql.SqlUser.get_ids_from_query('papier pierre')) == 1
assert sql.SqlUser.get(sql.SqlUser.get_ids_from_query('papier pierre')[0]).id == user_id
del pub.cfg['users']['formdef']
pub.write_cfg()
@postgresql
def test_urlname_change():
global formef
data_class = formdef.data_class(mode='sql')
data_class.wipe()
assert formdef.url_name == 'tests'
formdef.name = 'tests2'
formdef.store()
assert formdef.url_name == 'tests'
formdef.name = 'tests'
formdef.store()
assert formdef.url_name == 'tests'
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.status = 'wf-0'
formdata.user_id = '5'
formdata.store()
formdef.name = 'tests2'
formdef.store()
assert formdef.url_name == 'tests'
assert data_class.count() == 1
@postgresql
def test_sql_table_add_and_remove_fields():
test_formdef = FormDef()
test_formdef.name = 'tests and and remove fields'
test_formdef.fields = []
test_formdef.store()
assert test_formdef.table_name is not None
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
test_formdef.fields = [
fields.StringField(label='string'),
fields.EmailField(label='email'),
]
for field in test_formdef.fields:
if field.id is None:
field.id = test_formdef.get_new_field_id()
test_formdef.store()
test_formdef.fields.append(
fields.ItemField(label='item', items=('apple', 'pear', 'peach', 'apricot')))
test_formdef.fields[-1].id = test_formdef.get_new_field_id()
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
data_class.select()
previous_id = test_formdef.fields[-1].id
test_formdef.fields = test_formdef.fields[:-1]
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
data_class.select()
test_formdef.fields.append(fields.StringField(label='item'))
test_formdef.fields[-1].id = test_formdef.get_new_field_id()
test_formdef.store()
assert test_formdef.fields[-1].id != previous_id
data_class = test_formdef.data_class(mode='sql')
data_class.select()
test_formdef.fields = test_formdef.fields[:-1]
test_formdef.fields.append(
fields.ItemField(label='item', items=('apple', 'pear', 'peach', 'apricot')))
test_formdef.fields[-1].id = test_formdef.get_new_field_id()
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
data_class.select()
@postgresql
def test_sql_table_wipe_and_drop():
test_formdef = FormDef()
test_formdef.name = 'tests wipe and drop'
test_formdef.fields = []
test_formdef.store()
assert test_formdef.table_name is not None
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
conn, cur = sql.get_connection_and_cursor()
assert table_exists(cur, test_formdef.table_name)
conn.commit()
cur.close()
data_class.wipe(drop=True)
conn, cur = sql.get_connection_and_cursor()
assert not table_exists(cur, test_formdef.table_name)
assert not table_exists(cur, test_formdef.table_name + '_evolutions')
conn.commit()
cur.close()
test_formdef.store()
conn, cur = sql.get_connection_and_cursor()
assert table_exists(cur, test_formdef.table_name)
@postgresql
def test_sql_indexes():
test_formdef = FormDef()
test_formdef.name = 'tests indexes'
test_formdef.fields = []
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
conn, cur = sql.get_connection_and_cursor()
assert index_exists(cur, test_formdef.table_name + '_evolutions_fid')
conn.commit()
cur.close()
data_class.wipe(drop=True)
conn, cur = sql.get_connection_and_cursor()
assert not index_exists(cur, test_formdef.table_name + '_evolutions_fid')
conn.commit()
cur.close()
@postgresql
def test_sql_table_select():
test_formdef = FormDef()
test_formdef.name = 'table select'
test_formdef.fields = []
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
for i in range(50):
t = data_class()
t.store()
assert data_class.count() == 50
assert len(data_class.select()) == 50
assert len(data_class.select(lambda x: x.id < 26)) == 25
assert len(data_class.select([st.Less('id', 26)])) == 25
assert len(data_class.select([st.Less('id', 25), st.GreaterOrEqual('id', 10)])) == 15
assert len(data_class.select([st.Less('id', 25), st.GreaterOrEqual('id', 10), lambda x: x.id >= 15])) == 10
assert len(data_class.select([st.NotEqual('id', 26)])) == 49
assert len(data_class.select([st.Contains('id', [])])) == 0
assert len(data_class.select([st.Contains('id', [24, 25, 26])])) == 3
assert len(data_class.select([st.Contains('id', [24, 25, 86])])) == 2
assert len(data_class.select([st.NotContains('id', [24, 25, 86])])) == 48
assert len(data_class.select([st.NotContains('id', [])])) == 50
@postgresql
def test_sql_table_select_iterator():
test_formdef = FormDef()
test_formdef.name = 'table select'
test_formdef.fields = []
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
for i in range(50):
t = data_class()
t.store()
assert data_class.count() == 50
with pytest.raises(TypeError):
assert len(data_class.select(iterator=True)) == 50
# TypeError: object of type 'generator' has no len()
assert len(list(data_class.select(iterator=True))) == 50
assert len(list(data_class.select(lambda x: True, iterator=True))) == 50
assert len(list(data_class.select(lambda x: x.id < 26, iterator=True))) == 25
assert len(list(data_class.select([st.Less('id', 26)], iterator=True))) == 25
assert len(list(data_class.select([st.Less('id', 25), st.GreaterOrEqual('id', 10)],
iterator=True))) == 15
assert len(list(data_class.select([st.Less('id', 25), st.GreaterOrEqual('id', 10), lambda x:
x.id >= 15], iterator=True))) == 10
@postgresql
def test_sql_table_select_datetime():
test_formdef = FormDef()
test_formdef.name = 'table select datetime'
test_formdef.fields = []
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
d = datetime.datetime(2014, 1, 1)
for i in range(50):
t = data_class()
t.receipt_time = (d + datetime.timedelta(days=i)).timetuple()
t.store()
assert data_class.count() == 50
assert len(data_class.select()) == 50
assert len(data_class.select(lambda x: x.receipt_time == d.timetuple())) == 1
assert len(data_class.select([st.Equal('receipt_time', d.timetuple())])) == 1
assert len(data_class.select([
st.Less('receipt_time', (d + datetime.timedelta(days=20)).timetuple())])) == 20
assert len(data_class.select([
st.Greater('receipt_time', (d + datetime.timedelta(days=20)).timetuple())])) == 29
@postgresql
def test_select_limit_offset():
test_formdef = FormDef()
test_formdef.name = 'table select limit offset'
test_formdef.fields = []
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
for i in range(50):
t = data_class()
t.store()
assert len(data_class.select()) == 50
for iterator in (False, True):
for func_clause in (lambda x: True, None):
assert [x.id for x in data_class.select(func_clause, order_by='id', iterator=iterator)] == list(range(1, 51))
assert [x.id for x in data_class.select(func_clause, order_by='id', limit=10, iterator=iterator)] == list(range(1, 11))
assert [x.id for x in data_class.select(func_clause, order_by='id', limit=10, offset=10, iterator=iterator)] == list(range(11, 21))
assert [x.id for x in data_class.select(func_clause, order_by='id', limit=20, offset=20, iterator=iterator)] == list(range(21, 41))
assert [x.id for x in data_class.select(func_clause, order_by='id', offset=10, iterator=iterator)] == list(range(11, 51))
assert len([x.id for x in data_class.select(lambda x: x.id > 10, limit=10, iterator=iterator)]) == 10
@postgresql
def test_select_criteria_intersects():
data_class = formdef.data_class(mode='sql')
data_class.wipe()
formdata = data_class()
formdata.status = 'wf-0'
formdata.user_id = '5'
formdata.data = {'6': ['apricot']}
formdata.store()
formdata = data_class()
formdata.status = 'wf-0'
formdata.user_id = '5'
formdata.data = {'6': ['apricot', 'pear']}
formdata.store()
formdata = data_class()
formdata.status = 'wf-0'
formdata.user_id = '5'
formdata.data = {'6': []}
formdata.store()
assert len(data_class.select([st.Intersects('f6', ['apricot'])])) == 2
assert len(data_class.select([st.Intersects('f6', ['pear'])])) == 1
assert len(data_class.select([st.Intersects('f6', ['apple'])])) == 0
@postgresql
def test_count():
test_formdef = FormDef()
test_formdef.name = 'table select count'
test_formdef.fields = []
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
for i in range(50):
t = data_class()
t.store()
assert data_class.count() == 50
assert data_class.count([st.Less('id', 26)]) == 25
@postgresql
def test_select_criteria_or_and():
test_formdef = FormDef()
test_formdef.name = 'table select criteria or and'
test_formdef.fields = []
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
for i in range(50):
t = data_class()
t.store()
assert [x.id for x in data_class.select([st.Or([st.Less('id', 10)])], order_by='id')] == list(range(1, 10))
assert [x.id for x in data_class.select([st.Or([
st.Less('id', 10), st.Equal('id', 15)])], order_by='id')] == list(range(1, 10)) + [15]
assert [x.id for x in data_class.select([st.And([
st.Less('id', 10), st.Greater('id', 5)])], order_by='id')] == list(range(6, 10))
@postgresql
def test_select_criteria_null():
test_formdef = FormDef()
test_formdef.name = 'table select criteria null'
test_formdef.fields = []
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
for x in range(50):
t = data_class()
if x % 3:
t.submission_channel = None
else:
t.submission_channel = 'mail'
t.store()
assert len(data_class.select([st.Null('submission_channel')])) == 33
assert len(data_class.select([st.NotNull('submission_channel')])) == 17
@postgresql
def test_sql_table_select_bool():
test_formdef = FormDef()
test_formdef.name = 'table select bool'
test_formdef.fields = [fields.BoolField(id='3', label='bool')]
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
for i in range(50):
t = data_class()
t.data = {'3': False}
t.store()
t.data = {'3': True}
t.store()
assert data_class.count() == 50
assert len(data_class.select()) == 50
assert len(data_class.select([st.Equal('f3', True)])) == 1
assert len(data_class.select([st.Equal('f3', False)])) == 49
@postgresql
def test_sql_criteria_ilike():
test_formdef = FormDef()
test_formdef.name = 'table select bool'
test_formdef.fields = [fields.StringField(id='3', label='string')]
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
for i in range(50):
t = data_class()
if i < 20:
t.data = {'3': 'foo'}
else:
t.data = {'3': 'bar'}
t.store()
t.store()
assert data_class.count() == 50
assert len(data_class.select()) == 50
assert [x.id for x in data_class.select([st.ILike('f3', 'bar')], order_by='id')] == list(range(21, 51))
assert [x.id for x in data_class.select([st.ILike('f3', 'BAR')], order_by='id')] == list(range(21, 51))
@postgresql
def test_sql_criteria_fts():
test_formdef = FormDef()
test_formdef.name = 'table select fts'
test_formdef.fields = [fields.StringField(id='3', label='string')]
test_formdef.store()
data_class = test_formdef.data_class(mode='sql')
assert data_class.count() == 0
for i in range(50):
t = data_class()
if i < 20:
t.data = {'3': 'foo'}
else:
t.data = {'3': 'bar'}
t.just_created()
t.store()
assert data_class.count() == 50
assert len(data_class.select()) == 50
assert [x.id for x in data_class.select([st.FtsMatch('BAR')], order_by='id')] == list(range(21, 51))
# check fts against data in history
assert len(data_class.select([st.FtsMatch('XXX')])) == 0
formdata1 = data_class.select([st.FtsMatch('BAR')])[0]
formdata1.evolution[0].comment = 'XXX'
formdata1.store()
assert len(data_class.select([st.FtsMatch('XXX')])) == 1
assert data_class.select([st.FtsMatch('XXX')])[0].id == formdata1.id
assert len(data_class.select([st.FtsMatch('yyy')])) == 0
item = RegisterCommenterWorkflowStatusItem()
item.comment = '<span>ÿÿÿ</span>'
item.perform(formdata1)
assert formdata1.evolution[-1].display_parts()[-1] == '<span>ÿÿÿ</span>'
formdata1.store()
assert len(data_class.select([st.FtsMatch('yyy')])) == 1
assert len(data_class.select([st.FtsMatch('span')])) == 0
assert data_class.count([st.FtsMatch('Pierre')]) == 0
sql.SqlUser.wipe()
user = sql.SqlUser()
user.name = 'Pierre'
user.store()
t.user_id = user.id
t.store()
assert data_class.count([st.FtsMatch('Pierre')]) == 1
# check unaccent
user = sql.SqlUser()
user.name = force_str(u'Frédéric')
user.store()
t.user_id = user.id
t.store()
assert data_class.count([st.FtsMatch(user.name)]) == 1
assert data_class.count([st.FtsMatch('Frederic')]) == 1
def table_exists(cur, table_name):
cur.execute('''SELECT COUNT(*) FROM information_schema.tables
WHERE table_name = %s''', (table_name,))
return bool(cur.fetchone()[0] == 1)
def column_exists_in_table(cur, table_name, column_name):
cur.execute('''SELECT COUNT(*) FROM information_schema.columns
WHERE table_name = %s
AND column_name = %s''', (table_name, column_name))
return bool(cur.fetchone()[0] == 1)
def index_exists(cur, index_name):
cur.execute('''SELECT COUNT(*) FROM pg_indexes
WHERE schemaname = 'public'
AND indexname = %s''', (index_name,))
return bool(cur.fetchone()[0] == 1)
@postgresql
def test_sql_level():
conn, cur = sql.get_connection_and_cursor()
cur.execute('DROP TABLE wcs_meta')
assert sql.get_sql_level(conn, cur) == 0
sql.migrate()
assert sql.get_sql_level(conn, cur) == sql.SQL_LEVEL
# insert negative SQL level, to trigger an error, and check it's not
# changed.
cur.execute('''UPDATE wcs_meta SET value = %s WHERE key = %s''',
(str(-1), 'sql_level'))
assert sql.get_sql_level(conn, cur) == -1
with pytest.raises(RuntimeError):
sql.migrate()
assert sql.get_sql_level(conn, cur) == -1
conn.commit()
cur.close()
def migration_level(cur):
cur.execute('SELECT value FROM wcs_meta WHERE key = %s', ('sql_level',))
row = cur.fetchone()
return int(row[0])
@postgresql
def test_migration_1_tracking_code():
conn, cur = sql.get_connection_and_cursor()
cur.execute('DROP TABLE wcs_meta')
cur.execute('DROP TABLE tracking_codes')
sql.migrate()
assert table_exists(cur, 'tracking_codes')
assert table_exists(cur, 'wcs_meta')
assert migration_level(cur) >= 1
conn.commit()
cur.close()
@postgresql
def test_migration_2_formdef_id_in_views():
conn, cur = sql.get_connection_and_cursor()
cur.execute('UPDATE wcs_meta SET value = 1 WHERE key = %s', ('sql_level',))
cur.execute('DROP VIEW wcs_all_forms CASCADE')
# hack a formdef table the wrong way, to check it is reconstructed
# properly before the views are created
formdef.fields[4] = fields.StringField(id='4', label='item')
cur.execute('DROP VIEW wcs_view_1_tests')
cur.execute('ALTER TABLE formdata_1_tests DROP COLUMN f4_display')
sql.redo_views(conn, cur, formdef, rebuild_global_views=False)
formdef.fields[4] = fields.ItemField(id='4', label='item', items=('apple', 'pear', 'peach', 'apricot'))
assert table_exists(cur, 'wcs_view_1_tests')
assert not column_exists_in_table(cur, 'wcs_view_1_tests', 'f4_display')
view_names = []
cur.execute('''SELECT table_name FROM information_schema.views
WHERE table_name LIKE %s''', ('wcs\_view\_%',))
while True:
row = cur.fetchone()
if row is None:
break
view_names.append(row[0])
fake_formdef = FormDef()
common_fields = sql.get_view_fields(fake_formdef)
# remove formdef_id for the purpose of this test
common_fields.remove([x for x in common_fields if x[1] == 'formdef_id'][0])
union = ' UNION '.join(['''SELECT %s FROM %s''' % (
', '.join([y[1] for y in common_fields]), x) for x in view_names])
assert not 'formdef_id' in union
cur.execute('''CREATE VIEW wcs_all_forms AS %s''' % union)
sql.migrate()
assert column_exists_in_table(cur, 'wcs_all_forms', 'formdef_id')
assert migration_level(cur) >= 2
conn.commit()
cur.close()
@postgresql
def test_migration_6_actions_roles():
conn, cur = sql.get_connection_and_cursor()
cur.execute('UPDATE wcs_meta SET value = 5 WHERE key = %s', ('sql_level',))
cur.execute('DROP VIEW wcs_all_forms CASCADE')
# hack a formdef table the wrong way, to check it is reconstructed
# properly before the views are created
formdef.fields[4] = fields.StringField(id='4', label='item')
cur.execute('DROP VIEW wcs_view_1_tests')
cur.execute('ALTER TABLE formdata_1_tests DROP COLUMN actions_roles_array')
sql.drop_views(formdef, conn, cur)
formdef.fields[4] = fields.ItemField(id='4', label='item', items=('apple', 'pear', 'peach', 'apricot'))
assert not column_exists_in_table(cur, 'formdata_1_tests', 'actions_roles_array')
sql.migrate()
assert column_exists_in_table(cur, 'formdata_1_tests', 'actions_roles_array')
assert column_exists_in_table(cur, 'wcs_view_1_tests', 'actions_roles_array')
assert column_exists_in_table(cur, 'wcs_all_forms', 'actions_roles_array')
assert migration_level(cur) >= 6
conn.commit()
cur.close()
@postgresql
def test_migration_10_submission_channel():
conn, cur = sql.get_connection_and_cursor()
cur.execute('UPDATE wcs_meta SET value = 9 WHERE key = %s', ('sql_level',))
cur.execute('DROP VIEW wcs_all_forms CASCADE')
# hack a formdef table the wrong way, to check it is reconstructed
# properly before the views are created
formdef.fields[4] = fields.StringField(id='4', label='item')
cur.execute('DROP VIEW wcs_view_1_tests')
cur.execute('ALTER TABLE formdata_1_tests DROP COLUMN submission_channel')
sql.drop_views(formdef, conn, cur)
formdef.fields[4] = fields.ItemField(id='4', label='item', items=('apple', 'pear', 'peach', 'apricot'))
assert not column_exists_in_table(cur, 'formdata_1_tests', 'submission_channel')
sql.migrate()
assert column_exists_in_table(cur, 'formdata_1_tests', 'submission_channel')
assert column_exists_in_table(cur, 'wcs_view_1_tests', 'submission_channel')
assert column_exists_in_table(cur, 'wcs_all_forms', 'submission_channel')
assert migration_level(cur) >= 10
conn.commit()
cur.close()
@postgresql
def test_migration_12_users_fts():
conn, cur = sql.get_connection_and_cursor()
cur.execute('UPDATE wcs_meta SET value = 11 WHERE key = %s', ('sql_level',))
sql.SqlUser.wipe()
user = sql.SqlUser()
user.name = 'Pierre'
user.store()
# remove the fts column
cur.execute('ALTER TABLE users DROP COLUMN fts')
assert not column_exists_in_table(cur, 'users', 'fts')
sql.migrate()
assert column_exists_in_table(cur, 'users', 'fts')
assert migration_level(cur) >= 12
# no fts, migration only prepare re-index
assert len(sql.SqlUser.get_ids_from_query('pierre')) == 0
assert sql.is_reindex_needed('user', conn=conn, cur=cur) is True
assert sql.is_reindex_needed('formdata', conn=conn, cur=cur) is True
call_command('cron') # first cron = reindex
assert sql.is_reindex_needed('user', conn=conn, cur=cur) is False
assert sql.is_reindex_needed('formdata', conn=conn, cur=cur) is False
# make sure the fts is filled after the migration
assert len(sql.SqlUser.get_ids_from_query('pierre')) == 1
conn.commit()
cur.close()
@postgresql
def test_migration_21_users_ascii_name():
conn, cur = sql.get_connection_and_cursor()
cur.execute('UPDATE wcs_meta SET value = 11 WHERE key = %s', ('sql_level',))
sql.SqlUser.wipe()
user = sql.SqlUser()
user.name = 'Jean Sénisme'
user.store()
# remove the ascii_name column
cur.execute('ALTER TABLE users DROP COLUMN ascii_name')
assert not column_exists_in_table(cur, 'users', 'ascii_name')
sql.migrate()
assert column_exists_in_table(cur, 'users', 'ascii_name')
assert migration_level(cur) >= 21
# no fts, migration only prepare re-index
assert sql.SqlUser.count([st.Equal('ascii_name', 'jean senisme')]) == 0
assert sql.is_reindex_needed('user', conn=conn, cur=cur) is True
assert sql.is_reindex_needed('formdata', conn=conn, cur=cur) is True
call_command('cron') # first cron = reindex
assert sql.is_reindex_needed('user', conn=conn, cur=cur) is False
assert sql.is_reindex_needed('formdata', conn=conn, cur=cur) is False
# make sure the ascii_name is filled after the migration
assert sql.SqlUser.count([st.Equal('ascii_name', 'jean senisme')]) == 1
conn.commit()
cur.close()
@postgresql
def test_migration_24_evolution_index():
formdef = FormDef()
formdef.name = 'tests migration 24'
formdef.fields = []
formdef.store()
conn, cur = sql.get_connection_and_cursor()
cur.execute('DROP INDEX %s_evolutions_fid' % formdef.table_name)
cur.execute('UPDATE wcs_meta SET value = 23 WHERE key = %s', ('sql_level',))
conn.commit()
cur.close()
conn, cur = sql.get_connection_and_cursor()
assert not index_exists(cur, formdef.table_name + '_evolutions_fid')
conn.commit()
cur.close()
sql.migrate()
conn, cur = sql.get_connection_and_cursor()
assert index_exists(cur, formdef.table_name + '_evolutions_fid')
conn.commit()
cur.close()
def drop_formdef_tables():
conn, cur = sql.get_connection_and_cursor()
cur.execute('''SELECT table_name FROM information_schema.tables''')
table_names = []
while True:
row = cur.fetchone()
if row is None:
break
table_names.append(row[0])
for table_name in table_names:
if table_name.startswith('formdata_'):
cur.execute('DROP TABLE %s CASCADE' % table_name)
@postgresql
def test_is_at_endpoint():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
wf = Workflow(name='test endpoint')
st1 = wf.add_status('Status1', 'st1')
st2 = wf.add_status('Status2', 'st2')
commentable = CommentableWorkflowStatusItem()
commentable.id = '_commentable'
commentable.by = ['_submitter', '_receiver']
st1.items.append(commentable)
commentable.parent = st1
wf.store()
assert [x.id for x in wf.get_endpoint_status()] == ['st2']
formdef = FormDef()
formdef.name = 'test endpoint'
formdef.fields = []
formdef.workflow = wf
formdef.store()
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.status = 'wf-st1'
formdata.store()
formdata = data_class()
formdata.status = 'wf-st2'
formdata.store()
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms''')
assert bool(cur.fetchone()[0] == 2)
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms WHERE status = 'wf-st1' ''')
assert bool(cur.fetchone()[0] == 1)
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms WHERE is_at_endpoint = true''')
assert bool(cur.fetchone()[0] == 1)
# check a change to workflow is reflected in the database
st1.forced_endpoint = True
wf.store()
assert [x.id for x in wf.get_endpoint_status()] == ['st1', 'st2']
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms WHERE is_at_endpoint = true''')
assert bool(cur.fetchone()[0] == 2)
@postgresql
def test_views_fts():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
formdef = FormDef()
formdef.name = 'test fts'
formdef.fields = [
fields.StringField(id='0', label='string'),
]
formdef.store()
data_class = formdef.data_class(mode='sql')
formdata1 = data_class()
formdata1.data = {'0': 'foo bar'}
formdata1.store()
formdata2 = data_class()
formdata2.data = {'0': 'foo'}
formdata2.store()
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms WHERE fts @@ plainto_tsquery(%s)''', ('foo',))
assert bool(cur.fetchone()[0] == 2)
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms WHERE fts @@ plainto_tsquery(%s)''', ('bar',))
assert bool(cur.fetchone()[0] == 1)
@postgresql
def test_select_any_formdata():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
now = datetime.datetime.now()
cnt = 0
for i in range(5):
formdef = FormDef()
formdef.name = 'test any %d' % i
formdef.fields = []
formdef.store()
data_class = formdef.data_class(mode='sql')
for j in range(20):
formdata = data_class()
formdata.just_created()
formdata.user_id = '%s' % ((i+j)%11)
# set receipt_time to make sure all entries are unique.
formdata.receipt_time = (now + datetime.timedelta(seconds=cnt)).timetuple()
formdata.status = ['wf-new', 'wf-accepted', 'wf-rejected', 'wf-finished'][(i+j)%4]
if j < 5:
formdata.submission_channel = 'mail'
formdata.store()
cnt += 1
# test generic select
objects = sql.AnyFormData.select()
assert len(objects) == 100
# make sure valid formdefs are used
assert len([x for x in objects if x.formdef.name == 'test any 0']) == 20
assert len([x for x in objects if x.formdef.name == 'test any 1']) == 20
# test sorting
objects = sql.AnyFormData.select(order_by='receipt_time')
assert len(objects) == 100
objects2 = sql.AnyFormData.select(order_by='-receipt_time')
assert [(x.formdef_id, x.id) for x in objects2] == list(reversed(
[(x.formdef_id, x.id) for x in objects]))
# test clauses
objects2 = sql.AnyFormData.select([st.Equal('user_id', '0')])
assert len(objects2) == len([x for x in objects if x.user_id == '0'])
objects2 = sql.AnyFormData.select([st.Equal('is_at_endpoint', True)])
assert len(objects2) == len([x for x in objects if x.status in ('wf-rejected', 'wf-finished')])
objects2 = sql.AnyFormData.select([st.Equal('submission_channel', 'mail')])
assert len(objects2) == len([x for x in objects if x.submission_channel == 'mail'])
assert objects2[0].submission_channel == 'mail'
# test offset/limit
objects2 = sql.AnyFormData.select(order_by='receipt_time', limit=10, offset=0)
assert [(x.formdef_id, x.id) for x in objects2] == [(x.formdef_id, x.id) for x in objects][:10]
objects2 = sql.AnyFormData.select(order_by='receipt_time', limit=10, offset=20)
assert [(x.formdef_id, x.id) for x in objects2] == [(x.formdef_id, x.id) for x in objects][20:30]
@postgresql
def test_geoloc_in_global_view():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
now = datetime.datetime.now()
formdef = FormDef()
formdef.name = 'test no geoloc'
formdef.fields = []
formdef.store()
formdef2 = FormDef()
formdef2.name = 'test with geoloc'
formdef2.fields = []
formdef2.geolocations = {'base': 'Plop'}
formdef2.store()
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.just_created()
formdata.store()
data_class = formdef2.data_class(mode='sql')
formdata = data_class()
formdata.just_created()
formdata.geolocations = {'base': {'lat': 12, 'lon': 21}}
formdata.store()
# test generic select
objects = sql.AnyFormData.select()
assert len(objects) == 2
# test clauses
objects2 = sql.AnyFormData.select([st.Null('geoloc_base_x')])
assert len(objects2) == 1
assert not objects2[0].geolocations
objects2 = sql.AnyFormData.select([st.NotNull('geoloc_base_x')])
assert len(objects2) == 1
assert int(objects2[0].geolocations['base']['lat']) == formdata.geolocations['base']['lat']
assert int(objects2[0].geolocations['base']['lon']) == formdata.geolocations['base']['lon']
@postgresql
def test_actions_roles():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
wf = Workflow(name='test endpoint')
st1 = wf.add_status('Status1', 'st1')
st2 = wf.add_status('Status2', 'st2')
commentable = CommentableWorkflowStatusItem()
commentable.id = '_commentable'
commentable.by = ['_submitter', '1']
st1.items.append(commentable)
commentable.parent = st1
wf.store()
assert [x.id for x in wf.get_endpoint_status()] == ['st2']
formdef = FormDef()
formdef.name = 'test actions roles'
formdef.fields = []
formdef.workflow = wf
formdef.store()
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.status = 'wf-st1'
formdata.store()
formdata_id = formdata.id
formdata = data_class()
formdata.status = 'wf-st2'
formdata.store()
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms''')
assert bool(cur.fetchone()[0] == 2)
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms
WHERE actions_roles_array && ARRAY['5', '1', '4']''')
assert bool(cur.fetchone()[0] == 1)
# check a change to workflow is reflected in the database
st1.items[0].by = ['2', '3']
wf.store()
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms
WHERE actions_roles_array && ARRAY['5', '1', '4']''')
assert bool(cur.fetchone()[0] == 0)
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms
WHERE actions_roles_array && ARRAY['2', '3']''')
assert bool(cur.fetchone()[0] == 1)
# using criterias
criterias = [st.Intersects('actions_roles_array', ['2', '3'])]
total_count = sql.AnyFormData.count(criterias)
formdatas = sql.AnyFormData.select(criterias)
assert total_count == 1
assert formdatas[0].id == formdata_id
@postgresql
def test_last_update_time():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
wf = Workflow(name='test last update time')
st1 = wf.add_status('Status1', 'st1')
commentable = CommentableWorkflowStatusItem()
commentable.id = '_commentable'
commentable.by = ['_submitter', '_receiver']
st1.items.append(commentable)
commentable.parent = st1
wf.store()
formdef = FormDef()
formdef.name = 'test last update time'
formdef.fields = []
formdef.workflow = wf
formdef.store()
data_class = formdef.data_class(mode='sql')
formdata1 = data_class()
formdata1.status = 'wf-st1'
formdata1.just_created()
formdata1.evolution[0].comment = 'comment'
formdata1.jump_status('st1') # will add another evolution entry
formdata1.evolution[0].time = datetime.datetime(2015, 1, 1, 0, 0, 0).timetuple()
formdata1.evolution[1].time = datetime.datetime(2015, 1, 2, 0, 0, 0).timetuple()
formdata1.store()
formdata2 = data_class()
formdata2.status = 'wf-st1'
formdata2.just_created()
formdata2.evolution[0].comment = 'comment'
formdata2.jump_status('st1') # will add another evolution entry
formdata2.evolution[0].time = datetime.datetime(2015, 1, 3, 0, 0, 0).timetuple()
formdata2.evolution[1].time = datetime.datetime(2015, 1, 4, 0, 0, 0).timetuple()
formdata2.store()
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms''')
assert bool(cur.fetchone()[0] == 2)
cur.execute('''SELECT id FROM wcs_all_forms WHERE last_update_time = '2015-01-02 00:00' ''')
assert bool(cur.fetchone()[0] == formdata1.id)
cur.execute('''SELECT id FROM wcs_all_forms WHERE last_update_time = '2015-01-04 00:00' ''')
assert bool(cur.fetchone()[0] == formdata2.id)
@postgresql
def test_view_formdef_name():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
formdef1 = FormDef()
formdef1.name = 'test formdef name 1'
formdef1.fields = []
formdef1.store()
data_class = formdef1.data_class()
formdata1 = data_class()
formdata1.just_created()
formdata1.store()
formdef2 = FormDef()
formdef2.name = 'test formdef name 2'
formdef2.fields = []
formdef2.store()
data_class = formdef2.data_class()
formdata2 = data_class()
formdata2.just_created()
formdata2.store()
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms''')
assert bool(cur.fetchone()[0] == 2)
cur.execute('''SELECT formdef_id FROM wcs_all_forms WHERE formdef_name = 'test formdef name 1' ''')
assert bool(str(cur.fetchone()[0]) == str(formdef1.id))
cur.execute('''SELECT formdef_id FROM wcs_all_forms WHERE formdef_name = 'test formdef name 2' ''')
assert bool(str(cur.fetchone()[0]) == str(formdef2.id))
@postgresql
def test_view_user_name():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
formdef1 = FormDef()
formdef1.name = 'test user name'
formdef1.fields = []
formdef1.store()
sql.SqlUser.wipe()
user = sql.SqlUser()
user.name = 'Foobar'
user.store()
data_class = formdef1.data_class()
formdata1 = data_class()
formdata1.just_created()
formdata1.store()
data_class = formdef1.data_class()
formdata2 = data_class()
formdata2.user_id = user.id
formdata2.just_created()
formdata2.store()
cur.execute('''SELECT user_name FROM wcs_all_forms WHERE id = %s ''', (formdata1.id,))
assert bool(cur.fetchone()[0] is None)
cur.execute('''SELECT user_name FROM wcs_all_forms WHERE id = %s ''', (formdata2.id,))
assert bool(cur.fetchone()[0] == user.name)
@postgresql
def test_select_formdata_after_formdef_removal():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
now = datetime.datetime.now()
for i in range(2):
formdef = FormDef()
formdef.name = 'test formdef removal'
formdef.fields = []
formdef.store()
data_class = formdef.data_class(mode='sql')
formdata = data_class()
formdata.just_created()
formdata.store()
# test generic select
objects = sql.AnyFormData.select()
assert len(objects) == 2
formdef.remove_self()
objects = sql.AnyFormData.select()
assert len(objects) == 1
@postgresql
def test_views_submission_info():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
formdef = FormDef()
formdef.name = 'test backoffice submission'
formdef.fields = []
formdef.store()
data_class = formdef.data_class(mode='sql')
formdata1 = data_class()
formdata1.submission_channel = 'mail'
formdata1.backoffice_submission = True
formdata1.store()
formdata2 = data_class()
formdata2.backoffice_submission = False
formdata2.store()
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms WHERE backoffice_submission IS TRUE''')
assert bool(cur.fetchone()[0] == 1)
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms WHERE backoffice_submission IS FALSE''')
assert bool(cur.fetchone()[0] == 1)
cur.execute('''SELECT COUNT(*) FROM wcs_all_forms WHERE submission_channel = %s''', ('mail',))
assert bool(cur.fetchone()[0] == 1)
@postgresql
def test_get_formdef_new_id():
test1_formdef = FormDef()
test1_formdef.name = 'new formdef'
test1_formdef.fields = []
test1_formdef.store()
test1_id = test1_formdef.id
test1_table_name = test1_formdef.table_name
test1_formdef.remove_self()
test2_formdef = FormDef()
test2_formdef.name = 'new formdef'
test2_formdef.fields = []
test2_formdef.store()
assert test1_id != test2_formdef.id
assert test1_table_name != test2_formdef.table_name
@postgresql
def test_criticality_levels():
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
workflow1 = Workflow(name='criticality1')
workflow1.criticality_levels = [
WorkflowCriticalityLevel(name='green'),
WorkflowCriticalityLevel(name='yellow'),
WorkflowCriticalityLevel(name='red'),
WorkflowCriticalityLevel(name='redder'),
WorkflowCriticalityLevel(name='reddest'),
]
workflow1.store()
workflow2 = Workflow(name='criticality2')
workflow2.criticality_levels = [
WorkflowCriticalityLevel(name='green'),
WorkflowCriticalityLevel(name='reddest'),
]
workflow2.store()
formdef1 = FormDef()
formdef1.name = 'test criticality levels 1'
formdef1.fields = []
formdef1.workflow_id = workflow1.id
formdef1.store()
formdef2 = FormDef()
formdef2.name = 'test criticality levels 2'
formdef2.fields = []
formdef2.workflow_id = workflow2.id
formdef2.store()
data_class = formdef1.data_class(mode='sql')
for i in range(5):
formdata = data_class()
formdata.set_criticality_level(i)
formdata.store()
data_class = formdef2.data_class(mode='sql')
for i in range(2):
formdata = data_class()
formdata.set_criticality_level(i)
formdata.store()
objects = sql.AnyFormData.select(order_by='-criticality_level')
# make sure first two formdata are the highest priority ones, and the last
# two formdata are the lowest priority ones.
assert objects[0].get_criticality_level_object().name == 'reddest'
assert objects[1].get_criticality_level_object().name == 'reddest'
assert objects[-1].get_criticality_level_object().name == 'green'
assert objects[-2].get_criticality_level_object().name == 'green'
@postgresql
def test_view_performances():
pytest.skip('takes too much time')
drop_formdef_tables()
conn, cur = sql.get_connection_and_cursor()
nb_users = 1000
nb_roles = 10
nb_workflows = 5
nb_formdefs = 10
nb_fields = 10
nb_formdatas = 1000
nb_users = 10
nb_formdatas = 10000
random.seed('foobar')
# create users
sql.SqlUser.wipe()
users = []
for i in range(nb_users):
user = sql.SqlUser()
user.name = 'user %s' % i
user.store()
users.append(user)
# create roles
roles = []
for i in range(nb_roles):
role = Role(name='role%s' % i)
role.store()
roles.append(role)
# create workflows
workflows = []
for i in range(nb_workflows):
workflow = Workflow(name='test perf wf %s' % i)
for j in range(5):
status = workflow.add_status('Status %d' % j, 'st%s' % j)
commentable = CommentableWorkflowStatusItem()
commentable.id = '_commentable%s' % j
commentable.by = [random.choice(roles).id, random.choice(roles).id]
status.items.append(commentable)
commentable.parent = status
if j != 4:
jump = JumpWorkflowStatusItem()
jump.id = '_jump%s' % j
jump.by = []
jump.timeout = 5
jump.status = 'st%s' % (j+1)
status.items.append(jump)
jump.parent = status
workflow.store()
workflows.append(workflow)
# create formdefs
formdefs = []
for i in range(nb_formdefs):
formdef = FormDef()
formdef.name = 'test performance view %s' % i
formdef.fields = []
for j in range(nb_fields):
formdef.fields.append(fields.StringField(id=str(j+1), label='string'))
formdef.workflow_id = workflows[i%5].id
formdef.store()
formdefs.append(formdef)
print('create formdatas')
# create formdatas
for i in range(nb_formdatas):
data_class = random.choice(formdefs).data_class()
formdata = data_class()
formdata.data = {}
for j in range(10):
formdata.data[str(j+1)] = ''.join([random.choice(string.letters) for x in range(random.randint(10, 30))])
formdata.user_id = random.choice(users).id
formdata.status = 'wf-st1'
formdata.just_created()
for j in range(5):
formdata.jump_status('st%s' % (j+2))
if random.random() < 0.5:
break
print('done')
t0 = time.time()
user_roles = [random.choice(roles).id, random.choice(roles).id]
criterias = []
criterias.append(st.NotEqual('status', 'draft'))
criterias.append(st.Equal('is_at_endpoint', False))
criterias.append(st.Intersects('actions_roles_array', user_roles))
formdatas = sql.AnyFormData.select(criterias, order_by='receipt_time', limit=20, offset=0)
print(time.time() - t0)
assert (time.time() - t0) < 0.5
@postgresql
def test_migration_30_anonymize_evo_who():
formdef = FormDef()
formdef.name = 'tests migration 24'
formdef.fields = []
formdef.store()
user = sql.SqlUser()
user.name = 'JohnDoe'
user.store()
klass = formdef.data_class()
formdata = klass()
formdata.evolution = []
formdata.anonymised = datetime.datetime.now()
evo = Evolution(formdata)
evo.who = user.id
evo.time = time.localtime()
formdata.evolution.append(evo)
formdata.store()
conn, cur = sql.get_connection_and_cursor()
cur.execute('UPDATE wcs_meta SET value = 29 WHERE key = %s', ('sql_level',))
conn.commit()
cur.close()
conn, cur = sql.get_connection_and_cursor()
cur.execute('SELECT COUNT(*) FROM %s_evolutions WHERE who IS NULL' % formdef.table_name)
assert cur.fetchone() == (0,)
cur.execute('SELECT COUNT(*) FROM wcs_meta WHERE key = %s AND value::integer > 29', ('sql_level',))
assert cur.fetchone() == (0,)
conn.commit()
cur.close()
sql.migrate()
conn, cur = sql.get_connection_and_cursor()
cur.execute('SELECT COUNT(*) FROM %s_evolutions WHERE who IS NULL' % formdef.table_name)
assert cur.fetchone() == (1,)
cur.execute('SELECT COUNT(*) FROM wcs_meta WHERE key = %s AND value::integer > 29', ('sql_level',))
assert cur.fetchone() == (1,)
conn.commit()
cur.close()
@postgresql
def test_migration_31_user_label():
conn, cur = sql.get_connection_and_cursor()
cur.execute('UPDATE wcs_meta SET value = 30 WHERE key = %s', ('sql_level',))
cur.execute('DROP VIEW wcs_all_forms CASCADE')
cur.execute('DROP VIEW wcs_view_1_tests')
cur.execute('ALTER TABLE formdata_1_tests DROP COLUMN user_label')
sql.drop_views(formdef, conn, cur)
assert not column_exists_in_table(cur, 'formdata_1_tests', 'user_label')
sql.migrate()
assert column_exists_in_table(cur, 'formdata_1_tests', 'user_label')
assert column_exists_in_table(cur, 'wcs_view_1_tests', 'user_label')
assert column_exists_in_table(cur, 'wcs_all_forms', 'user_label')
assert migration_level(cur) >= 31
conn.commit()
cur.close()
View Git Blame Copy Permalink