Merge pull request #191 from userlocalhost/bugfix/issue-176/evaluate-yaql-context

Fixed a problem not to be able to parse input value which is type of dict in array in some YAQL operators.

Author: m4dcoder

CHANGELOG.rst

@@ -39,6 +39,9 @@ Fixed
 * Check syntax on with items task to ensure action is indented correctly. Fixes #184 (bug fix)
 * Fix variable inspection where ctx().get() method calls are identified as errors.
   Fixes StackStorm/st2#4866 (bug fix)
+* Fix a problem of TypeError orccuring when a list (or dict) value that contains unhashable typed
+  value (list or dict) is passed in some YAQL operators (e.g. distinct()). Fixes #176 (bug fix)
+  Contributed by Hiroyasu Ohyama (@userlocalhost)
 
 1.0.0
 -----

orquesta/expressions/functions/workflow.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import collections
+
 from orquesta import constants
 from orquesta import exceptions as exc
 from orquesta import statuses
@@ -115,8 +117,8 @@ def item_(context, key=None):
     if not key:
         return current_item
 
-    if not isinstance(current_item, dict):
-        raise exc.ExpressionEvaluationException('Item is not type of dict.')
+    if not isinstance(current_item, collections.Mapping):
+        raise exc.ExpressionEvaluationException('Item is not type of collections.Mapping.')
 
     if key not in current_item:
         raise exc.ExpressionEvaluationException('Item does not have key "%s".' % key)

orquesta/expressions/yql.py

@@ -20,6 +20,7 @@
 
 import yaql
 import yaql.language.exceptions as yaql_exc
+import yaql.language.utils as yaql_utils
 
 from orquesta import exceptions as exc
 from orquesta.expressions import base as expr_base
@@ -79,7 +80,15 @@ class YAQLEvaluator(expr_base.Evaluator):
     @classmethod
     def contextualize(cls, data):
         ctx = cls._root_ctx.create_child_context()
-        ctx['__vars'] = data or {}
+
+        # Some yaql expressions (e.g. distinct()) refer to hash value of variable.
+        # But some built-in Python type values (e.g. list and dict) don't have __hash__() method.
+        # The convert_input_data method parses specified variable and convert it to hashable one.
+        if isinstance(data, yaql_utils.SequenceType) or isinstance(data, yaql_utils.MappingType):
+            ctx['__vars'] = yaql_utils.convert_input_data(data)
+        else:
+            ctx['__vars'] = data or {}
+
         ctx['__state'] = ctx['__vars'].get('__state')
         ctx['__current_task'] = ctx['__vars'].get('__current_task')
         ctx['__current_item'] = ctx['__vars'].get('__current_item')

orquesta/tests/unit/conducting/native/test_task_rendering_for_with_items.py

@@ -82,7 +82,7 @@ def test_bad_item_type(self):
                 'type': 'error',
                 'message': (
                     'YaqlEvaluationException: Unable to evaluate expression \'<% item(x) %>\'. '
-                    'ExpressionEvaluationException: Item is not type of dict.'
+                    'ExpressionEvaluationException: Item is not type of collections.Mapping.'
                 ),
                 'task_id': 'task1',
                 'route': 0

orquesta/tests/unit/conducting/test_workflow_conductor_data_flow.py

@@ -20,47 +20,84 @@
 from orquesta.specs import native as native_specs
 from orquesta import statuses
 from orquesta.tests.unit import base as test_base
+import yaql.language.utils as yaql_utils
 
 
 class WorkflowConductorDataFlowTest(test_base.WorkflowConductorTest):
 
