Properly collect tasks with or conditional in step file.

Tasks with or conditional (step == 1 or step == 0) were not showing up
in the step file.

To make sure we collect the tasks properly we refactor the filter
logic:
 1. make all conditional a big string with no spaces and no Boolean
 2. use findall to collect all the conditional step
 3. if step found and it does belong to the current step add it.
 4. if step found and it doesn't belong to the current step remove it.
 5. if step condition not found, let strict decide.
 6. if strict remove it else add it (default behavior)

We also add a unit test related to that filtering.

Backport unit tests were adjusted to accommodate the new interface to
_write_tasks_per_step[1]

[1] See I2781cae14663094d531ad70c3d412b3153b46612

Change-Id: I815443fa7cfd54c9884aff751caf701fd77a8f3a
Closes-Bug: #1887135
(cherry picked from commit 5493304944)
This commit is contained in:
Sofer Athlan-Guyot 2020-07-10 15:35:15 +02:00
parent 4396f91bf4
commit 4c4ba31248
2 changed files with 120 additions and 15 deletions

View File

@ -1034,3 +1034,107 @@ class OvercloudConfigTest(base.TestCase):
'config-download-latest'))
mock_os_unlink.assert_called_once_with(
'/var/lib/mistral/config-download-latest')
@patch.object(ooo_config.Config, '_open_file')
def test_overcloud_config__write_tasks_per_step(self, mock_open_file):
heat = mock.MagicMock()
self.config = ooo_config.Config(heat)
# TODO: how can I share this tasks definition between to test
# as a fixture So that I do two several test cases instead of
# a big one.
tasks = [
{
"when": "step|int == 0",
"name": "Simple check"
},
{
"when": "(step|int == 0)",
"name": "Check within parenthesis"
},
{
"when": ["step|int == 0", "test1", False],
"name": "Check with list with boolean"
},
{
"when": ["test1", False, "step|int == 0"],
"name": "Check with list with boolean other order"
},
{
"when": "step|int == 0 or step|int == 3",
"name": "Check with boolean expression"
},
{
"when": "(step|int == 0 or step|int == 3) and other_cond",
"name": "Complex boolean expression"
},
{
"name": "Task with no conditional"
}
]
# Everything should come back
tasks_per_step = self.config._write_tasks_per_step(
tasks,
'Compute/update_tasks_step0.yaml',
0
)
self.assertEqual(tasks, tasks_per_step)
# Using stict the tasks with no conditional will be dropped
tasks_per_step = self.config._write_tasks_per_step(
tasks,
'Compute/update_tasks_step0.yaml',
0,
strict=True,
)
expected_tasks = [task for task in tasks
if task != {"name": "Task with no conditional"}]
self.assertEqual(expected_tasks,
tasks_per_step)
# Some tasks will be filtered out for step 3.
tasks_per_step = self.config._write_tasks_per_step(
tasks,
'Compute/update_tasks_step3.yaml',
3
)
self.assertEqual(
[
{
"when": "step|int == 0 or step|int == 3",
"name": "Check with boolean expression"
},
{
"when": "(step|int == 0 or step|int == 3) and other_cond",
"name": "Complex boolean expression"
},
{
"name": "Task with no conditional"
}
],
tasks_per_step)
# Even more tasks will be filtered out for step 3 with strict.
tasks_per_step = self.config._write_tasks_per_step(
tasks,
'Compute/update_tasks_step3.yaml',
3,
strict=True,
)
self.assertEqual(
[
{
"when": "step|int == 0 or step|int == 3",
"name": "Check with boolean expression"
},
{
"when": "(step|int == 0 or step|int == 3) and other_cond",
"name": "Complex boolean expression"
},
],
tasks_per_step)

View File

@ -135,21 +135,22 @@ class Config(object):
return False
if not isinstance(whenexpr, list):
whenexpr = [whenexpr]
for w in whenexpr:
# "when" accepts booleans and regex only work for strings
if not isinstance(w, six.string_types):
continue
# remove all spaces to make the regex match easier
w = re.sub(r'\s+', '', w)
# make \|int optional incase forgotten; use only step digit:
# ()'s around step|int are also optional
match = re.search(r'\(?step(\|int)?\)?==([0-9]+)$', "%s" % w)
if match:
if match.group(2) == str(step):
return True
else:
return False
# No match
# Filter out boolean value and remove blanks
flatten_when = "".join([re.sub(r'\s+', '', x)
for x in whenexpr
if isinstance(x, six.string_types)])
# make \|int optional incase forgotten; use only step digit:
# ()'s around step|int are also optional
steps_found = re.findall(r'\(?step(?:\|int)?\)?==(\d+)',
flatten_when)
if steps_found:
if str(step) in steps_found:
return True
else:
return False
# No step found
if strict:
return False
return True