0001-workflow-check-datasource-existence-on-import-48164.patch

     import pytest
     import xml.etree.ElementTree as ET
     from quixote.http_request import Upload
     from django.utils.six import BytesIO
     from wcs.carddef import CardDef
     from wcs.formdef import FormDef
     from wcs.mail_templates import MailTemplate
     from wcs.workflows import (
-...
     from wcs.wf.redirect_to_url import RedirectToUrlWorkflowStatusItem
     from wcs.wf.create_formdata import CreateFormdataWorkflowStatusItem, Mapping
     from wcs.wf.external_workflow import ExternalWorkflowGlobalAction
     from wcs.roles import Role
     from wcs.wf.form import FormWorkflowStatusItem, WorkflowFormFieldsFormDef
     from wcs.wf.jump import JumpWorkflowStatusItem
     from wcs.fields import StringField, FileField
     from wcs.qommon.form import UploadedFile
     from wcs.roles import Role
     from wcs.workflows import ExportToModel, WorkflowVariablesFieldsFormDef, DisplayMessageWorkflowStatusItem
     from wcs.qommon.misc import indent_xml as indent
-...
         role.name = 'Test Role'
         role.store()
         from wcs.wf.dispatch import DispatchWorkflowStatusItem
         dispatch = DispatchWorkflowStatusItem()
         dispatch.id = '_x'
         dispatch.role_id = 5
-...
         wf = Workflow(name='status')
         st1 = wf.add_status('Status1', 'st1')
         from wcs.wf.form import FormWorkflowStatusItem, WorkflowFormFieldsFormDef
         display_form = FormWorkflowStatusItem()
         display_form.id = '_x'
         display_form.formdef = WorkflowFormFieldsFormDef(item=display_form)
-...
         wf.store()
         st1 = wf.add_status('Status1', 'st1')
         from quixote.http_request import Upload
         from wcs.qommon.form import UploadedFile
         from wcs.workflows import ExportToModel
         export_to = ExportToModel()
         export_to.label = 'test'
         upload = Upload('/foo/bar', content_type='application/vnd.oasis.opendocument.text')
-...
         st1 = wf.add_status('Status1', 'st1')
         st2 = wf.add_status('Status2', 'st2')
         from wcs.wf.jump import JumpWorkflowStatusItem
         jump = JumpWorkflowStatusItem()
         jump.id = '_jump'
         jump.by = ['_submitter', '_receiver']
-...
     def test_variables_formdef(pub):
         wf = Workflow(name='variables')
         from wcs.workflows import WorkflowVariablesFieldsFormDef
         wf.variables_formdef = WorkflowVariablesFieldsFormDef(workflow=wf)
         wf.variables_formdef.fields.append(StringField(label='Test', type='string'))
         wf2 = assert_import_export_works(wf)
-...
         wf = Workflow(name='status')
         st1 = wf.add_status('Status1', 'st1')
         from wcs.workflows import DisplayMessageWorkflowStatusItem
         display = DisplayMessageWorkflowStatusItem()
         display.message = 'hey'
         display.to = ['_submitter', '1']
-...
         export = ET.tostring(wf.export_to_xml(include_id=True))
         with pytest.raises(WorkflowImportError, match='Unknown referenced mail template'):
             Workflow.import_from_xml_tree(ET.fromstring(export), include_id=True)
     def test_unknown_data_source(pub):
         wf1 = Workflow(name='status')
         st1 = wf1.add_status('Status1', 'st1')
         display_form = FormWorkflowStatusItem()
         display_form.id = '_x'
         display_form.formdef = WorkflowFormFieldsFormDef(item=display_form)
         display_form.formdef.fields = [StringField(label='Test', type='string', data_source={'type': 'foobar'})]
         st1.items.append(display_form)
         display_form.parent = st1
         wf2 = Workflow(name='variables')
         wf2.variables_formdef = WorkflowVariablesFieldsFormDef(workflow=wf2)
         wf2.variables_formdef.fields = [StringField(label='Test', type='string', data_source={'type': 'foobar'})]
         wf3 = Workflow(name='bo fields')
         wf3.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf3)
         wf3.backoffice_fields_formdef.fields = [
             StringField(
                 id='bo1', label='1st backoffice field',
                 type='string', varname='backoffice_blah',
                 data_source={'type': 'foobar'})
+        ]
         for wf in [wf1, wf2, wf3]:
             export = ET.tostring(export_to_indented_xml(wf))
             with pytest.raises(WorkflowImportError, match='Unknown datasources'):
                 Workflow.import_from_xml(BytesIO(export))
         # carddef as datasource
         CardDef.wipe()
         carddef = CardDef()
         carddef.name = 'foo'
         carddef.fields = [StringField(id='1', label='Test', type='string', varname='foo')]
         carddef.store()
         display_form.formdef.fields[0].data_source = {'type': 'carddef:foo'}
         wf2.variables_formdef.fields[0].data_source = {'type': 'carddef:foo'}
         wf3.backoffice_fields_formdef.fields[0].data_source = {'type': 'carddef:foo'}
         for wf in [wf1, wf2, wf3]:
             export = ET.tostring(export_to_indented_xml(wf))
             Workflow.import_from_xml(BytesIO(export))
         display_form.formdef.fields[0].data_source = {'type': 'carddef:unknown'}
         wf2.variables_formdef.fields[0].data_source = {'type': 'carddef:unknown'}
         wf3.backoffice_fields_formdef.fields[0].data_source = {'type': 'carddef:unknown'}
         for wf in [wf1, wf2, wf3]:
             export = ET.tostring(export_to_indented_xml(wf))
             with pytest.raises(WorkflowImportError, match='Unknown datasources'):
                 Workflow.import_from_xml(BytesIO(export))
         # carddef custom view as datasource
         pub.custom_view_class.wipe()
         custom_view = pub.custom_view_class()
         custom_view.title = 'card view'
         custom_view.formdef = carddef
         custom_view.columns = {'list': [{'id': 'id'}]}
         custom_view.filters = {}
         custom_view.visibility = 'datasource'
         custom_view.store()
         display_form.formdef.fields[0].data_source = {'type': 'carddef:foo:card-view'}
         wf2.variables_formdef.fields[0].data_source = {'type': 'carddef:foo:card-view'}
         wf3.backoffice_fields_formdef.fields[0].data_source = {'type': 'carddef:foo:card-view'}
         for wf in [wf1, wf2, wf3]:
             export = ET.tostring(export_to_indented_xml(wf))
             Workflow.import_from_xml(BytesIO(export))
         display_form.formdef.fields[0].data_source = {'type': 'carddef:foo:unknown'}
         wf2.variables_formdef.fields[0].data_source = {'type': 'carddef:foo:unknown'}
         wf3.backoffice_fields_formdef.fields[0].data_source = {'type': 'carddef:foo:unknown'}
         for wf in [wf1, wf2, wf3]:
             export = ET.tostring(export_to_indented_xml(wf))
             with pytest.raises(WorkflowImportError, match='Unknown datasources'):
                 Workflow.import_from_xml(BytesIO(export))

                 new_formdef = self.formdef_class.import_from_xml(fp, include_id=True)
             except FormdefImportError as e:
                 error = True
                 reason = _(e.msg)
                 reason = _(e.msg) % e.msg_args
                 if e.details:
                     reason += ' [%s]' % e.details
             except ValueError:
