Merge "Working on "join": allowed value "one" for "join" property"

mistral/tests/unit/workbook/v2/test_dsl_specs_v2.py

@@ -124,6 +124,10 @@ workflows:
       task10:
         join: 2
         action: std.echo output="Task 10 echo"
+
+      task11:
+        join: one
+        action: std.echo output="Task 11 echo"
 """
 
 
@@ -244,7 +248,7 @@ class DSLv2ModelTest(base.BaseTest):
         self.assertEqual('wf2', wf2_spec.get_name())
         self.assertListEqual(['test', 'v2'], wf2_spec.get_tags())
         self.assertEqual('direct', wf2_spec.get_type())
-        self.assertEqual(8, len(wf2_spec.get_tasks()))
+        self.assertEqual(9, len(wf2_spec.get_tasks()))
 
         task_defaults_spec = wf2_spec.get_task_defaults()
 
@@ -325,6 +329,10 @@ class DSLv2ModelTest(base.BaseTest):
 
         self.assertEqual(2, task10_spec.get_join())
 
+        task11_spec = wf2_spec.get_tasks().get('task11')
+
+        self.assertEqual('one', task11_spec.get_join())
+
     def test_adhoc_action_with_base_in_one_string(self):
         wb_spec = spec_parser.get_workbook_spec_from_yaml(VALID_WB)
 

mistral/workbook/v2/tasks.py

@@ -68,20 +68,42 @@ class TaskSpec(base.BaseSpec):
         self._on_success = self._as_list_of_tuples('on-success')
         self._on_error = self._as_list_of_tuples('on-error')
 
-        self._process_for_each()
         self._process_action_and_workflow()
 
-    def _process_action_and_workflow(self):
-        if self._action and self._workflow:
+    def validate(self):
+        super(TaskSpec, self).validate()
+
+        if 'join' in self._data:
+            join = self._data.get('join')
+
+            if not (isinstance(join, int) or join in ['all', 'one']):
+                msg = ("Task property 'join' is only allowed to be an"
+                       " integer, 'all' or 'one': %s" % self._data)
+                raise exc.InvalidModelException(msg)
+
+        action = self._data.get('action')
+        workflow = self._data.get('workflow')
+
+        if action and workflow:
             msg = ("Task properties 'action' and 'workflow' can't be"
-                   " specified both:" % self._data)
+                   " specified both: %s" % self._data)
             raise exc.InvalidModelException(msg)
 
-        if not self._action and not self._workflow:
+        if not action and not workflow:
             msg = ("One of task properties 'action' or 'workflow' must be"
-                   " specified:" % self._data)
+                   " specified: %s" % self._data)
             raise exc.InvalidModelException(msg)
 
+        for_each = self._data.get('for-each')
+
+        if for_each:
+            for _, v in for_each.iteritems():
+                if not isinstance(v, (list, six.string_types)):
+                    msg = ("Items of task property 'for-each' can only be "
+                           "a list or an expression string: %s" % self._data)
+                    raise exc.InvalidModelException(msg)
+
+    def _process_action_and_workflow(self):
         params = {}
 
         if self._action:
@@ -92,14 +114,6 @@ class TaskSpec(base.BaseSpec):
 
         utils.merge_dicts(self._input, params)
 
-    def _process_for_each(self):
-        if self._for_each:
-            for key, value in self._for_each.items():
-                if not isinstance(value, (list, six.string_types)):
-                    msg = ("Items of task property 'for-each' can only be "
-                           "a list or an expression string: %s" % self._data)
-                    raise exc.InvalidModelException(msg)
-
     def get_name(self):
         return self._name