Merge pull request #104 from StackStorm/fix-app-ctx

Fix vars and input that references app context

orquesta/conducting.py

@@ -192,9 +192,19 @@ def flow(self):
         if not self._flow:
             self._flow = TaskFlow()
 
-            # Render runtime and default workflow inputs.
-            rendered_inputs, input_errors = self.spec.render_input(self.get_workflow_input())
-            rendered_vars, var_errors = self.spec.render_vars(rendered_inputs)
+            # Set any given context as the initial context.
+            init_ctx = self.get_workflow_parent_context()
+
+            # Render workflow inputs and merge into the initial context.
+            workflow_input = self.get_workflow_input()
+            rendered_inputs, input_errors = self.spec.render_input(workflow_input, init_ctx)
+            init_ctx = dx.merge_dicts(init_ctx, rendered_inputs, True)
+
+            # Render workflow variables and merge into the initial context.
+            rendered_vars, var_errors = self.spec.render_vars(init_ctx)
+            init_ctx = dx.merge_dicts(init_ctx, rendered_vars, True)
+
+            # Fail workflow if there are errors.
             errors = input_errors + var_errors
 
             if errors:
@@ -204,8 +214,6 @@ def flow(self):
             # Proceed if there is no issue with rendering of inputs and vars.
             if self.get_workflow_state() not in states.ABENDED_STATES:
                 # Set the initial workflow context.
-                init_ctx = dict(list(rendered_inputs.items()) + list(rendered_vars.items()))
-                init_ctx = dx.merge_dicts(init_ctx, self.get_workflow_parent_context(), True)
                 self._flow.contexts.append({'srcs': [], 'value': init_ctx})
 
                 # Identify the starting tasks and set the pointer to the initial context entry.

orquesta/specs/mistral/v2/workflows.py

@@ -68,7 +68,7 @@ class WorkflowSpec(base.Spec):
         'vars'
     ]
 
-    def render_input(self, runtime_inputs):
+    def render_input(self, runtime_inputs, in_ctx=None):
         input_specs = getattr(self, 'input') or []
         default_inputs = dict([list(i.items())[0] for i in input_specs if isinstance(i, dict)])
         merged_inputs = dx.merge_dicts(default_inputs, runtime_inputs, True)

orquesta/specs/native/v1/models.py

@@ -533,8 +533,8 @@ def __init__(self, spec, name=None, member=False):
 
         super(WorkflowSpec, self).__init__(spec, name=name, member=member)
 
-    def render_input(self, runtime_inputs):
-        rolling_ctx = {}
+    def render_input(self, runtime_inputs, in_ctx=None):
+        rolling_ctx = copy.deepcopy(in_ctx) if in_ctx else {}
         errors = []
 
         for input_spec in (getattr(self, 'input') or []):

orquesta/tests/unit/conducting/test_workflow_conductor_context.py

@@ -0,0 +1,116 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from orquesta import conducting
+from orquesta import events
+from orquesta.specs import native as specs
+from orquesta import states
+from orquesta.tests.unit import base
+
+
+class WorkflowConductorContextTest(base.WorkflowConductorTest):
+
+    def test_bad_app_ctx_references(self):
+        wf_def = """
+        version: 1.0
+
+        input:
+          - a: <% ctx().x %>
+
+        vars:
+          - b: <% ctx().y %>
+
+        output:
+          - x: <% ctx().a %>
+          - y: <% ctx().b %>
+          - z: <% ctx().z %>
+
+        tasks:
+          task1:
+            action: core.noop
+        """
+
+        expected_errors = [
+            {
+                'type': 'error',
+                'message': (
+                    'YaqlEvaluationException: Unable to resolve key \'x\' in '
+                    'expression \'<% ctx().x %>\' from context.'
+                )
+            },
+            {
+                'type': 'error',
+                'message': (
+                    'YaqlEvaluationException: Unable to resolve key \'y\' in '
+                    'expression \'<% ctx().y %>\' from context.'
+                )
+            }
+        ]
+
+        spec = specs.WorkflowSpec(wf_def)
+
+        # Run the workflow.
+        conductor = conducting.WorkflowConductor(spec)
+        conductor.request_workflow_state(states.RUNNING)
+
+        # Check workflow status and result.
+        self.assertEqual(conductor.get_workflow_state(), states.FAILED)
+        self.assertListEqual(conductor.errors, expected_errors)
+        self.assertIsNone(conductor.get_workflow_output())
+
+    def test_app_ctx_references(self):
+        app_ctx = {
+            'x': 'foobar',
+            'y': 'fubar',
+            'z': 'phobar'
+        }
+
+        wf_def = """
+        version: 1.0
+
+        input:
+          - a: <% ctx().x %>
+
+        vars:
+          - b: <% ctx().y %>
+
+        output:
+          - x: <% ctx().a %>
+          - y: <% ctx().b %>
+          - z: <% ctx().z %>
+
+        tasks:
+          task1:
+            action: core.noop
+        """
+
+        expected_output = app_ctx
+        expected_errors = []
+
+        spec = specs.WorkflowSpec(wf_def)
+        self.assertDictEqual(spec.inspect(app_ctx=app_ctx), {})
+
+        # Run the workflow.
+        conductor = conducting.WorkflowConductor(spec, context=app_ctx)
+        conductor.request_workflow_state(states.RUNNING)
+        self.assertEqual(conductor.get_workflow_state(), states.RUNNING)
+        self.assertListEqual(conductor.errors, expected_errors)
+
+        # Complete tasks
+        task_name = 'task1'
+        conductor.update_task_flow(task_name, events.ActionExecutionEvent(states.RUNNING))
+        conductor.update_task_flow(task_name, events.ActionExecutionEvent(states.SUCCEEDED))
+
+        # Check workflow status and output.
+        self.assertEqual(conductor.get_workflow_state(), states.SUCCEEDED)
+        self.assertListEqual(conductor.errors, expected_errors)
+        self.assertDictEqual(conductor.get_workflow_output(), expected_output)