The number of entries under ``roles`` and ``targets`` must match.
-Note the colon after every task name in the ``tasks`` section.
+Note the colon after every task name in the ``tasks`` section. Also note the
+dashes before each task. This is the YAML syntax for an ordered list and
+specifies the order in which tasks are executed.
The ``install`` task needs to precede all other tasks.
def validate_tasks(config):
"""
- Ensures that config tasks do not include 'kernel'.
+ Ensures that config tasks is a list and doesn't include 'kernel'.
Returns the original tasks key if found. If not, returns an
empty list.
# return the default value for tasks
return []
+ msg = "Expected list in 'tasks'; instead got: {0}".format(config['tasks'])
+ assert isinstance(config['tasks'], list), msg
+
for task in config['tasks']:
- msg = ('kernel installation shouldn be a base-level item, not part ' +
+ msg = ('kernel installation should be a base-level item, not part ' +
'of the tasks list')
assert 'kernel' not in task, msg
assert result == {"owner": "the_owner", "success": True}
def test_validate_tasks_invalid(self):
- config = {"tasks": {"kernel": "can't be here"}}
- with pytest.raises(AssertionError):
+ config = {"tasks": [{"kernel": "can't be here"}]}
+ with pytest.raises(AssertionError) as excinfo:
run.validate_tasks(config)
+ assert excinfo.value.message.startswith("kernel installation")
def test_validate_task_no_tasks(self):
result = run.validate_tasks({})
assert result == []
def test_validate_tasks_valid(self):
- expected = {"foo": "bar"}
+ expected = [{"foo": "bar"}, {"bar": "foo"}]
result = run.validate_tasks({"tasks": expected})
assert result == expected
+ def test_validate_tasks_is_list(self):
+ with pytest.raises(AssertionError) as excinfo:
+ run.validate_tasks({"tasks": {"foo": "bar"}})
+ assert excinfo.value.message.startswith("Expected list")
+
def test_get_initial_tasks_invalid(self):
- with pytest.raises(AssertionError):
+ with pytest.raises(AssertionError) as excinfo:
run.get_initial_tasks(True, {"targets": "can't be here"}, "machine_type")
+ assert excinfo.value.message.startswith("You cannot")
def test_get_inital_tasks(self):
config = {"roles": range(2), "kernel": "the_kernel", "use_existing_cluster": False}