test_assertion_error.py 6.84 KiB
from helper import *
from test_users import run_user, assert_exec_time_equals_profile
schedconf_file = 'test-instances/user_description_file.json'
create_dir_rec_if_needed("test-instances")
import json
def make_error_file(seed=None, red_windows=None, probas=None, yellow_windows=None):
"""Create a simple user_description_file with the given parameters to check for proper handling of errors"""
if probas is None:
probas = [("red_prob_see_you_later", 1.0)]
param = {"input_json": "test/workloads/dyn/two_jobs.json"}
for proba in probas:
proba_name, proba_value = proba
param[proba_name] = proba_value
error_file = {
"users": [
{
"name": "user14",
"category": "dm_user_multi_behavior",
"param": param
}
]
}
if seed:
error_file["seed"] = seed
if not (red_windows is None):
error_file["red_windows"] = red_windows
if not (yellow_windows is None):
error_file["yellow_windows"] = yellow_windows
with open(schedconf_file, 'w+') as error_description_file:
json.dump(error_file, error_description_file)
def error_user(user_name, platform_multiC, expected_error="", test_name=None, schedconf=None):
"""Launch a simulation and check if the expected_error is risen"""
if test_name == None:
test_name = f'{user_name}-{platform_multiC.name}'
if schedconf == None:
schedconf = 'test-instances/user_description_file.json'
output_dir, robin_filename, _ = init_instance(test_name)
batcmd = gen_batsim_cmd(platform_multiC.filename, empty_workload, output_dir,
"--energy --enable-compute-sharing --enable-dynamic-jobs --acknowledge-dynamic-jobs --enable-profile-reuse")
instance = RobinInstance(output_dir=output_dir,
batcmd=batcmd,
schedcmd=f"batmen -v bin_packing --variant_options_filepath {schedconf}",
simulation_timeout=30, ready_timeout=5,
success_timeout=10, failure_timeout=0
)
instance.to_file(robin_filename)
ret = run_robin(robin_filename)
assert ret.returncode == 2
error_filename = output_dir + "/log/sched.err.log"
error_file = open(error_filename)
error_message = error_file.readlines()[-3]
assert "with message" in error_message
assert expected_error in error_message
if has_expected_output(test_name):
assert_expected_output(test_name)
return output_dir
def test_error_no_seed(platform_multiC):
make_error_file(red_windows=[[1, 2]])
out_dir = error_user("dm_user_multi_behavior_no_seed", platform_multiC, "seed")
def test_error_invalid_seed(platform_multiC):
make_error_file(-1.0)
out_dir = error_user("dm_user_multi_behavior_invalid_seed", platform_multiC, "'seed' should be a 32-bit integer")
def test_error_no_window(platform_multiC):
make_error_file(3)
out_dir = error_user("dm_user_multi_behavior_no_windows", platform_multiC, "window")
def test_empty_red_window(platform_multiC):
make_error_file(3, red_windows=[])
out_dir = error_user("dm_user_multi_behavior_empty_red_windows", platform_multiC,
"red_windows should be a non-empty array")
def test_empty_yellow_window(platform_multiC):
make_error_file(3, yellow_windows=[])
out_dir = error_user("dm_user_multi_behavior_empty_yellow_windows", platform_multiC,
"yellow_windows should be a non-empty array")
def test_invalid_red_prob(platform_multiC):
invalid_probas = [[], [("red_prob_reconfig", 0.0)], [("red_prob_degrad", 0.0)], [("red_prob_rigid", 0.0)],
[("red_prob_see_you_later", 0.0)], [("red_prob_renonce", 0.0)],
[("red_prob_reconfig", 0.0), ("red_prob_degrad", 0.0), ("red_prob_rigid", 0.0),
("red_prob_renonce", 0.0), ("red_prob_see_you_later", 0.0)]]
i = 0
for probas in invalid_probas:
make_error_file(3, red_windows=[[0, 1]], probas=probas)
out_dir = error_user("dm_user_multi_behavior_invalid_red_prob_" + str(i), platform_multiC, "red")
i += 1
def test_invalid_yellow_prob(platform_multiC):
invalid_probas = [[], [("yellow_prob_reconfig", 0.0)], [("yellow_prob_degrad", 0.0)], [("yellow_prob_rigid", 0.0)],
[("yellow_prob_reconfig", 0.0), ("yellow_prob_degrad", 0.0), ("yellow_prob_rigid", 0.0)]]
i = 0
for probas in invalid_probas:
make_error_file(3, yellow_windows=[[0, 1]], probas=probas)
out_dir = error_user("dm_user_multi_behavior_invalid_yellow_prob_" + str(i), platform_multiC, "yellow")
i += 1
def test_error_invalid_window_order(platform_multiC):
invalid_windows_list = [[[2, 3], [0, 1]], [[0, 1], [1, 2]], [[1, 2], [0, 1]], [[0, 1], [2, 3], [4, 5], [5, 6]],
[[1, 4], [2, 6]], [[1, 4], [2, 3]]]
for i in range(len(invalid_windows_list)):
invalid_window = invalid_windows_list[i]
make_error_file(3, invalid_window)
out_dir = error_user("dm_user_multi_behavior_invalid_windows_order_" + str(i), platform_multiC,
"should be sorted and contains disjoint interval")
out_dir_2 = error_user("dm_user_multi_behavior_many_windows_wrong", platform_multiC,
schedconf="test/schedconf/dm_user_multi_behavior_many_windows_wrong.json")
out_dir_3 = error_user("dm_user_multi_behavior_many_windows_wrong_order", platform_multiC,
schedconf="test/schedconf/dm_user_multi_behavior_many_windows_wrong_order.json")
def test_error_invalid_window_type(platform_multiC):
invalid_windows_list = [[[0]], [[0.0, 1.0]], [[0, 1, 2]], [[0, 1], [2, 3], [4, 5], 5], [[1, 4.0]], [[1.0, 4]]]
for i in range(len(invalid_windows_list)):
invalid_window = invalid_windows_list[i]
make_error_file(3, invalid_window)
out_dir = error_user("dm_user_multi_behavior_invalid_windows_type" + str(i), platform_multiC,
"should be an array of intervals (pairs of integers)")
def test_error_intersection_red_yellow(platform_multiC):
intersecting_windows_list = [[[[300, 301], [400, 401]], [[399, 400]]],
[[[300, 400]], [[320, 350]]],
[[[300, 400]], [[250, 350]]]]
probas = [("red_prob_rigid",1.0),("yellow_prob_rigid",1.0)]
for i in range(len(intersecting_windows_list)):
intersecting_windows = intersecting_windows_list[i]
for j in range(1, 3):
make_error_file(3, red_windows=intersecting_windows[j % 2], yellow_windows=intersecting_windows[j // 2],probas=probas)
error_user(f"dm_user_multi_behavior_intersecting_red_yellow_{i}_{j}", platform_multiC,
"at least one common element in their interval")