Fix "join" when the last indirect inbound task failed
* See bug description for the example that didn't work. It was caused by a simple mistake in a python expression of type "my_set = my_set or set()" that didn't work as expected, i.e. it created a new set even if my_set is already an empty set. So, the proper expression that's needed is "my_set = set() if my_set is None else my_set" Change-Id: I2a787921449fecf3301013a770ffe712e9606baf Closes-Bug: #1803677
This commit is contained in:
parent
a0c8da92dd
commit
c9e08a8839
@ -1293,3 +1293,58 @@ class JoinEngineTest(base.EngineTestCase):
|
|||||||
self.assertEqual(states.SUCCESS, task1.state)
|
self.assertEqual(states.SUCCESS, task1.state)
|
||||||
self.assertEqual(states.SUCCESS, task2.state)
|
self.assertEqual(states.SUCCESS, task2.state)
|
||||||
self.assertEqual(states.SUCCESS, task3.state)
|
self.assertEqual(states.SUCCESS, task3.state)
|
||||||
|
|
||||||
|
def test_join_last_inbound_indirect_error(self):
|
||||||
|
wf_text = """---
|
||||||
|
version: '2.0'
|
||||||
|
|
||||||
|
wf:
|
||||||
|
tasks:
|
||||||
|
task1:
|
||||||
|
action: std.noop
|
||||||
|
on-success:
|
||||||
|
- join_task
|
||||||
|
|
||||||
|
task2:
|
||||||
|
action: std.fail
|
||||||
|
wait-before: 2
|
||||||
|
on-success:
|
||||||
|
- task3
|
||||||
|
|
||||||
|
task3:
|
||||||
|
action: std.noop
|
||||||
|
on-success:
|
||||||
|
- join_task
|
||||||
|
|
||||||
|
join_task:
|
||||||
|
join: all
|
||||||
|
"""
|
||||||
|
|
||||||
|
wf_service.create_workflows(wf_text)
|
||||||
|
|
||||||
|
wf_ex = self.engine.start_workflow('wf')
|
||||||
|
|
||||||
|
self.await_workflow_error(wf_ex.id)
|
||||||
|
|
||||||
|
with db_api.transaction():
|
||||||
|
wf_ex = db_api.get_workflow_execution(wf_ex.id)
|
||||||
|
|
||||||
|
task_execs = wf_ex.task_executions
|
||||||
|
|
||||||
|
self.assertEqual(3, len(task_execs))
|
||||||
|
|
||||||
|
self._assert_single_item(
|
||||||
|
task_execs,
|
||||||
|
name='task1',
|
||||||
|
state=states.SUCCESS
|
||||||
|
)
|
||||||
|
self._assert_single_item(
|
||||||
|
task_execs,
|
||||||
|
name='task2',
|
||||||
|
state=states.ERROR
|
||||||
|
)
|
||||||
|
self._assert_single_item(
|
||||||
|
task_execs,
|
||||||
|
name='join_task',
|
||||||
|
state=states.ERROR
|
||||||
|
)
|
||||||
|
@ -321,14 +321,16 @@ class DirectWorkflowController(base.WorkflowController):
|
|||||||
@profiler.trace('direct-wf-controller-find-downstream-joins')
|
@profiler.trace('direct-wf-controller-find-downstream-joins')
|
||||||
def _find_indirectly_affected_created_joins(self, task_name, result=None,
|
def _find_indirectly_affected_created_joins(self, task_name, result=None,
|
||||||
visited_task_names=None):
|
visited_task_names=None):
|
||||||
visited_task_names = visited_task_names or set()
|
visited_task_names = (
|
||||||
|
set() if visited_task_names is None else visited_task_names
|
||||||
|
)
|
||||||
|
|
||||||
if task_name in visited_task_names:
|
if task_name in visited_task_names:
|
||||||
return
|
return
|
||||||
|
|
||||||
visited_task_names.add(task_name)
|
visited_task_names.add(task_name)
|
||||||
|
|
||||||
result = result or set()
|
result = set() if result is None else result
|
||||||
|
|
||||||
def _process_clause(clause):
|
def _process_clause(clause):
|
||||||
for t_name, condition, params in clause:
|
for t_name, condition, params in clause:
|
||||||
|
@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
fixes:
|
||||||
|
- |
|
||||||
|
Fixed the issue when "join" task remained in WAITING state forever if
|
||||||
|
the last inbound task failed and it was not a direct predecessor.
|
Loading…
x
Reference in New Issue
Block a user