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

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

@@ -19,6 +19,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
@@ -71,7 +72,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,6 +20,7 @@
 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):
@@ -76,6 +77,25 @@ 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, input_value):
         inputs = {'a1': input_value}
         expected_output = {'a5': inputs['a1'], 'b5': inputs['a1']}
@@ -119,11 +139,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']))