-    def _prep_conductor(self, context=None, inputs=None, status=None):
-        wf_def = """
-        version: 1.0
-
-        description: A basic sequential workflow.
-
-        input:
-          - a1
-          - b1: <% ctx().a1 %>
-
-        vars:
-          - a2: <% ctx().b1 %>
-          - b2: <% ctx().a2 %>
-
-        output:
-          - a5: <% ctx().b4 %>
-          - b5: <% ctx().a5 %>
-
-        tasks:
-          task1:
-            action: core.noop
-            next:
-              - when: <% succeeded() %>
-                publish:
-                  - a3: <% ctx().b2 %>
-                  - b3: <% ctx().a3 %>
-                do: task2
-          task2:
-            action: core.noop
-            next:
-              - when: <% succeeded() %>
-                publish: a4=<% ctx().b3 %> b4=<% ctx().a4 %>
-                do: task3
-          task3:
-            action: core.noop
-        """
-
+    wf_def_yaql = """
+    version: 1.0
+
+    description: A basic sequential workflow.
+
+    input:
+      - a1
+      - b1: <% ctx().a1 %>
+
+    vars:
+      - a2: <% ctx().b1 %>
+      - b2: <% ctx().a2 %>
+
+    output:
+      - a5: <% ctx().b4 %>
+      - b5: <% ctx().a5 %>
+
+    tasks:
+      task1:
+        action: core.noop
+        next:
+          - when: <% succeeded() %>
+            publish:
+              - a3: <% ctx().b2 %>
+              - b3: <% ctx().a3 %>
+            do: task2
+      task2:
+        action: core.noop
+        next:
+          - when: <% succeeded() %>
+            publish: a4=<% ctx().b3 %> b4=<% ctx().a4 %>
+            do: task3
+      task3:
+        action: core.noop
+    """
+
+    wf_def_jinja = """
+    version: 1.0
+
+    description: A basic sequential workflow.
+
+    input:
+      - a1
+      - b1: '{{ ctx("a1") }}'
+
+    vars:
+      - a2: '{{ ctx("b1") }}'
+      - b2: '{{ ctx("a2") }}'
+
+    output:
+      - a5: '{{ ctx("b4") }}'
+      - b5: '{{ ctx("a5") }}'
+
+    tasks:
+      task1:
+        action: core.noop
+        next:
+          - when: '{{ succeeded() }}'
+            publish:
+              - a3: '{{ ctx("b2") }}'
+              - b3: '{{ ctx("a3") }}'
+            do: task2
+      task2:
+        action: core.noop
+        next:
+          - when: '{{ succeeded() }}'
+            publish: a4='{{ ctx("b3") }}' b4='{{ ctx("a4") }}'
+            do: task3
+      task3:
+        action: core.noop
+    """
+
+    def _prep_conductor(self, wf_def, context=None, inputs=None, status=None):
         spec = native_specs.WorkflowSpec(wf_def)
         self.assertDictEqual(spec.inspect(), {})
 
@@ -76,33 +113,52 @@ def _prep_conductor(self, context=None, inputs=None, status=None):
 
         return conductor
 
+    def _get_combined_value(self, callstack_depth=0):
+        # This returns dict typed value all Python built-in type values
+        # which orquesta spec could accept.
+        if callstack_depth < 2:
+            return {
+                'null': None,
+                'integer_positive': 123,
+                'integer_negative': -123,
+                'number_positive': 99.99,
+                'number_negative': -99.99,
+                'string': 'xyz',
+                'boolean_true': True,
+                'boolean_false': False,
+                'array': list(self._get_combined_value(callstack_depth + 1).values()),
+                'object': self._get_combined_value(callstack_depth + 1),
+            }
+        else:
+            return {}
+
+    def _assert_data_flow(self, inputs, expected_output):
+        # This assert method checks input value would be handled and published
+        # as an expected type and value with both YAQL and Jinja expressions.
+        for wf_def in [self.wf_def_jinja, self.wf_def_yaql]:
+            conductor = self._prep_conductor(wf_def, inputs=inputs, status=statuses.RUNNING)
+
+            for i in range(1, len(conductor.spec.tasks) + 1):
+                task_name = 'task' + str(i)
+                forward_statuses = [statuses.RUNNING, statuses.SUCCEEDED]
+                self.forward_task_statuses(conductor, task_name, forward_statuses)
+
+            # Render workflow output and checkout workflow status and output.
+            conductor.render_workflow_output()
+            self.assertEqual(conductor.get_workflow_status(), statuses.SUCCEEDED)
+            self.assertDictEqual(conductor.get_workflow_output(), expected_output)
+
     def assert_data_flow(self, input_value):
         inputs = {'a1': input_value}
         expected_output = {'a5': inputs['a1'], 'b5': inputs['a1']}
