Fix schema of with items task to not allow additional properties

CHANGELOG.rst

@@ -36,6 +36,7 @@ Fixed
   must be called separately. (bug fix)
 * When inspecting custom YAQL/Jinja function to see if there is a context arg, use getargspec
   for py2 and getfullargspec for py3. (bug fix)
+* Check syntax on with items task to ensure action is indented correctly. Fixes #184 (bug fix)
 
 1.0.0
 -----

orquesta/specs/native/v1/models.py

@@ -111,7 +111,8 @@ class ItemizedSpec(native_v1_specs.Spec):
                 'pattern': _items_regex
             },
             'concurrency': spec_types.STRING_OR_POSITIVE_INTEGER
-        }
+        },
+        'additionalProperties': False
     }
 
     _context_evaluation_sequence = [
@@ -377,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 = []
 
@@ -514,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/base.py

@@ -364,12 +364,15 @@ class WorkflowConductorWithItemsTest(WorkflowConductorTest):
 
     def assert_task_items(self, conductor, task_id, task_route, task_ctx, items, action_specs,
                           mock_ac_ex_statuses, expected_task_statuses, expected_workflow_statuses,
-                          concurrency=None):
+                          concurrency=None, mock_ac_ex_results=None):
 
         # Set up test cases.
         tests = list(zip(mock_ac_ex_statuses, expected_task_statuses, expected_workflow_statuses))
         tk_ex_result = [None] * len(items)
 
+        if mock_ac_ex_results is None:
+            mock_ac_ex_results = items
+
         # Verify the first set of action executions.
         expected_task = self.format_task_item(
             task_id,
@@ -431,7 +434,7 @@ def assert_task_items(self, conductor, task_id, task_route, task_ctx, items, act
 
         # Mock the action execution for each item.
         for item_id in range(0, len(tests)):
-            ac_ex_result = items[item_id]
+            ac_ex_result = mock_ac_ex_results[item_id]
             tk_ex_result[item_id] = ac_ex_result
             ac_ex_status = tests[item_id][0]
 

orquesta/tests/unit/conducting/test_workflow_conductor_with_items.py

@@ -78,6 +78,89 @@ def test_empty_items_list(self):
         expected_output = {'items': []}
         self.assertDictEqual(conductor.get_workflow_output(), expected_output)
 
+    def test_bad_with_items_syntax(self):
+        wf_def = """
+        version: 1.0
+
+        vars:
+          - xs:
+              - fee
+              - fi
+              - fo
+              - fum
+
+        tasks:
+          task1:
+            with:
+                items: <% ctx(xs) %>
+                action: core.echo message=<% item() %>
+            next:
+              - publish:
+                  - items: <% result() %>
+
+        output:
+          - items: <% ctx(items) %>
+        """
+
+        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)',
+                    'schema_path': (
+                        'properties.tasks.patternProperties.^\\w+$.'
+                        'properties.with.additionalProperties'
+                    ),
+                    'spec_path': 'tasks.task1.with'
+                }
+            ]
+        }
+
+        spec = native_specs.WorkflowSpec(wf_def)
+        self.assertDictEqual(spec.inspect(), expected_errors)
+
+    def test_with_items_that_is_action_less(self):
+        wf_def = """
+        version: 1.0
+
+        vars:
+          - xs:
+              - fee
+              - fi
+              - fo
+              - fum
+
+        tasks:
+          task1:
+            with:
+                items: <% ctx(xs) %>
+            next:
+              - publish:
+                  - items: <% result() %>
+
+        output:
+          - items: <% ctx(items) %>
+        """
+
+        expected_errors = {
+            'semantics': [
+                {
+                    'message': 'The action property is required for with items task.',
+                    'schema_path': 'properties.tasks.patternProperties.^\\w+$',
+                    'spec_path': 'tasks.task1'
+                }
+            ]
+        }
+
+        spec = native_specs.WorkflowSpec(wf_def)
+        self.assertDictEqual(spec.inspect(), expected_errors)
+
     def test_basic_items_list(self):
         wf_def = """
         version: 1.0