wcs/tests/api/test_statistics.py

1258 lines
45 KiB
Python

import datetime
import os
import pytest
from wcs import fields
from wcs.backoffice.management import format_time
from wcs.blocks import BlockDef
from wcs.carddef import CardDef
from wcs.categories import CardDefCategory, Category
from wcs.formdef import FormDef
from wcs.qommon.http_request import HTTPRequest
from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef
from ..utilities import clean_temporary_pub, create_temporary_pub, get_app
from .utils import sign_uri
def get_humanized_duration_serie(json_resp):
return [format_time(x) for x in json_resp['data']['series'][0]['data']]
@pytest.fixture
def pub():
pub = create_temporary_pub()
BlockDef.wipe()
Category.wipe()
FormDef.wipe()
Workflow.wipe()
CardDef.wipe()
req = HTTPRequest(None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net'})
pub.set_app_dir(req)
pub.cfg['identification'] = {'methods': ['password']}
pub.cfg['language'] = {'language': 'en'}
pub.write_cfg()
with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd:
fd.write(
'''\
[api-secrets]
coucou = 1234
'''
)
return pub
@pytest.fixture
def formdef(pub):
workflow = Workflow(name='Workflow One')
new_status = workflow.add_status(name='New status')
workflow.add_status(name='End status')
middle_status1 = workflow.add_status(name='Middle status 1')
middle_status2 = workflow.add_status(name='Middle status 2')
jump = new_status.add_action('jump', id='_jump')
jump.status = '2'
jump.timeout = 86400
jump = new_status.add_action('jump', id='_jump')
jump.status = '3'
jump = middle_status1.add_action('jump', id='_jump')
jump.status = '4'
jump = middle_status2.add_action('jump', id='_jump')
jump.status = '2'
workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow)
workflow.backoffice_fields_formdef.fields = [
fields.BoolField(
id='1', varname='checkbox', label='Checkbox', type='bool', display_locations=['statistics']
),
]
workflow.store()
block = BlockDef()
block.name = 'foobar'
block.fields = [
fields.BoolField(id='1', label='Bool', type='bool', varname='bool', display_locations=['statistics'])
]
block.store()
formdef = FormDef()
formdef.name = 'test'
formdef.workflow_id = workflow.id
item_field = fields.ItemField(
id='2', varname='test-item', label='Test item', type='item', items=['foo', 'bar', 'baz']
)
item_field.display_locations = ['statistics']
items_field = fields.ItemsField(
id='3',
varname='test-items',
label='Test items',
type='items',
items=['foo', 'bar', 'baz'],
anonymise=False,
)
items_field.display_locations = ['statistics']
block_field = fields.BlockField(id='4', label='Block Data', varname='blockdata', type='block:foobar')
formdef.fields = [item_field, items_field, block_field]
formdef.store()
formdef.data_class().wipe()
return formdef
def teardown_module(module):
clean_temporary_pub()
def test_statistics_index(pub):
get_app(pub).get('/api/statistics/', status=403)
resp = get_app(pub).get(sign_uri('/api/statistics/'))
assert resp.json['data'][0]['name'] == 'Forms Count'
assert resp.json['data'][0]['url'] == 'http://example.net/api/statistics/forms/count/'
def test_statistics_index_categories(pub):
Category(name='Category A').store()
Category(name='Category B').store()
resp = get_app(pub).get(sign_uri('/api/statistics/'))
category_filter = [x for x in resp.json['data'][0]['filters'] if x['id'] == 'category'][0]
assert category_filter['options'] == [
{'id': '_all', 'label': 'All'},
{'id': 'category-a', 'label': 'Category A'},
{'id': 'category-b', 'label': 'Category B'},
]
assert category_filter['deprecated'] is True
def test_statistics_index_forms(pub):
formdef = FormDef()
formdef.name = 'test 1'
formdef.fields = []
formdef.store()
formdef.data_class().wipe()
formdef2 = FormDef()
formdef2.name = 'test 2'
formdef2.fields = []
formdef2.store()
formdef2.data_class().wipe()
resp = get_app(pub).get(sign_uri('/api/statistics/'))
form_filter = [x for x in resp.json['data'][0]['filters'] if x['id'] == 'form'][0]
assert form_filter['options'] == [
{'id': '_all', 'label': 'All Forms'},
{'id': 'test-1', 'label': 'test 1'},
{'id': 'test-2', 'label': 'test 2'},
]
category_a = Category(name='Category A')
category_a.store()
category_b = Category(name='Category B')
category_b.store()
formdef2.category_id = category_a.id
formdef2.store()
formdef3 = FormDef()
formdef3.name = 'test 3'
formdef3.category_id = category_b.id
formdef3.store()
formdef3.data_class().wipe()
resp = get_app(pub).get(sign_uri('/api/statistics/'))
form_filter = [x for x in resp.json['data'][0]['filters'] if x['id'] == 'form'][0]
assert form_filter['options'] == [
[None, [{'id': '_all', 'label': 'All Forms'}]],
[
'Category A',
[
{'id': 'category:category-a', 'label': 'All forms of category Category A'},
{'id': 'test-2', 'label': 'test 2'},
],
],
[
'Category B',
[
{'id': 'category:category-b', 'label': 'All forms of category Category B'},
{'id': 'test-3', 'label': 'test 3'},
],
],
['Misc', [{'id': 'test-1', 'label': 'test 1'}]],
]
# check Misc is not shown if all forms have categories
formdef.category_id = category_a.id
formdef.store()
resp = get_app(pub).get(sign_uri('/api/statistics/'))
form_filter = [x for x in resp.json['data'][0]['filters'] if x['id'] == 'form'][0]
assert form_filter['options'] == [
[None, [{'id': '_all', 'label': 'All Forms'}]],
[
'Category A',
[
{'id': 'category:category-a', 'label': 'All forms of category Category A'},
{'id': 'test-1', 'label': 'test 1'},
{'id': 'test-2', 'label': 'test 2'},
],
],
[
'Category B',
[
{'id': 'category:category-b', 'label': 'All forms of category Category B'},
{'id': 'test-3', 'label': 'test 3'},
],
],
]
resp = get_app(pub).get(sign_uri('/api/statistics/'))
form_filter = [x for x in resp.json['data'][0]['filters'] if x['id'] == 'channel'][0]
assert form_filter['options'] == [
{'id': '_all', 'label': 'All'},
{'id': 'mail', 'label': 'Mail'},
{'id': 'email', 'label': 'Email'},
{'id': 'phone', 'label': 'Phone'},
{'id': 'counter', 'label': 'Counter'},
{'id': 'fax', 'label': 'Fax'},
{'id': 'web', 'label': 'Web'},
{'id': 'social-network', 'label': 'Social Network'},
]
def test_statistics_index_cards(pub):
carddef = CardDef()
carddef.name = 'test 1'
carddef.fields = []
carddef.store()
carddef.data_class().wipe()
carddef2 = CardDef()
carddef2.name = 'test 2'
carddef2.fields = []
carddef2.store()
carddef2.data_class().wipe()
resp = get_app(pub).get(sign_uri('/api/statistics/'))
form_filter = [x for x in resp.json['data'][1]['filters'] if x['id'] == 'form'][0]
assert form_filter['options'] == [
{'id': 'test-1', 'label': 'test 1'},
{'id': 'test-2', 'label': 'test 2'},
]
category_a = CardDefCategory(name='Category A')
category_a.store()
category_b = CardDefCategory(name='Category B')
category_b.store()
carddef2.category_id = category_a.id
carddef2.store()
carddef3 = CardDef()
carddef3.name = 'test 3'
carddef3.category_id = category_b.id
carddef3.store()
carddef3.data_class().wipe()
resp = get_app(pub).get(sign_uri('/api/statistics/'))
form_filter = [x for x in resp.json['data'][1]['filters'] if x['id'] == 'form'][0]
assert form_filter['options'] == [
['Category A', [{'id': 'test-2', 'label': 'test 2'}]],
['Category B', [{'id': 'test-3', 'label': 'test 3'}]],
['Misc', [{'id': 'test-1', 'label': 'test 1'}]],
]
def test_statistics_index_resolution_time(pub):
formdef = FormDef()
formdef.name = 'test 1'
formdef.fields = []
formdef.store()
formdef.data_class().wipe()
resp = get_app(pub).get(sign_uri('/api/statistics/'))
resolution_time_stat = [x for x in resp.json['data'] if x['id'] == 'resolution_time'][0]
form_filter = [x for x in resolution_time_stat['filters'] if x['id'] == 'form'][0]
assert form_filter['options'] == [{'id': 'test-1', 'label': 'test 1'}]
def test_statistics_index_resolution_time_cards(pub):
carddef = CardDef()
carddef.name = 'test 1'
carddef.fields = []
carddef.store()
carddef.data_class().wipe()
resp = get_app(pub).get(sign_uri('/api/statistics/'))
resolution_time_stat = [x for x in resp.json['data'] if x['id'] == 'resolution_time_cards'][0]
card_filter = [x for x in resolution_time_stat['filters'] if x['id'] == 'form'][0]
assert card_filter['options'] == [{'id': 'test-1', 'label': 'test 1'}]
def test_statistics_forms_count(pub):
category_a = Category(name='Category A')
category_a.store()
category_b = Category(name='Category B')
category_b.store()
formdef = FormDef()
formdef.name = 'test 1'
formdef.category_id = category_a.id
formdef.fields = []
formdef.store()
formdef.data_class().wipe()
formdef2 = FormDef()
formdef2.name = 'test 2'
formdef2.category_id = category_b.id
formdef2.fields = []
formdef2.store()
formdef2.data_class().wipe()
for i in range(20):
formdata = formdef.data_class()()
formdata.just_created()
formdata.receipt_time = datetime.datetime(2021, 1, 1, 0, 0).timetuple()
# "Web" channel has three equivalent values
if i == 0:
formdata.submission_channel = 'web'
elif i == 1:
formdata.submission_channel = ''
else:
formdata.submission_channel = None
formdata.store()
for _i in range(30):
formdata = formdef2.data_class()()
formdata.just_created()
formdata.receipt_time = datetime.datetime(2021, 3, 1, 2, 0).timetuple()
formdata.submission_channel = 'mail'
formdata.store()
# draft should not be counted
formdata = formdef.data_class()()
formdata.receipt_time = datetime.datetime(2021, 3, 1, 2, 0).timetuple()
formdata.status = 'draft'
formdata.store()
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/'))
assert resp.json == {
'data': {
'series': [{'data': [20, 0, 30], 'label': 'Forms Count'}],
'x_labels': ['2021-01', '2021-02', '2021-03'],
'subfilters': [],
},
'err': 0,
}
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?time_interval=year'))
assert resp.json == {
'data': {
'series': [{'data': [50], 'label': 'Forms Count'}],
'x_labels': ['2021'],
'subfilters': [],
},
'err': 0,
}
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?time_interval=weekday'))
assert resp.json == {
'data': {
'series': [{'data': [30, 0, 0, 0, 20, 0, 0], 'label': 'Forms Count'}],
'x_labels': ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'],
'subfilters': [],
},
'err': 0,
}
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?time_interval=hour'))
assert resp.json == {
'data': {
'series': [
{
'label': 'Forms Count',
'data': [20, 0, 30, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
}
],
'x_labels': list(range(24)),
'subfilters': [],
},
'err': 0,
}
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?time_interval=none'))
assert resp.json == {
'data': {'series': [{'label': 'Forms Count', 'data': [50]}], 'x_labels': [''], 'subfilters': []},
'err': 0,
}
# time_interval=day is not supported
get_app(pub).get(sign_uri('/api/statistics/forms/count/?time_interval=day'), status=400)
# apply category filter through form parameter
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=category:category-a'))
assert resp.json == {
'data': {
'series': [{'data': [20], 'label': 'Forms Count'}],
'x_labels': ['2021-01'],
'subfilters': [],
},
'err': 0,
}
# apply category filter (legacy)
new_resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?category=category-a'))
assert new_resp.json == resp.json
# apply category id filter (legacy)
new_resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?category=%s' % category_a.id))
assert new_resp.json == resp.json
# apply form filter
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=%s' % formdef.url_name))
assert resp.json['data']['series'] == [{'data': [20], 'label': 'Forms Count'}]
assert resp.json['data']['x_labels'] == ['2021-01']
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=%s' % 'invalid'), status=400)
assert resp.text == 'invalid form'
# apply period filter
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?end=2021-02-01'))
assert resp.json == {
'data': {
'series': [{'data': [20], 'label': 'Forms Count'}],
'x_labels': ['2021-01'],
'subfilters': [],
},
'err': 0,
}
# apply channel filter
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?channel=mail'))
assert resp.json['data']['series'] == [{'data': [30], 'label': 'Forms Count'}]
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?channel=web'))
assert resp.json['data']['series'] == [{'data': [20], 'label': 'Forms Count'}]
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?channel=_all'))
assert resp.json['data']['series'] == [{'data': [20, 0, 30], 'label': 'Forms Count'}]
def test_statistics_forms_count_subfilters(pub, formdef):
for i in range(2):
formdata = formdef.data_class()()
formdata.data['2'] = 'foo' if i % 2 else 'baz'
formdata.data['2_display'] = 'Foo' if i % 2 else 'Baz'
formdata.data['3'] = ['foo'] if i % 2 else ['bar', 'baz']
formdata.data['3_display'] = 'Foo' if i % 2 else 'Bar, Baz'
formdata.just_created()
formdata.receipt_time = datetime.datetime(2021, 1, 1, 0, 0).timetuple()
formdata.store()
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=%s' % formdef.url_name))
# check group-by subfilter
assert resp.json['data']['subfilters'][0] == {
'id': 'group-by',
'label': 'Group by',
'options': [
{'id': 'channel', 'label': 'Channel'},
{'id': 'simple-status', 'label': 'Simplified status'},
{'id': 'test-item', 'label': 'Test item'},
{'id': 'test-items', 'label': 'Test items'},
{'id': 'checkbox', 'label': 'Checkbox'},
{'id': 'status', 'label': 'Status'},
],
'has_subfilters': True,
}
# check item field subfilter
assert resp.json['data']['subfilters'][1] == {
'id': 'filter-test-item',
'label': 'Test item',
'options': [{'id': 'baz', 'label': 'Baz'}, {'id': 'foo', 'label': 'Foo'}],
'required': False,
}
# check items field subfilter
assert resp.json['data']['subfilters'][2] == {
'id': 'filter-test-items',
'label': 'Test items',
'options': [
{'id': 'bar', 'label': 'Bar'},
{'id': 'baz', 'label': 'Baz'},
{'id': 'foo', 'label': 'Foo'},
],
'required': False,
}
# check block boolean field subfilter
assert resp.json['data']['subfilters'][3] == {
'id': 'filter-blockdata_bool',
'label': 'Bool',
'options': [{'id': 'true', 'label': 'Yes'}, {'id': 'false', 'label': 'No'}],
'required': False,
}
# check boolean backoffice field subfilter
assert resp.json['data']['subfilters'][4] == {
'id': 'filter-checkbox',
'label': 'Checkbox',
'options': [{'id': 'true', 'label': 'Yes'}, {'id': 'false', 'label': 'No'}],
'required': False,
}
# check status subfilter
assert resp.json['data']['subfilters'][-1] == {
'default': '_all',
'id': 'filter-status',
'label': 'Status',
'options': [
{'id': '_all', 'label': 'All'},
{'id': 'pending', 'label': 'Open'},
{'id': 'done', 'label': 'Done'},
{'id': '1', 'label': 'New status'},
{'id': '2', 'label': 'End status'},
],
'required': True,
}
# group by triggers new subfilter
new_resp = get_app(pub).get(
sign_uri('/api/statistics/forms/count/?form=%s&group-by=test-item' % formdef.url_name)
)
assert new_resp.json['data']['subfilters'][1] == {
'id': 'hide_none_label',
'label': 'Ignore forms where "Test item" is empty.',
'options': [{'id': 'true', 'label': 'Yes'}, {'id': 'false', 'label': 'No'}],
'required': True,
'default': 'false',
}
assert len(new_resp.json['data']['subfilters']) == len(resp.json['data']['subfilters']) + 1
# add item field with no formdata, it should not appear
item_field = fields.ItemField(
id='20',
varname='test-item-no-formdata',
label='Test item no formdata',
type='item',
items=['foo', 'bar', 'baz'],
display_locations=['statistics'],
)
formdef.fields.append(item_field)
formdef.store()
new_resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=%s' % formdef.url_name))
assert new_resp.json == resp.json
# add boolean field with no varname, it should not appear
bool_field = fields.BoolField(id='21', label='Checkbox', type='bool', display_locations=['statistics'])
formdef.fields.append(bool_field)
formdef.store()
new_resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=%s' % formdef.url_name))
assert new_resp.json == resp.json
# add boolean field with no display location, it should not appear
bool_field = fields.BoolField(
id='22', varname='checkbox', label='Checkbox', type='bool', display_locations=['validation']
)
formdef.fields.append(bool_field)
formdef.store()
new_resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=%s' % formdef.url_name))
assert new_resp.json == resp.json
# add not filterable field, it should not appear
formdef.fields.append(fields.StringField(id='23', varname='test string', label='Test', type='string'))
formdef.store()
new_resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=%s' % formdef.url_name))
assert new_resp.json == resp.json
# add items field inside block field, it should not appear
items_field = fields.ItemsField(
id='2',
varname='items',
label='Block items',
type='items',
items=['foo', 'bar', 'baz'],
anonymise=False,
display_locations=['statistics'],
)
formdef.fields[2].block.fields.append(bool_field)
formdef.store()
formdata.data['4'] = {'data': [{'2': ['bar']}]}
formdata.store()
new_resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=%s' % formdef.url_name))
assert new_resp.json == resp.json
# remove fields and statuses
workflow = Workflow(name='Empty wf')
workflow.store()
formdef.workflow = workflow
formdef.fields.clear()
formdef.store()
formdef.data_class().wipe()
resp = get_app(pub).get(sign_uri('/api/statistics/forms/count/?form=%s' % formdef.url_name))
assert resp.json['data'] == {
'series': [{'data': [], 'label': 'Forms Count'}],
'subfilters': [],
'x_labels': [],
}
def test_statistics_forms_count_subfilters_query(pub, formdef):
for i in range(20):
formdata = formdef.data_class()()
formdata.just_created()
if i % 3:
formdata.data['1'] = True
formdata.data['2'] = 'foo'
formdata.data['3'] = ['bar', 'baz']
formdata.data['4'] = {'data': [{'1': True}, {'1': False}]}
elif i % 2:
formdata.data['1'] = False
formdata.data['2'] = 'baz'
formdata.data['3'] = ['baz']
formdata.data['4'] = {'data': [{'1': False}]}
formdata.jump_status('2')
formdata.receipt_time = datetime.datetime(2021, 1, 1, 0, 0).timetuple()
formdata.store()
# query all formdata
url = '/api/statistics/forms/count/?form=%s' % formdef.url_name
resp = get_app(pub).get(sign_uri(url))
assert resp.json['data']['series'][0]['data'][0] == 20
# filter on boolean field
resp = get_app(pub).get(sign_uri(url + '&filter-checkbox=true'))
assert resp.json['data']['series'][0]['data'][0] == 13
resp = get_app(pub).get(sign_uri(url + '&filter-checkbox=false'))
assert resp.json['data']['series'][0]['data'][0] == 3
resp = get_app(pub).get(sign_uri(url + '&filter-checkbox='))
assert resp.json['data']['series'][0]['data'][0] == 20
resp = get_app(pub).get(sign_uri(url + '&filter-checkbox=xxx'), status=400)
assert resp.text == 'Invalid value "xxx" for "filter-checkbox"'
# filter on item field
resp = get_app(pub).get(sign_uri(url + '&filter-test-item=foo'))
assert resp.json['data']['series'][0]['data'][0] == 13
resp = get_app(pub).get(sign_uri(url + '&filter-test-item=baz'))
assert resp.json['data']['series'][0]['data'][0] == 3
resp = get_app(pub).get(sign_uri(url + '&filter-test-item=bar'))
assert resp.json['data']['series'][0]['data'] == []
resp = get_app(pub).get(sign_uri(url + '&filter-test-item='))
assert resp.json['data']['series'][0]['data'][0] == 20
resp = get_app(pub).get(sign_uri(url + '&filter-test-item=xxx'))
assert resp.json['data']['series'][0]['data'] == []
# filter on items field
resp = get_app(pub).get(sign_uri(url + '&filter-test-items=foo'))
assert resp.json['data']['series'][0]['data'] == []
resp = get_app(pub).get(sign_uri(url + '&filter-test-items=bar'))
assert resp.json['data']['series'][0]['data'][0] == 13
resp = get_app(pub).get(sign_uri(url + '&filter-test-items=baz'))
assert resp.json['data']['series'][0]['data'][0] == 16
# filter on block boolean field
resp = get_app(pub).get(sign_uri(url + '&filter-blockdata_bool=true'))
assert resp.json['data']['series'][0]['data'][0] == 13
resp = get_app(pub).get(sign_uri(url + '&filter-blockdata_bool=false'))
assert resp.json['data']['series'][0]['data'][0] == 16
# filter on status
resp = get_app(pub).get(sign_uri(url + '&filter-status=_all'))
assert resp.json['data']['series'][0]['data'][0] == 20
resp = get_app(pub).get(sign_uri(url + '&filter-status=1'))
assert resp.json['data']['series'][0]['data'][0] == 17
resp = get_app(pub).get(sign_uri(url + '&filter-status=pending'))
assert resp.json['data']['series'][0]['data'][0] == 17
resp = get_app(pub).get(sign_uri(url + '&filter-status=2'))
assert resp.json['data']['series'][0]['data'][0] == 3
resp = get_app(pub).get(sign_uri(url + '&filter-status=done'))
assert resp.json['data']['series'][0]['data'][0] == 3
resp = get_app(pub).get(sign_uri(url + '&filter-status='))
assert resp.json['data']['series'][0]['data'][0] == 20
resp = get_app(pub).get(sign_uri(url + '&filter-status=xxx'))
assert resp.json['data']['series'][0]['data'][0] == 20
# invalid filter
resp = get_app(pub).get(sign_uri(url + '&filter-xxx=yyy'))
assert resp.json['data']['series'][0]['data'] == []
def test_statistics_forms_count_subfilters_query_same_varname(pub, formdef):
formdef = FormDef()
formdef.name = 'test'
formdef.fields = [
fields.ItemField(id='1', varname='test', label='Test', type='item', items=['foo', 'bar']),
fields.ItemField(
id='2',
varname='test',
label='Test',
type='item',
items=['foo', 'bar'],
display_locations=['statistics'],
),
]
formdef.store()
formdatas = []
for i in range(5):
formdata = formdef.data_class()()
formdata.just_created()
formdata.receipt_time = datetime.datetime(2021, 1, 1, 0, 0).timetuple()
if i == 0:
formdata.data['1'] = 'foo'
if i == 1:
formdata.data['1'] = 'bar'
formdata.data['2'] = 'foo'
formdata.store()
formdatas.append(formdata)
url = '/api/statistics/forms/count/?form=%s' % formdef.url_name
resp = get_app(pub).get(sign_uri(url + '&filter-test=foo'))
assert resp.json['data']['series'] == [{'data': [5], 'label': 'Forms Count'}]
formdef.fields[0].display_locations = ['statistics']
formdef.store()
for formdata in formdatas:
formdata.store() # refresh statistics_data column
# first non empty value is used : 4 are 'foo' and one is 'bar' hence 4 results
resp = get_app(pub).get(sign_uri(url + '&filter-test=foo'))
assert resp.json['data']['series'] == [{'data': [4], 'label': 'Forms Count'}]
resp = get_app(pub).get(sign_uri(url + '&filter-test=bar'))
assert resp.json['data']['series'] == [{'data': [1], 'label': 'Forms Count'}]
def test_statistics_forms_count_subfilters_query_integer_items(pub, formdef):
for i in range(10):
formdata = formdef.data_class()()
formdata.just_created()
if i % 2:
formdata.data['3'] = ['1', '2']
else:
formdata.data['3'] = ['1']
formdata.receipt_time = datetime.datetime(2021, 1, 1, 0, 0).timetuple()
formdata.store()
url = '/api/statistics/forms/count/?form=%s' % formdef.url_name
resp = get_app(pub).get(sign_uri(url + '&filter-test-items=1'))
assert resp.json['data']['series'][0]['data'][0] == 10
resp = get_app(pub).get(sign_uri(url + '&filter-test-items=2'))
assert resp.json['data']['series'][0]['data'][0] == 5
@pytest.mark.parametrize('anonymise', [False, True])
def test_statistics_forms_count_group_by(pub, formdef, anonymise):
for i in range(20):
formdata = formdef.data_class()()
formdata.just_created()
formdata.receipt_time = datetime.datetime(2021, 1, 1, 0, 0).timetuple()
if i % 3:
formdata.data['1'] = True
formdata.data['2'] = 'foo'
formdata.data['2_display'] = 'Foo'
formdata.data['3'] = ['bar', 'baz']
formdata.data['3_display'] = 'Bar, Baz'
# "Web" channel has three equivalent values
if i == 1:
formdata.submission_channel = 'web'
elif i == 2:
formdata.submission_channel = ''
else:
formdata.submission_channel = None
elif i % 2:
formdata.data['1'] = False
formdata.data['2'] = 'baz'
formdata.data['3'] = ['baz']
if i == 3:
formdata.jump_status('3')
elif i == 9:
formdata.jump_status('3')
formdata.jump_status('4')
else:
formdata.jump_status('2')
formdata.submission_channel = 'mail'
else:
formdata.receipt_time = datetime.datetime(2021, 3, 1, 2, 0).timetuple()
formdata.store()
if anonymise:
formdata.anonymise()
# group by item field
url = '/api/statistics/forms/count/?form=%s' % formdef.url_name
resp = get_app(pub).get(sign_uri(url + '&group-by=test-item'))
assert resp.json['data']['x_labels'] == ['2021-01', '2021-02', '2021-03']
assert resp.json['data']['series'] == [
{'data': [13, None, None], 'label': 'Foo'},
{'data': [3, None, None], 'label': 'baz'},
{'data': [None, None, 4], 'label': 'None'},
]
resp = get_app(pub).get(sign_uri(url + '&group-by=test-item&time_interval=year'))
assert resp.json['data']['x_labels'] == ['2021']
assert resp.json['data']['series'] == [
{'label': 'Foo', 'data': [13]},
{'label': 'baz', 'data': [3]},
{'label': 'None', 'data': [4]},
]
resp = get_app(pub).get(sign_uri(url + '&group-by=test-item&time_interval=hour'))
assert resp.json['data']['x_labels'] == list(range(24))
assert resp.json['data']['series'][0]['data'][0] == 13
assert resp.json['data']['series'][1]['data'][0] == 3
assert resp.json['data']['series'][2]['data'][2] == 4
resp = get_app(pub).get(sign_uri(url + '&group-by=test-item&time_interval=weekday'))
assert len(resp.json['data']['x_labels']) == 7
assert resp.json['data']['series'] == [
{'label': 'Foo', 'data': [None, None, None, None, 13, None, None]},
{'label': 'baz', 'data': [None, None, None, None, 3, None, None]},
{'label': 'None', 'data': [4, None, None, None, None, None, None]},
]
# hide None label
resp = get_app(pub).get(sign_uri(url + '&group-by=test-item&hide_none_label=true'))
assert resp.json['data']['x_labels'] == ['2021-01']
assert resp.json['data']['series'] == [
{'data': [13], 'label': 'Foo'},
{'data': [3], 'label': 'baz'},
]
# group by items field
url = '/api/statistics/forms/count/?form=%s' % formdef.url_name
resp = get_app(pub).get(sign_uri(url + '&group-by=test-items'))
assert resp.json['data']['x_labels'] == ['2021-01', '2021-02', '2021-03']
assert resp.json['data']['series'] == [
{'label': 'Bar', 'data': [13, None, None]},
{'label': 'Baz', 'data': [16, None, None]},
{'label': 'None', 'data': [None, None, 4]},
]
# group by boolean field
resp = get_app(pub).get(sign_uri(url + '&group-by=checkbox'))
assert resp.json['data']['x_labels'] == ['2021-01', '2021-02', '2021-03']
assert resp.json['data']['series'] == [
{'data': [13, None, None], 'label': 'Yes'},
{'data': [3, None, None], 'label': 'No'},
{'data': [None, None, 4], 'label': 'None'},
]
# group by status
resp = get_app(pub).get(sign_uri(url + '&group-by=status'))
assert resp.json['data']['x_labels'] == ['2021-01', '2021-02', '2021-03']
assert resp.json['data']['series'] == [
{'label': 'New status', 'data': [13, None, 4]},
{'label': 'End status', 'data': [1, None, None]},
{'label': 'Middle status 1', 'data': [1, None, None]},
{'label': 'Middle status 2', 'data': [1, None, None]},
]
# group by simplified status
resp = get_app(pub).get(sign_uri(url + '&group-by=simple-status'))
assert resp.json['data']['x_labels'] == ['2021-01', '2021-02', '2021-03']
assert resp.json['data']['series'] == [
{'label': 'New', 'data': [13, None, 4]},
{'label': 'Done', 'data': [1, None, None]},
{'label': 'In progress', 'data': [2, None, None]},
]
# group by channel
resp = get_app(pub).get(sign_uri(url + '&group-by=channel'))
assert resp.json['data']['x_labels'] == ['2021-01', '2021-02', '2021-03']
assert resp.json['data']['series'] == [
{'data': [3, None, None], 'label': 'Mail'},
{'data': [13, None, 4], 'label': 'Web'},
]
# group by item field without time interval
resp = get_app(pub).get(sign_uri(url + '&group-by=test-item&time_interval=none'))
# Foo is first because it has a display value, baz is second because it has not, None is always last
assert resp.json['data']['x_labels'] == ['Foo', 'baz', 'None']
assert resp.json['data']['series'] == [{'data': [13, 3, 4], 'label': 'Forms Count'}]
# group by items field without time interval
resp = get_app(pub).get(sign_uri(url + '&group-by=test-items&time_interval=none'))
assert resp.json['data']['x_labels'] == ['Bar', 'Baz', 'None']
assert resp.json['data']['series'] == [{'label': 'Forms Count', 'data': [13, 16, 4]}]
# group by submission channel without time interval
resp = get_app(pub).get(sign_uri(url + '&group-by=channel&time_interval=none'))
assert resp.json['data']['x_labels'] == ['Mail', 'Web']
assert resp.json['data']['series'] == [{'data': [3, 17], 'label': 'Forms Count'}]
# group by status without time interval
resp = get_app(pub).get(sign_uri(url + '&group-by=status&time_interval=none'))
assert resp.json['data']['x_labels'] == ['New status', 'End status', 'Middle status 1', 'Middle status 2']
assert resp.json['data']['series'] == [{'data': [17, 1, 1, 1], 'label': 'Forms Count'}]
# group by simplfified status without time interval
resp = get_app(pub).get(sign_uri(url + '&group-by=simple-status&time_interval=none'))
assert resp.json['data']['x_labels'] == ['New', 'Done', 'In progress']
assert resp.json['data']['series'] == [{'label': 'Forms Count', 'data': [17, 1, 2]}]
# check statuses order
formdef.workflow.possible_status = list(reversed(formdef.workflow.possible_status))
formdef.workflow.store()
resp = get_app(pub).get(sign_uri(url + '&group-by=status&time_interval=none'))
assert resp.json['data']['x_labels'] == ['Middle status 2', 'Middle status 1', 'End status', 'New status']
assert resp.json['data']['series'] == [{'data': [1, 1, 1, 17], 'label': 'Forms Count'}]
# group by on block field is not supported
resp = get_app(pub).get(sign_uri(url + '&group-by=blockdata_bool'))
assert resp.json['data']['series'] == [{'data': [16, 0, 4], 'label': 'Forms Count'}]
# invalid field
resp = get_app(pub).get(sign_uri(url + '&group-by=xxx'))
assert resp.json['data']['series'] == [{'data': [16, 0, 4], 'label': 'Forms Count'}]
def test_statistics_forms_count_group_by_same_varname(pub, formdef):
formdef = FormDef()
formdef.name = 'test'
formdef.fields = [
fields.ItemField(id='1', varname='test', label='Test', type='item', items=['foo']),
fields.ItemField(
id='2', varname='test', label='Test', type='item', items=['bar'], display_locations=['statistics']
),
]
formdef.store()
formdata = formdef.data_class()()
formdata.just_created()
formdata.receipt_time = datetime.datetime(2021, 1, 1, 0, 0).timetuple()
formdata.data['1'] = 'foo'
formdata.data['2'] = 'bar'
formdata.store()
url = '/api/statistics/forms/count/?form=%s' % formdef.url_name
resp = get_app(pub).get(sign_uri(url + '&group-by=test'))
assert resp.json['data']['series'] == [{'data': [1], 'label': 'bar'}]
formdef.fields[0].display_locations = ['statistics']
formdef.store()
formdata.store() # refresh statistics_data column
# group by uses first field marked for statistics
resp = get_app(pub).get(sign_uri(url + '&group-by=test'))
assert resp.json['data']['series'] == [{'data': [1], 'label': 'foo'}]
formdata = formdef.data_class()()
formdata.just_created()
formdata.receipt_time = datetime.datetime(2021, 1, 1, 0, 0).timetuple()
formdata.data['2'] = 'foo'
formdata.store()
resp = get_app(pub).get(sign_uri(url + '&group-by=test'))
assert resp.json['data']['series'] == [{'data': [2], 'label': 'foo'}]
def test_statistics_cards_count(pub):
carddef = CardDef()
carddef.name = 'test 1'
carddef.fields = []
carddef.store()
carddef.data_class().wipe()
for _i in range(20):
carddata = carddef.data_class()()
carddata.just_created()
carddata.receipt_time = datetime.datetime(2021, 1, 1, 0, 0).timetuple()
carddata.store()
# apply (required) card filter
resp = get_app(pub).get(sign_uri('/api/statistics/cards/count/?form=%s' % carddef.url_name))
assert resp.json['data']['series'] == [{'data': [20], 'label': 'Cards Count'}]
assert resp.json['data']['x_labels'] == ['2021-01']
resp = get_app(pub).get(sign_uri('/api/statistics/cards/count/?card=%s' % 'invalid'), status=400)
assert resp.text == 'invalid form'
def test_statistics_resolution_time(pub, freezer):
workflow = Workflow(name='Workflow One')
new_status = workflow.add_status(name='New status')
middle_status = workflow.add_status(name='Middle status')
workflow.add_status(name='End status')
workflow.add_status(name='End status 2')
# add jump from new to end
jump = new_status.add_action('jump', id='_jump')
jump.status = '3'
# add jump form new to middle and from middle to end 2
jump = new_status.add_action('jump', id='_jump')
jump.status = '2'
jump = middle_status.add_action('jump', id='_jump')
jump.status = '4'
workflow.store()
formdef = FormDef()
formdef.name = 'test'
formdef.workflow_id = workflow.id
formdef.store()
freezer.move_to(datetime.date(2021, 1, 1))
formdata_list = []
for i in range(3):
formdata = formdef.data_class()()
formdata.just_created()
formdata_list.append(formdata)
# one formdata resolved in one day
freezer.move_to(datetime.date(2021, 1, 2))
formdata_list[0].jump_status('3')
formdata_list[0].store()
# one formdata resolved in two days, passing by middle status
formdata_list[1].jump_status('2')
freezer.move_to(datetime.date(2021, 1, 3))
formdata_list[1].jump_status('4')
formdata_list[1].store()
# one formdata blocked in middle status for three days
freezer.move_to(datetime.date(2021, 1, 4))
formdata_list[2].jump_status('2')
formdata_list[2].store()
# by default, count forms between initial status and final statuses
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test'))
assert resp.json['data'] == {
'series': [
{
'data': [86400.0, 172800.0, 129600.0, 129600.0],
'label': 'Time between two statuses',
}
],
'subfilters': [
{
'id': 'start_status',
'label': 'Start status',
'options': [
{'id': '1', 'label': 'New status'},
{'id': '2', 'label': 'Middle status'},
{'id': '3', 'label': 'End status'},
{'id': '4', 'label': 'End status 2'},
],
'required': True,
'default': '1',
},
{
'default': 'done',
'id': 'end_status',
'label': 'End status',
'options': [
{'id': 'done', 'label': 'Any final status'},
{'id': '2', 'label': 'Middle status'},
{'id': '3', 'label': 'End status'},
{'id': '4', 'label': 'End status 2'},
],
'required': True,
},
],
'x_labels': ['Minimum time', 'Maximum time', 'Mean', 'Median'],
}
assert get_humanized_duration_serie(resp.json) == [
'1 day(s) and 0 hour(s)',
'2 day(s) and 0 hour(s)',
'1 day(s) and 12 hour(s)',
'1 day(s) and 12 hour(s)',
]
# specify end status
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test&end_status=3'))
assert get_humanized_duration_serie(resp.json) == [
'1 day(s) and 0 hour(s)',
'1 day(s) and 0 hour(s)',
'1 day(s) and 0 hour(s)',
'1 day(s) and 0 hour(s)',
]
# specify start status
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test&start_status=2'))
assert get_humanized_duration_serie(resp.json) == [
'1 day(s) and 0 hour(s)',
'1 day(s) and 0 hour(s)',
'1 day(s) and 0 hour(s)',
'1 day(s) and 0 hour(s)',
]
# specify start and end statuses
resp = get_app(pub).get(
sign_uri('/api/statistics/resolution-time/?form=test&start_status=2&end_status=4')
)
assert get_humanized_duration_serie(resp.json) == [
'1 day(s) and 0 hour(s)',
'1 day(s) and 0 hour(s)',
'1 day(s) and 0 hour(s)',
'1 day(s) and 0 hour(s)',
]
resp = get_app(pub).get(
sign_uri('/api/statistics/resolution-time/?form=test&start_status=1&end_status=2')
)
assert get_humanized_duration_serie(resp.json) == [
'1 day(s) and 0 hour(s)',
'3 day(s) and 0 hour(s)',
'2 day(s) and 0 hour(s)',
'2 day(s) and 0 hour(s)',
]
# unknown statuses
default_resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test'))
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test&start_status=42'))
assert resp.json == default_resp.json
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test&end_status=42'))
assert resp.json == default_resp.json
# specify start and end statuses which does not match any formdata
resp = get_app(pub).get(
sign_uri('/api/statistics/resolution-time/?form=test&start_status=2&end_status=3')
)
assert resp.json['data']['series'][0]['data'] == []
# unknown form
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=xxx'), status=400)
def test_statistics_resolution_time_median(pub, freezer):
workflow = Workflow(name='Workflow One')
new_status = workflow.add_status(name='New status')
workflow.add_status(name='End status')
jump = new_status.add_action('jump', id='_jump')
jump.status = '2'
workflow.store()
formdef = FormDef()
formdef.name = 'test'
formdef.workflow_id = workflow.id
formdef.store()
for i in range(2, 11):
formdata = formdef.data_class()()
freezer.move_to(datetime.date(2021, 1, 1))
formdata.just_created()
if i != 10:
# add lots of formdata resolved in a few days
freezer.move_to(datetime.date(2021, 1, i))
else:
# one formdata took 3 months
freezer.move_to(datetime.date(2021, 4, 1))
formdata.jump_status('2')
formdata.store()
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test'))
assert get_humanized_duration_serie(resp.json) == [
'1 day(s) and 0 hour(s)', # min
'89 day(s) and 22 hour(s)', # max
'13 day(s) and 23 hour(s)', # mean
'5 day(s) and 0 hour(s)', # median
]
def test_statistics_resolution_time_start_end_filter(pub, freezer):
workflow = Workflow(name='Workflow One')
new_status = workflow.add_status(name='New status')
workflow.add_status(name='End status')
jump = new_status.add_action('jump', id='_jump')
jump.status = '2'
workflow.store()
formdef = FormDef()
formdef.name = 'test'
formdef.workflow_id = workflow.id
formdef.store()
# create formdata, the latest being the longest to resolve
for i in range(1, 10):
formdata = formdef.data_class()()
freezer.move_to(datetime.date(2021, 1, i))
formdata.just_created()
freezer.move_to(datetime.date(2021, 1, i * 2))
formdata.jump_status('2')
formdata.store()
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test'))
assert get_humanized_duration_serie(resp.json) == [
'1 day(s) and 0 hour(s)', # min
'9 day(s) and 0 hour(s)', # max
'5 day(s) and 0 hour(s)', # mean
'5 day(s) and 0 hour(s)', # median
]
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test&start=2021-01-05'))
assert get_humanized_duration_serie(resp.json) == [
'5 day(s) and 0 hour(s)', # min
'9 day(s) and 0 hour(s)', # max
'7 day(s) and 0 hour(s)', # mean
'7 day(s) and 0 hour(s)', # median
]
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time/?form=test&end=2021-01-05'))
assert get_humanized_duration_serie(resp.json) == [
'1 day(s) and 0 hour(s)', # min
'4 day(s) and 0 hour(s)', # max
'2 day(s) and 12 hour(s)', # mean
'2 day(s) and 12 hour(s)', # median
]
resp = get_app(pub).get(
sign_uri('/api/statistics/resolution-time/?form=test&start=2021-01-04&end=2021-01-05')
)
assert get_humanized_duration_serie(resp.json) == [
'4 day(s) and 0 hour(s)', # min
'4 day(s) and 0 hour(s)', # max
'4 day(s) and 0 hour(s)', # mean
'4 day(s) and 0 hour(s)', # median
]
def test_statistics_resolution_time_cards(pub, freezer):
workflow = Workflow(name='Workflow One')
new_status = workflow.add_status(name='New status')
workflow.add_status(name='End status')
jump = new_status.add_action('jump', id='_jump')
jump.status = '2'
workflow.store()
carddef = CardDef()
carddef.name = 'test'
carddef.workflow_id = workflow.id
carddef.store()
for i in range(1, 10):
carddata = carddef.data_class()()
freezer.move_to(datetime.date(2021, 1, i))
carddata.just_created()
freezer.move_to(datetime.date(2021, 1, i * 2))
carddata.jump_status('2')
carddata.store()
resp = get_app(pub).get(sign_uri('/api/statistics/resolution-time-cards/?form=test'))
assert get_humanized_duration_serie(resp.json) == [
'1 day(s) and 0 hour(s)',
'9 day(s) and 0 hour(s)',
'5 day(s) and 0 hour(s)',
'5 day(s) and 0 hour(s)',
]