workflows: allow set backoffice field action to work with files (#11440)

tests/test_workflows.py

@@ -1675,3 +1675,61 @@ def test_set_backoffice_field(pub):
 
     formdata = formdef.data_class().get(formdata.id)
     assert formdata.data['bo1'] == 'HELLO'
+
+def test_set_backoffice_field_file(pub):
+    Workflow.wipe()
+    wf = Workflow(name='xxx')
+    wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf)
+    wf.backoffice_fields_formdef.fields = [
+        FileField(id='bo1', label='1st backoffice field',
+            type='file', varname='backoffice_file'),
+    ]
+    st1 = wf.add_status('Status1')
+    wf.store()
+
+    formdef = FormDef()
+    formdef.name = 'baz'
+    formdef.fields = [
+        StringField(id='1', label='File', type='file', varname='file'),
+    ]
+    formdef.workflow_id = wf.id
+    formdef.store()
+
+    upload = PicklableUpload('test.jpeg', 'image/jpeg')
+    upload.receive([open(os.path.join(os.path.dirname(__file__), 'image-with-gps-data.jpeg')).read()])
+
+    formdata = formdef.data_class()()
+    formdata.data = {'1': upload}
+    formdata.just_created()
+    formdata.store()
+
+    pub.substitutions.feed(formdata)
+    item = SetBackofficeFieldsWorkflowStatusItem()
+    item.fields = [{'field_id': 'bo1', 'value': '=form_var_file_raw'}]
+    item.perform(formdata)
+
+    formdata = formdef.data_class().get(formdata.id)
+    assert formdata.data['bo1'].base_filename == 'test.jpeg'
+
+    # check storing response as attachment
+    pub.substitutions.feed(formdata)
+    item = WebserviceCallStatusItem()
+    item.url = 'http://remote.example.net/xml'
+    item.post = False
+    item.varname = 'xxx'
+    item.response_type = 'attachment'
+    item.record_errors = True
+    item.perform(formdata)
+    attachment = formdata.evolution[-1].parts[-1]
+    assert isinstance(attachment, AttachmentEvolutionPart)
+    assert attachment.base_filename == 'xxx.xml'
+    assert attachment.content_type == 'text/xml'
+
+    formdata = formdef.data_class().get(formdata.id)
+    pub.substitutions.feed(formdata)
+    item = SetBackofficeFieldsWorkflowStatusItem()
+    item.fields = [{'field_id': 'bo1', 'value': '=attachments.xxx'}]
+    item.perform(formdata)
+
+    formdata = formdef.data_class().get(formdata.id)
+    assert formdata.data['bo1'].base_filename == 'xxx.xml'

wcs/wf/backoffice_fields.py

@@ -20,9 +20,11 @@ import xml.etree.ElementTree as ET
 from quixote import get_publisher
 
 from qommon import get_logger
-from qommon.form import WidgetListAsTable, CompositeWidget, SingleSelectWidget, ComputedExpressionWidget
+from qommon.form import (WidgetListAsTable, CompositeWidget, SingleSelectWidget,
+        ComputedExpressionWidget, PicklableUpload)
 from wcs.fields import WidgetField
-from wcs.workflows import XmlSerialisable, WorkflowStatusItem, register_item_class
+from wcs.workflows import (XmlSerialisable, WorkflowStatusItem,
+        register_item_class, NamedAttachmentsSubstitutionProxy)
 from wcs.wf.profile import FieldNode
 
 
@@ -86,10 +88,14 @@ class SetBackofficeFieldsWorkflowStatusItem(WorkflowStatusItem):
             return
         for field in self.fields:
             try:
-                formdata.data['%s' % field['field_id']] = self.compute(
-                        field['value'], raises=True)
+                new_value = self.compute(field['value'], raises=True)
             except:
                 get_publisher().notify_of_exception(sys.exc_info())
+            if isinstance(new_value, NamedAttachmentsSubstitutionProxy):
+                upload = PicklableUpload(new_value.filename, new_value.content_type)
+                upload.receive([new_value.content])
+                new_value = upload
+            formdata.data['%s' % field['field_id']] = new_value
         formdata.store()
 
     def fields_export_to_xml(self, item, charset, include_id=False):