0001-statistics-filter-forms-count-by-fields-60777.patch

Valentin Deniaud, 03 février 2022 18:02

Voir les différences: en ligne côte à côte

Subject: [PATCH 1/2] statistics: filter forms count by fields (#60777)

 tests/admin_pages/test_form.py |  25 +++
 tests/api/test_statistics.py   | 298 +++++++++++++++++++++++++++++++++
 wcs/backoffice/management.py   |   7 +-
 wcs/fields.py                  |  15 +-
 wcs/statistics/views.py        | 105 +++++++++++-
 5 files changed, 444 insertions(+), 6 deletions(-)

+        ]
         resp = resp.click('edit page fields', index=0)
         assert '<h2>form title - page 1 - first page</h2>' in resp.text
     def test_field_display_locations_statistics_choice(pub):
         create_superuser(pub)
         create_role(pub)
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'form title'
         formdef.fields = [
             fields.StringField(id='1', label='String field', varname='1'),
             fields.ItemField(id='2', label='Item field', type='item', varname='2'),
             fields.ItemsField(id='3', label='Items field', type='items', varname='3'),
             fields.BoolField(id='4', label='Bool field', type='bool', varname='4'),
+        ]
         formdef.store()
         formdef.data_class().wipe()
         app = login(get_app(pub))
         resp = app.get('/backoffice/forms/%s/fields/1/' % formdef.id)
         assert 'Statistics' not in resp.text
         for i in range(2, 5):
             resp = app.get('/backoffice/forms/%s/fields/%s/' % (formdef.id, i))
             assert 'Statistics' in resp.text

     import pytest
     from wcs import fields
     from wcs.blocks import BlockDef
     from wcs.categories import Category
     from wcs.formdef import FormDef
     from wcs.qommon.http_request import HTTPRequest
     from wcs.wf.jump import JumpWorkflowStatusItem
     from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef
     from ..utilities import clean_temporary_pub, create_temporary_pub, get_app
     from .utils import sign_uri
-...
     @pytest.fixture
     def pub():
         pub = create_temporary_pub(sql_mode=True)
         BlockDef.wipe()
         Category.wipe()
         FormDef.wipe()
         Workflow.wipe()
         req = HTTPRequest(None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net'})
         pub.set_app_dir(req)
-...
         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')
         jump = JumpWorkflowStatusItem()
         jump.id = '_jump'
         jump.status = '2'
         jump.timeout = 86400
         new_status.items.append(jump)
         jump.parent = new_status
         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']
+        )
         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()
-...
         assert len(category_filter['options']) == 3
     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'},
             {'id': 'test-1', 'label': 'test 1'},
             {'id': 'test-2', 'label': 'test 2'},
+        ]
     def test_statistics_forms_count(pub):
         category_a = Category(name='Category A')
         category_a.store()
-...
             'data': {
                 'series': [{'data': [20, 0, 30], 'label': 'Forms Count'}],
                 'x_labels': ['2021-01', '2021-02', '2021-03'],
                 'subfilters': [],
             },
             'err': 0,
+        }
-...
             'data': {
                 'series': [{'data': [50], 'label': 'Forms Count'}],
                 'x_labels': ['2021'],
                 'subfilters': [],
             },
             'err': 0,
+        }
-...
             '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,
+        }
-...
+                    }
                 ],
                 'x_labels': list(range(24)),
                 'subfilters': [],
             },
             'err': 0,
+        }
-...
             'data': {
                 'series': [{'data': [20], 'label': 'Forms Count'}],
                 'x_labels': ['2021-01'],
                 'subfilters': [],
             },
             'err': 0,
+        }
         # 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,
