api: include workflow form data in evolution parts json view (#89017)

tests/api/test_formdata.py

@@ -29,6 +29,7 @@ from wcs.qommon import ods
 from wcs.qommon.http_request import HTTPRequest
 from wcs.qommon.upload_storage import PicklableUpload
 from wcs.wf.comment import WorkflowCommentPart
+from wcs.wf.form import WorkflowFormEvolutionPart, WorkflowFormFieldsFormDef
 from wcs.workflows import AttachmentEvolutionPart, Workflow, WorkflowBackofficeFieldsFormDef
 
 from ..utilities import clean_temporary_pub, create_temporary_pub, get_app, login
@@ -472,6 +473,106 @@ def test_formdata_backoffice_fields(pub, local_user):
     assert resp.json['workflow']['fields']['backoffice_blah'] == 'Hello world'
 
 
+@pytest.mark.parametrize('user', ['query-email', 'api-access'])
+@pytest.mark.parametrize('auth', ['signature', 'http-basic'])
+def test_formdata_workflow_form(pub, local_user, user, auth):
+    app = get_app(pub)
+
+    if user == 'api-access':
+        ApiAccess.wipe()
+        access = ApiAccess()
+        access.name = 'test'
+        access.access_identifier = 'test'
+        access.access_key = '12345'
+        access.store()
+
+        if auth == 'http-basic':
+
+            def get_url(url, **kwargs):
+                app.set_authorization(('Basic', ('test', '12345')))
+                return app.get(url, **kwargs)
+
+        else:
+
+            def get_url(url, **kwargs):
+                return app.get(sign_uri(url, orig=access.access_identifier, key=access.access_key), **kwargs)
+
+    else:
+        if auth == 'http-basic':
+            pytest.skip('http basic authentication requires ApiAccess')
+
+        def get_url(url, **kwargs):
+            return app.get(sign_uri(url, user=local_user), **kwargs)
+
+    role = pub.role_class(name='test')
+    role.id = '123'
+    role.store()
+
+    Workflow.wipe()
+    workflow = Workflow(name='foo')
+    st = workflow.add_status('st1')
+    form_action = st.add_action('form')
+    form_action.varname = 'blah'
+    form_action.formdef = WorkflowFormFieldsFormDef(item=form_action)
+    form_action.formdef.fields = [
+        fields.FileField(id='1', label='file', varname='file'),
+        fields.StringField(id='2', label='str', varname='str'),
+    ]
+    workflow.store()
+
+    FormDef.wipe()
+    formdef = FormDef()
+    formdef.name = 'test'
+    formdef.fields = []
+    formdef.workflow_id = workflow.id
+    formdef.workflow_roles = {'_receiver': role.id}
+    formdef.store()
+
+    formdef.data_class().wipe()
+    formdata = formdef.data_class()()
+    formdata.just_created()
+    data = {'1': PicklableUpload('test.txt', 'text/plain'), '2': 'text'}
+    data['1'].receive([b'hello world wf form'])
+    formdata.evolution[-1].parts = [
+        WorkflowFormEvolutionPart(form_action, data),
+    ]
+    formdata.store()
+
+    if user == 'api-access':
+        access.roles = [role]
+        access.store()
+    else:
+        local_user.roles = [role.id]
+        local_user.store()
+
+    resp = get_url('/api/forms/test/%s/' % formdata.id, status=200)
+    assert resp.json['evolution'][0]['parts'] == [
+        {
+            'data': {
+                'file': 'test.txt',
+                'file_raw': {
+                    'content': 'aGVsbG8gd29ybGQgd2YgZm9ybQ==',
+                    'content_type': 'text/plain',
+                    'filename': 'test.txt',
+                },
+                'file_url': None,
+                'str': 'text',
+            },
+            'key': 'blah',
+            'type': 'workflow-form',
+        }
+    ]
+
+    resp = get_url('/api/forms/test/%s/?include-files-content=off' % formdata.id, status=200)
+    assert resp.json['evolution'][0]['parts'] == [
+        {
+            'data': {'file': 'test.txt', 'file_url': None, 'str': 'text'},
+            'key': 'blah',
+            'type': 'workflow-form',
+        }
+    ]
+
+
 def test_formdata_duplicated_varnames(pub, local_user):
     pub.role_class.wipe()
     role = pub.role_class(name='test')

tests/test_formdef.py

@@ -479,8 +479,8 @@ def test_unused_file_removal_job(pub):
         display_form.formdef = WorkflowFormFieldsFormDef(item=display_form)
         display_form.formdef.fields = []
 
-        data = {'blah_1': PicklableUpload('test.txt', 'text/plain')}
-        data['blah_1'].receive([b'hello world wf form'])
+        data = {'1': PicklableUpload('test.txt', 'text/plain')}
+        data['1'].receive([b'hello world wf form'])
         formdata.evolution[-1].parts = [
             WorkflowFormEvolutionPart(display_form, data),
         ]

wcs/formdata.py

@@ -45,7 +45,7 @@ class NoContentSnapshotAt(RequestError):
     pass
 
 
-def get_dict_with_varnames(fields, data, formdata=None, varnames_only=False):
+def get_dict_with_varnames(fields, data, formdata=None, varnames_only=False, include_files=True):
     new_data = {}
     for field in fields:
         if not hasattr(field, 'get_view_value'):
@@ -80,6 +80,8 @@ def get_dict_with_varnames(fields, data, formdata=None, varnames_only=False):
                     new_data['var_%s_raw' % field.varname] = value
                     new_data['var_%s_url' % field.varname] = None
                     if value and hasattr(value, 'base_filename'):
+                        if include_files is False:
+                            del new_data[f'var_{field.varname}_raw']
                         new_data['var_%s' % field.varname] = value.base_filename
                         if formdata is not None:
                             new_data['var_%s_url' % field.varname] = '%s?f=%s' % (

wcs/wf/form.py

@@ -48,6 +48,21 @@ class WorkflowFormEvolutionPart(EvolutionPart):
         assert not getattr(self, 'live')
         return self.__dict__
 
+    def get_json_export_dict(self, anonymise=False, include_files=True):
+        if not self.varname or anonymise:
+            return None
+        d = {
+            'type': 'workflow-form',
+            'key': self.varname,
+            'data': {
+                k.removeprefix('var_'): v
+                for k, v in get_dict_with_varnames(
+                    self.formdef.fields, self.data, varnames_only=True, include_files=include_files
+                ).items()
+            },
+        }
+        return d
+
 
 def lookup_wf_form_file(self, filename):
     # supports for URLs such as /$formdata/$id/files/form-$formvar-$fieldvar/test.txt