Merge "Adding a revert_all option to retry controllers"

1 files changed, 251 insertions, 0 deletions

diff --git a/taskflow/tests/unit/test_retries.py b/taskflow/tests/unit/test_retries.py
index edcc6d8..ddb256b 100644
--- a/taskflow/tests/unit/test_retries.py
+++ b/taskflow/tests/unit/test_retries.py
@@ -336,6 +336,66 @@ class RetryTest(utils.EngineTestBase):
                     'flow-1.f SUCCESS']
         self.assertEqual(expected, capturer.values)
 
+    def test_nested_flow_with_retry_revert(self):
+        retry1 = retry.Times(0, 'r1', provides='x2')
+        flow = lf.Flow('flow-1').add(
+            utils.ProgressingTask("task1"),
+            lf.Flow('flow-2', retry1).add(
+                utils.ConditionalTask("task2", inject={'x': 1}))
+        )
+        engine = self._make_engine(flow)
+        engine.storage.inject({'y': 2})
+        with utils.CaptureListener(engine) as capturer:
+            try:
+                engine.run()
+            except Exception:
+                pass
+        self.assertEqual(engine.storage.fetch_all(), {'y': 2})
+        expected = ['flow-1.f RUNNING',
+                    'task1.t RUNNING',
+                    'task1.t SUCCESS(5)',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(1)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot!)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r REVERTING',
+                    'r1.r REVERTED',
+                    'flow-1.f REVERTED']
+        self.assertEqual(expected, capturer.values)
+
+    def test_nested_flow_with_retry_revert_all(self):
+        retry1 = retry.Times(0, 'r1', provides='x2', revert_all=True)
+        flow = lf.Flow('flow-1').add(
+            utils.ProgressingTask("task1"),
+            lf.Flow('flow-2', retry1).add(
+                utils.ConditionalTask("task2", inject={'x': 1}))
+        )
+        engine = self._make_engine(flow)
+        engine.storage.inject({'y': 2})
+        with utils.CaptureListener(engine) as capturer:
+            try:
+                engine.run()
+            except Exception:
+                pass
+        self.assertEqual(engine.storage.fetch_all(), {'y': 2})
+        expected = ['flow-1.f RUNNING',
+                    'task1.t RUNNING',
+                    'task1.t SUCCESS(5)',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(1)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot!)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r REVERTING',
+                    'r1.r REVERTED',
+                    'task1.t REVERTING',
+                    'task1.t REVERTED',
+                    'flow-1.f REVERTED']
+        self.assertEqual(expected, capturer.values)
+
     def test_revert_all_retry(self):
         flow = lf.Flow('flow-1', retry.Times(3, 'r1', provides='x')).add(
             utils.ProgressingTask("task1"),
@@ -594,6 +654,108 @@ class RetryTest(utils.EngineTestBase):
                     'flow-1.f REVERTED']
         self.assertItemsEqual(capturer.values, expected)
 
+    def test_nested_for_each_revert(self):
+        collection = [3, 2, 3, 5]
+        retry1 = retry.ForEach(collection, 'r1', provides='x')
+        flow = lf.Flow('flow-1').add(
+            utils.ProgressingTask("task1"),
+            lf.Flow('flow-2', retry1).add(
+                utils.FailingTaskWithOneArg('task2')
+            )
+        )
+        engine = self._make_engine(flow)
+        with utils.CaptureListener(engine) as capturer:
+            self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
+        expected = ['flow-1.f RUNNING',
+                    'task1.t RUNNING',
+                    'task1.t SUCCESS(5)',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(3)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot with 3)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(2)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot with 2)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(3)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot with 3)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(5)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot with 5)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r REVERTING',
+                    'r1.r REVERTED',
+                    'flow-1.f REVERTED']
+        self.assertEqual(expected, capturer.values)
+
+    def test_nested_for_each_revert_all(self):
+        collection = [3, 2, 3, 5]
+        retry1 = retry.ForEach(collection, 'r1', provides='x', revert_all=True)
+        flow = lf.Flow('flow-1').add(
+            utils.ProgressingTask("task1"),
+            lf.Flow('flow-2', retry1).add(
+                utils.FailingTaskWithOneArg('task2')
+            )
+        )
+        engine = self._make_engine(flow)
+        with utils.CaptureListener(engine) as capturer:
+            self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
+        expected = ['flow-1.f RUNNING',
+                    'task1.t RUNNING',
+                    'task1.t SUCCESS(5)',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(3)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot with 3)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(2)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot with 2)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(3)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot with 3)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(5)',
+                    'task2.t RUNNING',
+                    'task2.t FAILURE(Failure: RuntimeError: Woot with 5)',
+                    'task2.t REVERTING',
+                    'task2.t REVERTED',
+                    'r1.r REVERTING',
+                    'r1.r REVERTED',
+                    'task1.t REVERTING',
+                    'task1.t REVERTED',
+                    'flow-1.f REVERTED']
+        self.assertEqual(expected, capturer.values)
+
     def test_for_each_empty_collection(self):
         values = []
         retry1 = retry.ForEach(values, 'r1', provides='x')
