Skip to content
Snippets Groups Projects
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")