workflows: consider parent object when looking for related objects (#43022)

tests/test_workflows.py

@@ -4934,3 +4934,90 @@ def test_call_external_workflow_with_data_sourced_object(pub):
     logged_error = LoggedError.select()[0]
     assert logged_error.summary == 'Could not find linked "Data" object by id %s' % carddata.id
     assert logged_error.exception_class == 'KeyError'
+
+
+def test_call_external_workflow_with_parent_object(pub):
+    FormDef.wipe()
+    CardDef.wipe()
+    LoggedError.wipe()
+
+    # carddef workflow, with global action to increment a counter in its
+    # backoffice fields.
+    carddef_wf = Workflow(name='Carddef Workflow')
+    carddef_wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(carddef_wf)
+    carddef_wf.backoffice_fields_formdef.fields = [
+        StringField(id='bo0', varname='bo', type='string', label='bo variable'),
+    ]
+    increment_global_action = carddef_wf.add_global_action('Update')
+    increment_global_action.append_item('set-backoffice-fields')
+    setbo = increment_global_action.items[0]
+    setbo.fields = [{'field_id': 'bo0',
+                     'value': '{{ form_var_bo|default:"0"|add:1 }}'}]
+    trigger = increment_global_action.append_trigger('webservice')
+    trigger.identifier = 'update'
+    carddef_wf.store()
+
+    # associated carddef
+    carddef = CardDef()
+    carddef.name = 'Data'
+    carddef.fields = [
+        StringField(id='0', label='string', varname='card_string'),
+    ]
+    carddef.workflow = carddef_wf
+    carddef.store()
+
+    # and sample carddata
+    carddata = carddef.data_class()()
+    carddata.data = {'0': 'Text'}
+    carddata.store()
+
+    # formdef workflow that will trigger the global action
+    wf = Workflow(name='External actions')
+    wf.add_status('Action')
+
+    update_global_action = wf.add_global_action('Update linked object data')
+    update_action = update_global_action.append_item('external_workflow_global_action')
+    update_action.slug = 'carddef:%s' % carddef.url_name
+    update_action.trigger_id = 'action:update'
+    wf.store()
+
+    # associated formdef
+    formdef = FormDef()
+    formdef.name = 'External action form'
+    formdef.fields = [EmailField(id='1', label='Email', varname='email')]
+    formdef.workflow = wf
+    formdef.store()
+
+    # and formdata
+    formdata = formdef.data_class()()
+    formdata.data = {'1': 'foo@example.com'}
+    formdata.store()
+    formdata.just_created()
+    formdata.perform_workflow()
+
+    # run, against no parent
+    perform_items([update_action], formdata)
+    card = carddef.data_class().get(carddata.id)
+    assert 'bo0' not in card.data  # not called
+
+    # other parent
+    formdata.submission_context = {
+        'orig_object_type': 'formdef',
+        'orig_formdata_id': str(formdata.id),
+        'orig_formdef_id': str(formdef.id),
+    }
+    formdata.store()
+    perform_items([update_action], formdata)
+    card = carddef.data_class().get(carddata.id)
+    assert 'bo0' not in card.data  # not called
+
+    # appropriate parent
+    formdata.submission_context = {
+        'orig_object_type': 'carddef',
+        'orig_formdata_id': str(carddata.id),
+        'orig_formdef_id': str(carddef.id),
+    }
+    formdata.store()
+    perform_items([update_action], formdata)
+    card = carddef.data_class().get(carddata.id)
+    assert card.data['bo0'] == '1'  # got called

wcs/wf/external_workflow.py

@@ -112,6 +112,13 @@ class ExternalWorkflowGlobalAction(WorkflowStatusItem):
         return _('not completed')
 
     def iter_target_datas(self, formdata, objectdef):
+        parent = formdata.get_parent()
+        if parent:
+            parent_identifier = '%s:%s' % (parent.formdef.xml_root_node,
+                    parent.formdef.url_name)
+            if parent_identifier == self.slug:
+                yield parent
+
         data_ids = []
         # search linked objects in data sources
         for field in formdata.get_formdef().get_all_fields():