+        }
     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 item field subfilter
         assert resp.json['data']['subfilters'][0] == {
             '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'][1] == {
             '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'][2] == {
             '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'][3] == {
             '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,
+        }
         # 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
         # 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}]}
             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] == 3
         # 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'] == []

                 return ('start', 'end')
             return ()
         def get_item_filter_options(self, filter_field, selected_filter, criterias):
         def get_item_filter_options(self, filter_field, selected_filter, criterias=None):
             criterias = (criterias or [])[:]
             # remove potential filter on self (Equal for item, Intersects for items)
             criterias = [
-...
                     field.has_relations = True
                     yield RelatedField(carddef, card_field, field)
             yield FakeField('status', 'status', _('Status'))
             yield FakeField('status', 'status', _('Status'), include_in_statistics=True)
             yield FakeField('anonymised', 'anonymised', _('Anonymised'))
         def get_default_columns(self):
-...
     class FakeField:
         def __init__(self, id, type_, label, addable=True):
         def __init__(self, id, type_, label, addable=True, include_in_statistics=False):
             self.id = id
             self.contextual_id = self.id
             self.type = type_
-...
             self.contextual_varname = self.varname
             self.store_display_value = None
             self.addable = addable
             self.include_in_statistics = include_in_statistics
         def get_view_value(self, value):
             # just here to quack like a duck

         convert_value_to_str = None
         convert_value_from_anything = None
         allow_complex = False
         allow_statistics = False
         display_locations = []
         prefill = None
         keep_raw_value = True
-...
         def include_in_summary_page(self):
             return 'summary' in (self.display_locations or [])
         @property
         def include_in_statistics(self):
             return self.varname and 'statistics' in (self.display_locations or [])
         @property
         def unhtmled_label(self):
             return force_str(html.unescape(force_text(re.sub('<.*?>', ' ', self.label or ''))).strip())
-...
                     widget.extra_css_class = self.extra_css_class
         def get_display_locations_options(self):
             return [
             options = [
                 ('validation', _('Validation Page')),
                 ('summary', _('Summary Page')),
                 ('listings', _('Management Listings')),
+            ]
             if self.allow_statistics and self.varname:
                 options.append(('statistics', _('Statistics')))
             return options
         def fill_admin_form(self, form):
             form.add(StringWidget, 'label', title=_('Label'), value=self.label, required=True, size=50)
             form.add(CheckboxWidget, 'required', title=_('Required'), value=self.required)
-...
         key = 'bool'
         description = _('Check Box (single choice)')
         allow_complex = True
         allow_statistics = True
         widget_class = CheckboxWidget
         required = False
-...
         key = 'item'
         description = _('List')
         allow_complex = True
         allow_statistics = True
         items = []
         show_as_radio = None
-...
         key = 'items'
         description = _('Multiple choice list')
         allow_complex = True
         allow_statistics = True
         items = []
         min_choices = 0

     from quixote import get_publisher
     from wcs.api_utils import is_url_signed
     from wcs.backoffice.management import FormPage
     from wcs.categories import Category
     from wcs.formdef import FormDef
     from wcs.qommon import _, misc
     from wcs.qommon.misc import C_
     from wcs.qommon.storage import Equal
     from wcs.qommon.storage import Equal, NotEqual, Or
     class RestrictedView(View):
-...
             category_options = [{'id': '_all', 'label': C_('categories|All')}] + [
                 {'id': x.id, 'label': x.name} for x in categories
+            ]
             forms = FormDef.select()
             forms.sort(key=lambda x: misc.simplify(x.name))
             form_options = [{'id': '_all', 'label': _('All')}] + [
                 {'id': x.url_name, 'label': x.name} for x in forms
+            ]
             return JsonResponse(
+                {
                     'data': [
-...
                                     'required': True,
                                     'default': '_all',
                                 },
+                                {
                                     'id': 'form',
                                     'label': _('Form'),
                                     'options': form_options,
                                     'required': True,
                                     'default': '_all',
                                 },
                             ],
+                        }
+                    ]
-...
                 'criterias': [],
+            }
             category_id = request.GET.get('category', '_all')
             if category_id != '_all':
             formdef_slug = request.GET.get('form', '_all')
             subfilters = []
             if formdef_slug != '_all':
                 try:
                     formdef = FormDef.get_by_urlname(formdef_slug, ignore_migration=True)
                 except KeyError:
                     return HttpResponseBadRequest('invalid form')
                 form_page = FormPage(formdef=formdef, update_breadcrumbs=False)
                 totals_kwargs['criterias'].append(Equal('formdef_id', formdef.id))
                 totals_kwargs['criterias'].extend(self.get_filters_criterias(formdef, form_page))
                 subfilters = self.get_subfilters(form_page)
             elif category_id != '_all':
                 totals_kwargs['criterias'].append(Equal('category_id', category_id))
             time_interval_methods = {
                 'month': sql.get_monthly_totals,
                 'year': sql.get_yearly_totals,
-...
                                 'data': [x[1] for x in totals],
+                            }
                         ],
                         'subfilters': subfilters,
                     },
                     'err': 0,
+                }
+            )
         def get_filters_criterias(self, formdef, form_page):
             criterias = form_page.get_criterias_from_query()
             selected_status = self.request.GET.get('filter-status')
             applied_filters = None
             if selected_status and selected_status != '_all':
                 if selected_status == 'pending':
                     applied_filters = ['wf-%s' % x.id for x in formdef.workflow.get_not_endpoint_status()]
                 elif selected_status == 'done':
                     applied_filters = ['wf-%s' % x.id for x in formdef.workflow.get_endpoint_status()]
                 else:
                     try:
                         formdef.workflow.get_status(selected_status)
                         applied_filters = ['wf-%s' % selected_status]
                     except KeyError:
                         pass
             if applied_filters:
                 criterias.append(Or([Equal('status', x) for x in applied_filters]))
             else:
                 criterias = [NotEqual('status', 'draft')] + criterias
             return criterias
         @staticmethod
         def get_subfilters(form_page):
             subfilters = []
             for field in form_page.get_formdef_fields():
                 if not getattr(field, 'include_in_statistics', False) or not field.contextual_varname:
                     continue
                 field_key = 'filter-%s' % field.contextual_varname
                 field.required = False
                 if field.type == 'status':
                     waitpoint_status = form_page.formdef.workflow.get_waitpoint_status()
                     if not waitpoint_status:
                         continue
                     field.required = True
                     field.default_filter_value = '_all'
                     options = [
                         ('_all', _('All')),
                         ('pending', C_('statistics|Open')),
                         ('done', _('Done')),
+                    ]
                     for status in waitpoint_status:
                         options.append((status.id, status.name))
                 elif field.type in ('item', 'items'):
                     if not get_publisher().is_using_postgresql():
                         continue
                     options = form_page.get_item_filter_options(field, selected_filter='all')
                     if not options:
                         continue
                 elif field.type == 'bool':
                     options = [('true', _('Yes')), ('false', _('No'))]
                 else:
                     continue
                 filter_description = {
                     'id': field_key,
                     'label': field.label,
                     'options': [{'id': x[0], 'label': x[1]} for x in options],
                     'required': field.required,
+                }
                 if hasattr(field, 'default_filter_value'):
                     filter_description['default'] = field.default_filter_value
                 subfilters.append(filter_description)
             return subfilters
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-statistics-filter-forms-count-by-fields-60777.patch