Add check to ensure with items task has action property specified

Normal task is allowed to leave action property empty to perform action less task transition. With items task should not leave the action property empty since it will just consume unnecessary system resources to iterate thru the items list.

Author: m4dcoder

orquesta/specs/native/v1/models.py

@@ -378,6 +378,20 @@ def has_cycles(self):
 
         return False
 
+    def detect_actionless_with_items(self, parent=None):
+        result = []
+
+        # Identify with items task with no action defined.
+        for task_name, task_spec in six.iteritems(self):
+            if task_spec.has_items() and not task_spec.action:
+                message = 'The action property is required for with items task.'
+                spec_path = parent.get('spec_path') + '.' + task_name
+                schema_path = parent.get('schema_path') + '.patternProperties.^\\w+$'
+                entry = {'message': message, 'spec_path': spec_path, 'schema_path': schema_path}
+                result.append(entry)
+
+        return result
+
     def detect_reserved_names(self, parent=None):
         result = []
 
@@ -515,6 +529,7 @@ def inspect_semantics(self, parent=None):
         result = self.detect_reserved_names(parent=parent)
         result.extend(self.detect_undefined_tasks(parent=parent))
         result.extend(self.detect_unreachable_tasks(parent=parent))
+        result.extend(self.detect_actionless_with_items(parent=parent))
 
         return result
 

orquesta/tests/unit/conducting/test_workflow_conductor_with_items.py

@@ -103,6 +103,13 @@ def test_bad_with_items_syntax(self):
         """
 
         expected_errors = {
+            'semantics': [
+                {
+                    'message': 'The action property is required for with items task.',
+                    'schema_path': 'properties.tasks.patternProperties.^\\w+$',
+                    'spec_path': 'tasks.task1'
+                }
+            ],
             'syntax': [
                 {
                     'message': 'Additional properties are not allowed (\'action\' was unexpected)',
@@ -141,45 +148,18 @@ def test_with_items_that_is_action_less(self):
           - items: <% ctx(items) %>
         """
 
-        spec = native_specs.WorkflowSpec(wf_def)
-        self.assertDictEqual(spec.inspect(), {})
-
-        conductor = conducting.WorkflowConductor(spec)
-        conductor.request_workflow_status(statuses.RUNNING)
-
-        # Mock the action execution for each item and assert expected task statuses.
-        task_route = 0
-        task_name = 'task1'
-        task_ctx = {'xs': ['fee', 'fi', 'fo', 'fum']}
-        task_action_specs = [{'action': None, 'input': None, 'item_id': i} for i in range(0, 4)]
-
-        mock_ac_ex_statuses = [statuses.SUCCEEDED] * 4
-        expected_task_statuses = [statuses.RUNNING] * 3 + [statuses.SUCCEEDED]
-        expected_workflow_statuses = [statuses.RUNNING] * 3 + [statuses.SUCCEEDED]
-
-        self.assert_task_items(
-            conductor,
-            task_name,
-            task_route,
-            task_ctx,
-            task_ctx['xs'],
-            task_action_specs,
-            mock_ac_ex_statuses,
-            expected_task_statuses,
-            expected_workflow_statuses,
-            mock_ac_ex_results=[None] * 4
-        )
-
-        # Assert the task is removed from staging.
-        self.assertIsNone(conductor.workflow_state.get_staged_task(task_name, task_route))
-
-        # Assert the workflow succeeded.
-        self.assertEqual(conductor.get_workflow_status(), statuses.SUCCEEDED)
+        expected_errors = {
+            'semantics': [
+                {
+                    'message': 'The action property is required for with items task.',
+                    'schema_path': 'properties.tasks.patternProperties.^\\w+$',
+                    'spec_path': 'tasks.task1'
+                }
+            ]
+        }
 
-        # Assert the workflow output is correct.
-        conductor.render_workflow_output()
-        expected_output = {'items': [None] * 4}
-        self.assertDictEqual(conductor.get_workflow_output(), expected_output)
+        spec = native_specs.WorkflowSpec(wf_def)
+        self.assertDictEqual(spec.inspect(), expected_errors)
 
     def test_basic_items_list(self):
         wf_def = """