@@ -674,6 +836,95 @@ class RetryTest(utils.EngineTestBase):
                     'flow-1.f REVERTED']
         self.assertItemsEqual(capturer.values, expected)
 
+    def test_nested_parameterized_for_each_revert(self):
+        values = [3, 2, 5]
+        retry1 = retry.ParameterizedForEach('r1', provides='x')
+        flow = lf.Flow('flow-1').add(
+            utils.ProgressingTask('task-1'),
+            lf.Flow('flow-2', retry1).add(
+                utils.FailingTaskWithOneArg('task-2')
+            )
+        )
+        engine = self._make_engine(flow)
+        engine.storage.inject({'values': values, 'y': 1})
+        with utils.CaptureListener(engine) as capturer:
+            self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
+        expected = ['flow-1.f RUNNING',
+                    'task-1.t RUNNING',
+                    'task-1.t SUCCESS(5)',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(3)',
+                    'task-2.t RUNNING',
+                    'task-2.t FAILURE(Failure: RuntimeError: Woot with 3)',
+                    'task-2.t REVERTING',
+                    'task-2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task-2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(2)',
+                    'task-2.t RUNNING',
+                    'task-2.t FAILURE(Failure: RuntimeError: Woot with 2)',
+                    'task-2.t REVERTING',
+                    'task-2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task-2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(5)',
+                    'task-2.t RUNNING',
+                    'task-2.t FAILURE(Failure: RuntimeError: Woot with 5)',
+                    'task-2.t REVERTING',
+                    'task-2.t REVERTED',
+                    'r1.r REVERTING',
+                    'r1.r REVERTED',
+                    'flow-1.f REVERTED']
+        self.assertEqual(expected, capturer.values)
+
+    def test_nested_parameterized_for_each_revert_all(self):
+        values = [3, 2, 5]
+        retry1 = retry.ParameterizedForEach('r1', provides='x',
+                                            revert_all=True)
+        flow = lf.Flow('flow-1').add(
+            utils.ProgressingTask('task-1'),
+            lf.Flow('flow-2', retry1).add(
+                utils.FailingTaskWithOneArg('task-2')
+            )
+        )
+        engine = self._make_engine(flow)
+        engine.storage.inject({'values': values, 'y': 1})
+        with utils.CaptureListener(engine) as capturer:
+            self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
+        expected = ['flow-1.f RUNNING',
+                    'task-1.t RUNNING',
+                    'task-1.t SUCCESS(5)',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(3)',
+                    'task-2.t RUNNING',
+                    'task-2.t FAILURE(Failure: RuntimeError: Woot with 3)',
+                    'task-2.t REVERTING',
+                    'task-2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task-2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(2)',
+                    'task-2.t RUNNING',
+                    'task-2.t FAILURE(Failure: RuntimeError: Woot with 2)',
+                    'task-2.t REVERTING',
+                    'task-2.t REVERTED',
+                    'r1.r RETRYING',
+                    'task-2.t PENDING',
+                    'r1.r RUNNING',
+                    'r1.r SUCCESS(5)',
+                    'task-2.t RUNNING',
+                    'task-2.t FAILURE(Failure: RuntimeError: Woot with 5)',
+                    'task-2.t REVERTING',
+                    'task-2.t REVERTED',
+                    'r1.r REVERTING',
+                    'r1.r REVERTED',
+                    'task-1.t REVERTING',
+                    'task-1.t REVERTED',
+                    'flow-1.f REVERTED']
+        self.assertEqual(expected, capturer.values)
+
     def test_parameterized_for_each_empty_collection(self):
         values = []
         retry1 = retry.ParameterizedForEach('r1', provides='x')