-...
                     get_session().message = ('info', _(self.import_error_message))
             except FormdefImportError as e:
                 error = True
                 reason = _(e.msg)
                 reason = _(e.msg) % e.msg_args
                 if e.details:
                     reason += ' [%s]' % e.details
             except ValueError:

                     reason = _('Not a valid export file')
                 except WorkflowImportError as e:
                     results = None
                     reason = _('Failed to import a workflow (%s); site import did not complete.') % (_(e) % e.msg_args)
                     msg = _(e.msg) % e.msg_args
                     if e.details:
                         msg += ' [%s]' % e.details
                     reason = _('Failed to import a workflow (%s); site import did not complete.') % (msg)
                 html_top('settings', title = _('Import'))
                 r = TemplateIO(html=True)
                 r += htmltext('<h2>%s</h2>') % _('Import')

                 workflow = Workflow.import_from_xml(fp)
             except WorkflowImportError as e:
                 error = True
                 reason = _(e) % e.msg_args
                 reason = _(e.msg) % e.msg_args
                 if e.details:
                     reason += ' [%s]' % e.details
             except ValueError:
                 error = True

     class FormdefImportError(Exception):
         def __init__(self, msg, details=None):
         def __init__(self, msg, msg_args=None, details=None):
             self.msg = msg
             self.msg_args = msg_args or ()
             self.details = details
-...
         @classmethod
         def import_from_xml(cls, fd, charset=None, include_id=False,
                 fix_on_error=False, check_datasources=True):
             from wcs.carddef import CardDef
             try:
                 tree = ET.parse(fd)
             except:
                 raise ValueError()
             formdef = cls.import_from_xml_tree(tree, charset=charset,
                     include_id=include_id, fix_on_error=fix_on_error)
             formdef = cls.import_from_xml_tree(
                 tree, charset=charset,
                 include_id=include_id, fix_on_error=fix_on_error,
                 check_datasources=check_datasources)
             if formdef.url_name:
                 try:
