wf/export_to_model: support XML templates (#56537)

tests/test_workflows.py

@@ -45,7 +45,7 @@ from wcs.fields import (
 from wcs.formdata import Evolution
 from wcs.formdef import FormDef
 from wcs.qommon.errors import ConnectionError
-from wcs.qommon.form import Form, UploadedFile
+from wcs.qommon.form import Form, UploadedFile, UploadValidationError
 from wcs.qommon.http_request import HTTPRequest
 from wcs.qommon.upload_storage import PicklableUpload
 from wcs.wf.aggregation_email import (
@@ -3987,6 +3987,63 @@ def test_export_to_model_django_template(pub):
     assert b'>A &lt;&gt; name<' in new_content
 
 
+def test_export_to_model_xml(two_pubs):
+    pub = two_pubs
+    LoggedError = pub.loggederror_class
+    if LoggedError:
+        LoggedError.wipe()
+
+    formdef = FormDef()
+    formdef.name = 'foo-export-to-template-with-django'
+    formdef.fields = [
+        StringField(id='1', label='String', type='string', varname='string'),
+    ]
+    formdef.store()
+    formdata = formdef.data_class()()
+    formdata.data = {'1': 'écho'}
+    formdata.just_created()
+    formdata.store()
+
+    # good XML
+    item = ExportToModel()
+    item.method = 'non-interactive'
+    item.attach_to_history = True
+
+    def run(template, filename='/foo/template.xml', content_type='application/xml'):
+        upload = QuixoteUpload(filename, content_type=content_type)
+        upload.fp = io.BytesIO()
+        upload.fp.write(template.encode())
+        upload.fp.seek(0)
+        item.model_file = UploadedFile(pub.app_dir, None, upload)
+        item.convert_to_pdf = False
+        pub.substitutions.reset()
+        pub.substitutions.feed(formdata)
+        item.perform(formdata)
+        with open(formdata.evolution[0].parts[-1].filename) as fd:
+            return fd.read()
+
+    # good XML
+    assert run(template='<a>{{ form_var_string }}</a>') == '<a>écho</a>'
+    assert (
+        run(template='<a>{{ form_var_string }}</a>', content_type='application/octet-stream') == '<a>écho</a>'
+    )
+    assert run(template='<a>{{ form_var_string }}</a>', filename='/foo/template.svg') == '<a>écho</a>'
+
+    # unknown file format
+    with pytest.raises(UploadValidationError):
+        run(
+            template='<a>{{ form_var_string }}</a>',
+            filename='/foo/template.txt',
+            content_type='application/octet-stream',
+        )
+
+    # malformed XML
+    assert not LoggedError or LoggedError.count() == 0
+    assert run(template='<a>{{ form_var_string }}<a>') == '<a>écho<a>'
+    # on error in the XML correctness no exception is raised but an error is logged
+    assert not LoggedError or LoggedError.count() == 1
+
+
 @pytest.mark.parametrize('filename', ['template-form-details.odt', 'template-form-details-no-styles.odt'])
 def test_export_to_model_form_details_section(pub, filename):
     BlockDef.wipe()

wcs/wf/export_to_model.py

@@ -282,6 +282,8 @@ class ExportToModel(WorkflowStatusItem):
             fp = upload.get_file()
         else:
             raise UploadValidationError('unknown upload object %r' % upload)
+
+        # RTF
         if upload.content_type and upload.content_type == 'application/rtf':
             return 'rtf'
         if (
@@ -292,6 +294,8 @@ class ExportToModel(WorkflowStatusItem):
         if fp.read(10).startswith(b'{\\rtf'):
             fp.seek(0)
             return 'rtf'
+
+        # OpenDocument
         fp.seek(0)
         if upload.content_type and upload.content_type.startswith('application/vnd.oasis.opendocument.'):
             return 'opendocument'
@@ -302,7 +306,16 @@ class ExportToModel(WorkflowStatusItem):
                 return 'opendocument'
         if is_opendocument(fp):
             return 'opendocument'
-        raise UploadValidationError(_('Only RTF and OpenDocument files can be used'))
+
+        # XML
+        fp.seek(0)
+        xml_prefix = fp.read(1)
+        fp.seek(0)
+        if (upload.content_type and upload.content_type in ('text/xml', 'application/xml')) or (
+            upload.base_filename and upload.base_filename.endswith('.xml') and xml_prefix == b'<'
+        ):
+            return 'xml'
+        raise UploadValidationError(_('Only OpenDocument and XML files can be used'))
 
     def get_parameters(self):
         parameters = ('model_file',)
@@ -486,14 +499,39 @@ class ExportToModel(WorkflowStatusItem):
             outstream = self.apply_rtf_template_to_formdata(formdata)
         elif kind == 'opendocument':
             outstream = self.apply_od_template_to_formdata(formdata)
+        elif kind == 'xml':
+            outstream = self.apply_text_template_to_formdata(formdata)
         else:
             raise Exception('unsupported model kind %r' % kind)
-        if self.convert_to_pdf:
-            if transform_to_pdf is None:
-                raise Exception('libreoffice is missing')
-            return transform_to_pdf(outstream)
+        if kind == 'xml':
+            outstream.seek(0)
+            try:
+                ET.parse(outstream)
+            except ET.ParseError as e:
+                get_publisher().record_error(
+                    _('The rendered template is not a valid XML document.'), formdata=formdata, exception=e
+                )
+                # we do not reraise to let people see the result of the
+                # templating to debug the XML correctness
+            finally:
+                outstream.seek(0)
+        else:
+            if self.convert_to_pdf:
+                if transform_to_pdf is None:
+                    raise Exception('libreoffice is missing')
+                return transform_to_pdf(outstream)
         return outstream
 
+    def apply_text_template_to_formdata(self, formdata):
+        return io.BytesIO(
+            force_bytes(
+                template_on_formdata(
+                    formdata,
+                    self.model_file.get_file().read().decode(errors='surrogateescape'),
+                )
+            )
+        )
+
     def apply_rtf_template_to_formdata(self, formdata):
         try:
             # force ezt_only=True because an RTF file may contain {{ characters