Add tests for serialization error check.

flowy/serialization.py

@@ -18,30 +18,100 @@
 
 
 def dumps(value):
-    return json.dumps(_tag(value))
+    """Serialize the data structure and checks for errors or placeholders.
 
-def loads(value):
-    return json.loads(value, object_hook=_obj_hook)
+    Returns a 3-tuple: serialized data, oldest error, placehoders flag
+
+    Serializing values should work as expected
+    >>> dumps(1)
+    ('1', None, False)
+    >>> dumps(u'abc')
+    ('"abc"', None, False)
+    >>> dumps([1, 2, 3, [4]])
+    ('[1, 2, 3, [4]]', None, False)
+
+    >>> from flowy.result import error, placeholder, result
+    >>> r0 = result(u'r0', 0)
+    >>> e1 = error('err1', 1)
+    >>> e2 = error('err2', 2)
+    >>> e3 = error('err3', 3)
+    >>> r4 = result(u'r4', 4)
+    >>> ph = placeholder()
+
+    Results work just like values
+    >>> dumps([r0, r4])
+    ('["r0", "r4"]', None, False)
+    >>> dumps({r0: r4})
+    ('{"r0": "r4"}', None, False)
+
+    Any placeholder should be detected
+    >>> dumps(ph)
+    None, None, True
+    >>> dumps([r0, [r4, [ph]]])
+    None, None, True
+    >>> dumps({'a': {r0: {ph: 'b'}}})
+    None, None, True
+    >>> dumps([[[ph], ph]])
+    None, None, True
+
+    (Oldest) Error has priority over placeholders flag
+    >>> r = dumps(e1)
+    >>> r[0], r[1] is e1, r[2]
+    None, True, False
+
+    >>> r = dumps([e3, e1, e2])
+    >>> r[0], r[1] is e1, r[2]
+    None, True, False
 
+    >>> r = dumps([e3, [e1, [e2]]])
+    >>> r[0], r[1] is e1, r[2]
+    None, True, False
 
-def _tag(value):
+    >>> r = dumps([r0, e2, r4)
+    >>> r[0], r[1] is e2, r[2]
+    None, True, False
+
+    >>> r = dumps({r0: {'xyz': {e3: {e2: e1}}}})
+    >>> r[0], r[1] is e1, r[2]
+    None, True, False
+
+    """
+    traversed_value, error, has_placehoders = _traverse(value)
+    if error:
+        return None, error, False
+    if has_placehoders:
+        return None, None, True
+    else:
+        return json.dumps(traversed_value), None, False
+
+def _traverse(value):
     if is_result_proxy(value):
         value = value.__wrapped__
     if isinstance(value, tuple):
-        return [_tag(x) for x in value]
+        return [_traverse(x) for x in value]
     elif isinstance(value, uuid.UUID):
         return {' u': value.hex}
     elif isinstance(value, bytes):
         return {' b': b64encode(value).decode('ascii')}
     elif callable(getattr(value, '__json__', None)):
-        return _tag(value.__json__())
+        return _traverse(value.__json__())
     elif isinstance(value, list):
-        return [_tag(x) for x in value]
+        return [_traverse(x) for x in value]
     elif isinstance(value, dict):
-        return dict((k, _tag(v)) for k, v in value.items())
+        d = {}
+        for k, v in value.items():
+            if is_result_proxy(k):
+                k = k.__wrapped__
+            v = _traverse(v)
+            d[k] = v
+        return d
     return value
 
 
+def loads(value):
+    return json.loads(value, object_hook=_obj_hook)
+
+
 def _obj_hook(obj):
     if len(obj) != 1:
         return obj

flowy/tests/swf_cases.py

@@ -52,6 +52,9 @@
 worker.register(task_activity_workflow, First2, version=1)
 worker.register(task_red_activities_workflow, ParallelReduce, version=1)
 worker.register(task_red_activities_workflow, ParallelReduceCombined, version=1)
+worker.register(task_activity_workflow, ArgsStructErrors, version=1)
+worker.register(task_activity_workflow, ArgsStructErrorsHandled, version=1)
+
 
 cases = [
     {'name': 'NotFound',
@@ -576,5 +579,38 @@
                  'input_args': ['c', 'xyz'],
              }, ],
          },
+     }, {
+         'name': 'ArgsStructErrors',
+         'version': 1,
+         'errors': {'task-0-0': 'Err1!', },
+         'expected': {'fail': 'Err!', },
+         'running': ['task-1-0', ],
+     }, {
+         'name': 'ArgsStructErrors',
+         'version': 1,
+         'errors': {
+             'task-0-0': 'Err1!',
+             'task-1-0': 'Err2!',
+          },
+         'order': ['task-1-0', 'task-0-0'],
+         'expected': {'fail': 'Err2!', },
+     }, {
+         'name': 'ArgsStructErrors',
+         'version': 1,
+         'errors': {
+             'task-0-0': 'Err1!',
+             'task-1-0': 'Err2!',
+          },
+         'order': ['task-0-0', 'task-1-0'],
+         'expected': {'fail': 'Err1!', },
+     }, {
+         'name': 'ArgsStructErrorsHandled',
+         'version': 1,
+         'errors': {
+             'task-0-0': 'Err1!',
+             'task-1-0': 'Err2!',
+          },
+         'order': ['task-0-0', 'task-1-0'],
+         'expected': {'finish': 8},
      }
 ]

flowy/tests/workflows.py

@@ -27,6 +27,30 @@ def __call__(self, a, b, c=1, d=2):
         return a, b, c, d
 
 
+class ArgsStructErrors(object):
+    def __init__(self, task):
+        self.task = task
+
+    def __call__(self):
+        a = self.task()
+        b = self.task()
+        return self.task([[b], a])
+
+
+class ArgsStructErrorsHandled(object):
+    def __init__(self, task):
+        self.task = task
+
+    def __call__(self):
+        from flowy import TaskError
+        a = self.task()
+        b = self.task()
+        try:
+            return wait(self.task([[b], a]))
+        except TaskError:
+            return 8
+
+
 class Dependency(object):
     def __init__(self, task):
         self.task = task