-        conductor = self._prep_conductor(inputs=inputs, status=statuses.RUNNING)
 
-        for i in range(1, len(conductor.spec.tasks) + 1):
-            task_name = 'task' + str(i)
-            self.forward_task_statuses(conductor, task_name, [statuses.RUNNING, statuses.SUCCEEDED])
-
-        # Render workflow output and checkout workflow status and output.
-        conductor.render_workflow_output()
-        self.assertEqual(conductor.get_workflow_status(), statuses.SUCCEEDED)
-        self.assertDictEqual(conductor.get_workflow_output(), expected_output)
+        self._assert_data_flow(inputs, expected_output)
 
     def assert_unicode_data_flow(self, input_value):
         inputs = {u'a1': unicode(input_value, 'utf8') if six.PY2 else input_value}
         expected_output = {u'a5': inputs['a1'], u'b5': inputs['a1']}
-        conductor = self._prep_conductor(inputs=inputs, status=statuses.RUNNING)
-
-        for i in range(1, len(conductor.spec.tasks) + 1):
-            task_name = 'task' + str(i)
-            self.forward_task_statuses(conductor, task_name, [statuses.RUNNING, statuses.SUCCEEDED])
 
-        # Render workflow output and checkout workflow status and output.
-        conductor.render_workflow_output()
-        self.assertEqual(conductor.get_workflow_status(), statuses.SUCCEEDED)
-        self.assertDictEqual(conductor.get_workflow_output(), expected_output)
+        self._assert_data_flow(inputs, expected_output)
 
     def test_data_flow_string(self):
         self.assert_data_flow('xyz')
@@ -119,11 +175,20 @@ def test_data_flow_boolean(self):
         self.assert_data_flow(True)
         self.assert_data_flow(False)
 
+    def test_data_flow_none(self):
+        self.assert_data_flow(None)
+
     def test_data_flow_dict(self):
-        self.assert_data_flow({'x': 123, 'y': 'abc'})
+        mapping_typed_data = self._get_combined_value()
+
+        self.assertIsInstance(mapping_typed_data, yaql_utils.MappingType)
+        self.assert_data_flow(mapping_typed_data)
 
     def test_data_flow_list(self):
-        self.assert_data_flow([123, 'abc', True])
+        sequence_typed_data = list(self._get_combined_value().values())
+
+        self.assertIsInstance(sequence_typed_data, yaql_utils.SequenceType)
+        self.assert_data_flow(sequence_typed_data)
 
     def test_data_flow_unicode(self):
         self.assert_unicode_data_flow('光合作用')

orquesta/tests/unit/expressions/test_facade_yaql_evaluate.py

@@ -269,3 +269,24 @@ def test_custom_function_failure(self):
             expr_base.evaluate,
             expr
         )
+
+    def test_distinct_operator(self):
+        test_cases = [
+            {
+                'expr': '<% ctx(val).distinct() %>',
+                'input': {'val': [1, 2, 3, 1]},
+                'expect': [1, 2, 3]
+            },
+            {
+                'expr': '<% ctx(val).distinct() %>',
+                'input': {'val': [{'a': 1}, {'b': 2}, {'a': 1}]},
+                'expect': [{'a': 1}, {'b': 2}]
+            },
+            {
+                'expr': '<% ctx(val).distinct($[1]) %>',
+                'input': {'val': [['a', 1], ['b', 2], ['c', 1], ['a', 3]]},
+                'expect': [['a', 1], ['b', 2], ['a', 3]]
+            }
+        ]
+        for case in test_cases:
+            self.assertEqual(case['expect'], expr_base.evaluate(case['expr'], case['input']))