api: add status filter to /api/user/forms (#56518)

tests/api/test_user.py

@@ -754,3 +754,49 @@ def test_user_drafts_categories_filter(pub, local_user):
         sign_uri('/api/users/%s/drafts?category_slugs=category-1,category-2' % local_user.id)
     )
     assert len(resp.json['data']) == 5
+
+
+def test_user_forms_filter_on_status(pub, local_user):
+
+    FormDef.wipe()
+    formdef = FormDef()
+    formdef.name = 'test'
+    formdef.fields = [
+        fields.StringField(id='0', label='foobar', varname='foobar'),
+        fields.StringField(id='1', label='foobar2'),
+    ]
+    formdef.store()
+    formdef.data_class().wipe()
+
+    formdata1 = formdef.data_class()()
+    formdata1.data = {'0': 'foo@localhost', '1': 'xxx'}
+    formdata1.user_id = local_user.id
+    formdata1.just_created()
+    formdata1.jump_status('new')
+    formdata1.store()
+
+    formdata2 = formdef.data_class()()
+    formdata2.data = {'0': 'foo@localhost', '1': 'xxx'}
+    formdata2.user_id = local_user.id
+    formdata2.just_created()
+    formdata2.jump_status('finished')
+    formdata2.store()
+
+    def get_ids(url):
+        resp = get_app(pub).get(url)
+        return {int(x['form_number_raw']) for x in resp.json['data']}
+
+    resp = get_ids(sign_uri('/api/user/forms', user=local_user))
+    assert resp == {int(formdata1.id), int(formdata2.id)}
+
+    resp = get_ids(sign_uri('/api/user/forms?status=all', user=local_user))
+    assert resp == {int(formdata1.id), int(formdata2.id)}
+
+    resp = get_ids(sign_uri('/api/user/forms?status=done', user=local_user))
+    assert resp == {int(formdata2.id)}
+
+    resp = get_ids(sign_uri('/api/user/forms?status=open', user=local_user))
+    assert resp == {int(formdata1.id)}
+
+    resp = get_ids(sign_uri('/api/user/forms?status=open', user=local_user))
+    assert resp == {int(formdata1.id)}

wcs/api.py

@@ -908,6 +908,17 @@ class ApiUserDirectory(Directory):
                 criterias = [Equal('user_id', str(user.id))]
             if category_slugs:
                 criterias.append(Contains('category_id', [c.id for c in categories]))
+
+            status_criteria = get_request().form.get('status') or 'all'
+            if status_criteria == 'open':
+                criterias.append(Equal('is_at_endpoint', False))
+            elif status_criteria == 'done':
+                criterias.append(Equal('is_at_endpoint', True))
+            elif status_criteria == 'all':
+                pass
+            else:
+                return HttpResponseBadRequest('invalid status parameter value')
+
             user_forms = sql.AnyFormData.select(
                 criterias,
                 limit=misc.get_int_or_400(get_request().form.get('limit')),
@@ -943,6 +954,17 @@ class ApiUserDirectory(Directory):
                 user_forms.extend(formdef.data_class().get_with_indexed_value('user_id', user.id))
             if category_slugs:
                 user_forms = [f for f in user_forms if f.formdef.category_id in [c.id for c in categories]]
+
+            status_criteria = get_request().form.get('status') or 'all'
+            if status_criteria == 'open':
+                user_forms = [x for x in user_forms if not x.is_at_endpoint_status()]
+            elif status_criteria == 'done':
+                user_forms = [x for x in user_forms if x.is_at_endpoint_status()]
+            elif status_criteria == 'all':
+                pass
+            else:
+                return HttpResponseBadRequest('invalid status parameter value')
+
             typed_none = time.gmtime(-(10 ** 10))
             user_forms.sort(key=lambda x: x.receipt_time or typed_none)
             if get_request().form.get('sort') == 'desc':