Merge "Add basic sanity tests for unordered flow"

taskflow/tests/unit/test_linear_flow.py

@@ -31,31 +31,10 @@ from taskflow.tests import utils
 class LinearFlowTest(test.TestCase):
     def _make_engine(self, flow):
         e = eng.SingleThreadedActionEngine(flow)
-        e.compile()
         e.storage.inject([('context', {})])
+        e.compile()
         return e
 
-    def make_reverting_task(self, token, blowup=False):
-
-        def do_revert(context, *args, **kwargs):
-            context[token] = 'reverted'
-
-        if blowup:
-
-            @decorators.task(name='blowup_%s' % token)
-            def blow_up(context, *args, **kwargs):
-                raise Exception("I blew up")
-
-            return blow_up
-        else:
-
-            @decorators.task(revert=do_revert,
-                             name='do_apply_%s' % token)
-            def do_apply(context, *args, **kwargs):
-                context[token] = 'passed'
-
-            return do_apply
-
     def test_result_access(self):
 
         @decorators.task(provides=['a', 'b'])
@@ -106,8 +85,8 @@ class LinearFlowTest(test.TestCase):
                 task_changes.append(state)
 
         wf = lw.Flow("the-test-action")
-        wf.add(self.make_reverting_task(2, False))
-        wf.add(self.make_reverting_task(1, True))
+        wf.add(utils.make_reverting_task(2, False))
+        wf.add(utils.make_reverting_task(1, True))
 
         e = self._make_engine(wf)
         e.notifier.register('*', listener)
@@ -144,7 +123,7 @@ class LinearFlowTest(test.TestCase):
             changes.append(state)
 
         wf = lw.Flow("the-test-action")
-        wf.add(self.make_reverting_task(1))
+        wf.add(utils.make_reverting_task(1))
 
         e = self._make_engine(wf)
         e.notifier.register('*', listener)
@@ -155,7 +134,7 @@ class LinearFlowTest(test.TestCase):
     def test_happy_flow(self):
         wf = lw.Flow("the-test-action")
         for i in range(0, 10):
-            wf.add(self.make_reverting_task(i))
+            wf.add(utils.make_reverting_task(i))
 
         e = self._make_engine(wf)
         capture_func, captured = self._capture_states()
@@ -183,8 +162,8 @@ class LinearFlowTest(test.TestCase):
 
     def test_reverting_flow(self):
         wf = lw.Flow("the-test-action")
-        wf.add(self.make_reverting_task(1))
-        wf.add(self.make_reverting_task(2, True))
+        wf.add(utils.make_reverting_task(1))
+        wf.add(utils.make_reverting_task(2, True))
 
         capture_func, captured = self._capture_states()
         e = self._make_engine(wf)

taskflow/tests/unit/test_unordered_flow.py

@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+#    Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
+#
+#    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 taskflow import decorators
+from taskflow.engines.action_engine import engine as eng
+from taskflow.patterns import unordered_flow as uf
+from taskflow import test
+
+from taskflow.tests import utils
+
+
+class UnorderedFlowTest(test.TestCase):
+    def _make_engine(self, flow):
+        e = eng.SingleThreadedActionEngine(flow)
+        e.storage.inject([('context', {})])
+        e.compile()
+        return e
+
+    def test_result_access(self):
+
+        @decorators.task(provides=['a', 'b'])
+        def do_apply1(context):
+            return [1, 2]
+
+        wf = uf.Flow("the-test-action")
+        wf.add(do_apply1)
+
+        e = self._make_engine(wf)
+        e.run()
+        data = e.storage.fetch_all()
+        self.assertIn('a', data)
+        self.assertIn('b', data)
+        self.assertEquals(2, data['b'])
+        self.assertEquals(1, data['a'])
+
+    def test_reverting_flow(self):
+        wf = uf.Flow("the-test-action")
+        wf.add(utils.make_reverting_task('1'))
+        wf.add(utils.make_reverting_task('2', blowup=True))
+        e = self._make_engine(wf)
+        self.assertRaises(Exception, e.run)
+
+    def test_functor_flow(self):
+
+        @decorators.task(provides=['a', 'b', 'c'])
+        def do_apply1(context):
+            context['1'] = True
+            return ['a', 'b', 'c']
+
+        @decorators.task
+        def do_apply2(context, **kwargs):
+            context['2'] = True
+
+        wf = uf.Flow("the-test-action")
+        wf.add(do_apply1)
+        wf.add(do_apply2)
+        e = self._make_engine(wf)
+        e.run()
+        self.assertEquals(2, len(e.storage.fetch('context')))

taskflow/tests/utils.py

@@ -16,6 +16,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from taskflow import decorators
 from taskflow import task
 
 ARGS_KEY = '__args__'
@@ -39,6 +40,27 @@ def drain(lst):
         lst.pop()
 
 
+def make_reverting_task(token, blowup=False):
+
+    def do_revert(context, *args, **kwargs):
+        context[token] = 'reverted'
+
+    if blowup:
+
+        @decorators.task(name='blowup_%s' % token)
+        def blow_up(context, *args, **kwargs):
+            raise Exception("I blew up")
+
+        return blow_up
+    else:
+
+        @decorators.task(revert=do_revert, name='do_apply_%s' % token)
+        def do_apply(context, *args, **kwargs):
+            context[token] = 'passed'
+
+        return do_apply
+
+
 class ProvidesRequiresTask(task.Task):
     def __init__(self, name, provides, requires, return_tuple=True):
         super(ProvidesRequiresTask, self).__init__(name=name,