test_broker.py

from helper import *


# def test_broker_mono(platform_monoC, workload_static):
#     Scheduler = namedtuple('Scheduler', ['name', 'short_name'])
#     sched_mono = Scheduler("fcfs", "fcfs")

def test_broker_mono(platform_monoC, workload_static, sched_mono):
    """Test the broker mode with monocore platform files and schedulers"""

    test_name = f'broker-{sched_mono.short_name}-{platform_monoC.name}-{workload_static.name}'
    output_dir, robin_filename, _ = init_instance(test_name)

    batcmd = gen_batsim_cmd(platform_monoC.filename, workload_static.filename, output_dir, "--enable-dynamic-jobs --acknowledge-dynamic-jobs --enable-profile-reuse")
    instance = RobinInstance(output_dir=output_dir,
        batcmd=batcmd,
        schedcmd=f"batmen -v {sched_mono.name} --variant_options_filepath test/schedconf/user_description_file.json",
        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 == 0

    if has_expected_output(test_name):
        assert_expected_output(test_name)


def test_broker_multi(platform_multiC, workload_static, sched_multi):
    """Test the broker mode with multicore platform files and schedulers"""

    test_name = f'broker-{sched_multi.short_name}-{platform_multiC.name}-{workload_static.name}'
    output_dir, robin_filename, _ = init_instance(test_name)

    batcmd = gen_batsim_cmd(platform_multiC.filename, workload_static.filename, 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 {sched_multi.name} --variant_options_filepath test/schedconf/user_description_file.json",
        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 == 0

    if has_expected_output(test_name):
        assert_expected_output(test_name)