-...
                 if formdef.max_field_id < max_field_id:
                     formdef.max_field_id = max_field_id
             if check_datasources:
                 # check if datasources are defined
                 unknown_datasources = set()
                 for field in formdef.fields:
                     data_source = getattr(field, 'data_source', None)
                     if data_source:
                         data_source_id = data_source.get('type')
                         if isinstance(data_sources.get_object(data_source),
                                       data_sources.StubNamedDataSource):
                             unknown_datasources.add(data_source_id)
                         elif data_source_id and data_source_id.startswith('carddef:'):
                             parts = data_source_id.split(':')
                             # check if carddef exists
                             url_name = parts[1]
                             if formdef.xml_root_node == 'carddef' and formdef.url_name == url_name:
                                 # reference to itself, it's ok
                                 continue
                             try:
                                 CardDef.get_by_urlname(url_name)
                             except KeyError:
                                 unknown_datasources.add(data_source_id)
                                 continue
                             if len(parts) == 2:
                                 continue
                             lookup_criterias = [
                                 Equal('formdef_type', 'carddef'),
                                 Equal('visibility', 'datasource'),
                                 Equal('slug', parts[2]),
+                            ]
                             try:
                                 get_publisher().custom_view_class.select(lookup_criterias)[0]
                             except IndexError:
                                 unknown_datasources.add(data_source_id)
                 if unknown_datasources:
                     raise FormdefImportError(N_('Unknown datasources'),
                             details=', '.join(sorted(unknown_datasources)))
             # check if all field id are unique
             known_field_ids = set()
             for field in formdef.fields:
-...
         @classmethod
         def import_from_xml_tree(cls, tree, include_id=False, charset=None,
                 fix_on_error=False, snapshot=False):
                 fix_on_error=False, snapshot=False, check_datasources=True):
             from wcs.carddef import CardDef
             if charset is None:
                 charset = get_publisher().site_charset
             assert charset == 'utf-8'
-...
                 for child in node:
                     formdef.required_authentication_contexts.append(str(child.text))
             if check_datasources:
                 # check if datasources are defined
                 unknown_datasources = set()
                 for field in formdef.fields:
                     data_source = getattr(field, 'data_source', None)
                     if data_source:
                         data_source_id = data_source.get('type')
                         if isinstance(data_sources.get_object(data_source),
                                       data_sources.StubNamedDataSource):
                             unknown_datasources.add(data_source_id)
                         elif data_source_id and data_source_id.startswith('carddef:'):
                             parts = data_source_id.split(':')
                             # check if carddef exists
                             url_name = parts[1]
                             if formdef.xml_root_node == 'carddef' and formdef.url_name == url_name:
                                 # reference to itself, it's ok
                                 continue
                             try:
                                 CardDef.get_by_urlname(url_name)
                             except KeyError:
                                 unknown_datasources.add(data_source_id)
                                 continue
                             if len(parts) == 2:
                                 continue
                             lookup_criterias = [
                                 Equal('formdef_type', 'carddef'),
                                 Equal('visibility', 'datasource'),
                                 Equal('slug', parts[2]),
+                            ]
                             try:
                                 get_publisher().custom_view_class.select(lookup_criterias)[0]
                             except IndexError:
                                 unknown_datasources.add(data_source_id)
                 if unknown_datasources:
                     raise FormdefImportError(
                         N_('Unknown datasources'),
                         details=', '.join(sorted(unknown_datasources)))
             return formdef
         def get_detailed_email_form(self, formdata, url):

             from wcs.workflows import Workflow
             for f in z.namelist():
                 if os.path.dirname(f) == 'workflows_xml' and os.path.basename(f):
                     workflow = Workflow.import_from_xml(z.open(f), include_id=True)
                     workflow = Workflow.import_from_xml(
                         z.open(f), include_id=True, check_datasources=False)
                     workflow.store()
                     results['workflows'] += 1

                 fields.append(field.export_to_xml(charset=charset, include_id=include_id))
             return item
         def init_with_xml(self, elem, charset, include_id=False, snapshot=False):
         def init_with_xml(self, elem, charset, include_id=False, snapshot=False, check_datasources=True):
             WorkflowStatusItem.init_with_xml(self, elem, charset)
             el = elem.find('formdef')
             if el is None:
                 return
             # we can always include id in the formdef export as it lives in
             # a different space, isolated from other formdefs.
             imported_formdef = FormDef.import_from_xml_tree(el, include_id=True, snapshot=snapshot)
             imported_formdef = FormDef.import_from_xml_tree(el, include_id=True, snapshot=snapshot, check_datasources=check_datasources)
             self.formdef = WorkflowFormFieldsFormDef(item=self)
             self.formdef.fields = imported_formdef.fields
             if self.formdef.max_field_id is None and self.formdef.fields:

     from .conditions import Condition
     from .roles import Role, logged_users_role, get_user_roles
     from .fields import FileField
     from .formdef import FormDef
     from .formdef import FormDef, FormdefImportError
     from .carddef import CardDef
     from .formdata import Evolution
     from .mail_templates import MailTemplate
