Fix retry policy

* After full refactoring policies were broken

Partially Implements blueprint mistral-refactor-task-output

Change-Id: I851b1e5f1982b142eea640b1b033e63be01bfbec

mistral/engine1/policies.py

@@ -264,10 +264,6 @@ class RetryPolicy(base.TaskPolicy):
         """
         super(RetryPolicy, self).after_task_complete(task_ex, task_spec)
 
-        # TODO(rakhmerov): This policy needs to be fixed.
-        if True:
-            return
-
         context_key = 'retry_task_policy'
 
         runtime_context = _ensure_context_has_key(
@@ -288,8 +284,6 @@ class RetryPolicy(base.TaskPolicy):
             % (task_ex.name, task_ex.state)
         )
 
-        outbound_context = task_ex.result
-
         policy_context = runtime_context[context_key]
 
         retry_no = 0
@@ -300,12 +294,7 @@ class RetryPolicy(base.TaskPolicy):
 
         retries_remain = retry_no + 1 < self.count
 
-        break_early = (
-            expressions.evaluate(self.break_on, outbound_context)
-            if self.break_on and outbound_context else False
-        )
-
-        if not retries_remain or break_early:
+        if not retries_remain or self.break_on:
             return
 
         _log_task_delay(task_ex, self.delay)
@@ -316,10 +305,10 @@ class RetryPolicy(base.TaskPolicy):
         runtime_context[context_key] = policy_context
 
         scheduler.schedule_call(
-            _ENGINE_CLIENT_PATH,
-            'run_task',
+            None,
+            _RUN_TASK_EXECUTION_PATH,
             self.delay,
-            task_id=task_ex.id
+            task_ex_id=task_ex.id,
         )
 
 

mistral/tests/unit/engine1/test_policies.py

@@ -379,7 +379,6 @@ class PoliciesTest(base.EngineTestCase):
         )
         self._await(lambda: self.is_task_success(task_ex.id))
 
-    @testtools.skip("Fix 'retry' policy.")
     def test_retry_policy(self):
         wb_service.create_workbook_v2(RETRY_WB)