-...
     class WorkflowImportError(Exception):
         def __init__(self, msg, msg_args=None):
             super(WorkflowImportError, self).__init__(msg)
         def __init__(self, msg, msg_args=None, details=None):
             self.msg = msg
             self.msg_args = msg_args or ()
             self.details = details
     class AbortActionException(Exception):
-...
             return root
         @classmethod
         def import_from_xml(cls, fd, include_id=False):
         def import_from_xml(cls, fd, include_id=False, check_datasources=True):
             try:
                 tree = ET.parse(fd)
             except:
                 raise ValueError()
             return cls.import_from_xml_tree(tree, include_id=include_id)
             return cls.import_from_xml_tree(tree, include_id=include_id, check_datasources=check_datasources)
         @classmethod
         def import_from_xml_tree(cls, tree, include_id=False, snapshot=False):
         def import_from_xml_tree(cls, tree, include_id=False, snapshot=False, check_datasources=True):
             charset = get_publisher().site_charset
             workflow = cls()
             if tree.find('name') is None or not tree.find('name').text:
-...
             for status in tree.find('possible_status'):
                 status_o = WorkflowStatus()
                 status_o.parent = workflow
                 status_o.init_with_xml(status, charset, include_id=include_id, snapshot=snapshot)
                 try:
                     status_o.init_with_xml(
                         status, charset, include_id=include_id, snapshot=snapshot, check_datasources=check_datasources)
                 except FormdefImportError as e:
                     raise WorkflowImportError(e.msg, details=e.details)
                 workflow.possible_status.append(status_o)
             workflow.global_actions = []
-...
             variables = tree.find('variables')
             if variables is not None:
                 formdef = variables.find('formdef')
                 imported_formdef = FormDef.import_from_xml_tree(formdef, include_id=True, snapshot=snapshot)
                 try:
                     imported_formdef = FormDef.import_from_xml_tree(
                         formdef, include_id=True, snapshot=snapshot, check_datasources=check_datasources)
                 except FormdefImportError as e:
                     raise WorkflowImportError(e.msg, details=e.details)
                 workflow.variables_formdef = WorkflowVariablesFieldsFormDef(workflow=workflow)
                 workflow.variables_formdef.fields = imported_formdef.fields
             variables = tree.find('backoffice-fields')
             if variables is not None:
                 formdef = variables.find('formdef')
                 imported_formdef = FormDef.import_from_xml_tree(formdef, include_id=True, snapshot=snapshot)
                 try:
                     imported_formdef = FormDef.import_from_xml_tree(
                         formdef, include_id=True, snapshot=snapshot, check_datasources=check_datasources)
                 except FormdefImportError as e:
                     raise WorkflowImportError(e.msg, details=e.details)
                 workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow=workflow)
                 workflow.backoffice_fields_formdef.fields = imported_formdef.fields
-...
                         el.text = str(val)
             return node
         def init_with_xml(self, elem, charset, include_id=False, snapshot=False):
         def init_with_xml(self, elem, charset, include_id=False, snapshot=False, check_datasources=True):
             if include_id and elem.attrib.get('id'):
                 self.id = elem.attrib.get('id')
             for attribute in self.get_parameters():
-...
                     include_id=include_id))
             return status
         def init_with_xml(self, elem, charset, include_id=False, snapshot=False):
         def init_with_xml(self, elem, charset, include_id=False, snapshot=False, check_datasources=True):
             self.id = xml_node_text(elem.find('id'))
             self.name = xml_node_text(elem.find('name'))
             if elem.find('colour') is not None:
-...
                 self.append_item(item_type)
                 item_o = self.items[-1]
                 item_o.parent = self
                 item_o.init_with_xml(item, charset, include_id=include_id, snapshot=snapshot)
                 item_o.init_with_xml(item, charset, include_id=include_id, snapshot=snapshot, check_datasources=check_datasources)
         def __repr__(self):
             return '<%s %s %r>' % (self.__class__.__name__, self.id, self.name)
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.