Skip to content
Snippets Groups Projects
Commit 24cc90ea authored by Maël Madon's avatar Maël Madon
Browse files

test: new test against expected outputs

parent 92ae8423
No related branches found
No related tags found
No related merge requests found
import os
import os, subprocess
class KTH_WL:
swf_file = "workloads/KTH-SP2-1996-2.1-cln.swf"
......@@ -11,4 +11,29 @@ thresholds = [0, 30, 1440]
def create_dir_rec_if_needed(dirname):
if not os.path.exists(dirname):
os.makedirs(dirname)
\ No newline at end of file
os.makedirs(dirname)
def run_script(delim,
threshold,
input_swf=KTH_WL.swf_file,
dyn_red=True,
graph=False):
if input_swf == KTH_WL.swf_file:
pfx = "KTH"
else:
pfx = input_swf.split("/")[-1].split(".")[0]
test_name = f"{pfx}_{delim}_t{threshold}"
out_dir = os.path.abspath(f'test-out/{test_name}')
create_dir_rec_if_needed(out_dir)
if not graph:
cp = subprocess.run(['python3', 'swf2sessions.py',
'-at', str(threshold), '-q', input_swf, out_dir],
check=True)
else:
cp = subprocess.run(['python3', 'swf2sessions.py',
'-at', str(threshold), '--graph', '-q', input_swf, out_dir],
check=True)
return out_dir, cp
\ No newline at end of file
{
"description": "Prototype for session-annotated Batsim JSON (SABjson) format. See diagram https://app.diagrams.net/#G1tbo7oHahsgxTmhICucCGam5XNtshOUOb",
"command": "command used to generate the file. We can suppose that it's generated with arrival delimitation approach and threshold = 60mn = 3600s.",
"date": "data of generation of the file",
"nb_res": 4,
"version": 2,
"sessions": [
{
"id": 1,
"first_submit_time": 0.0,
"preceding_sessions": [],
"thinking_time_after_preceding_session": [],
"nb_jobs": 2,
"jobs": [
{
"id": 1,
"profile": "2400",
"res": 4,
"subtime": 0.0,
"walltime": 2592000.0
},
{
"id": 2,
"profile": "9600",
"res": 4,
"subtime": 3000.0,
"walltime": 2592000.0
}
]
},
{
"id": 2,
"first_submit_time": 7800.0,
"preceding_sessions": [],
"thinking_time_after_preceding_session": [],
"nb_jobs": 1,
"jobs": [
{
"id": 3,
"profile": "10800",
"res": 4,
"subtime": 0.0,
"walltime": 2592000.0
}
]
},
{
"id": 3,
"first_submit_time": 16200.0,
"preceding_sessions": [
1
],
"thinking_time_after_preceding_session": [
3600.0
],
"nb_jobs": 1,
"jobs": [
{
"id": 4,
"profile": "3600",
"res": 4,
"subtime": 0.0,
"walltime": 2592000.0
}
]
},
{
"id": 4,
"first_submit_time": 24000.0,
"preceding_sessions": [
2,
3
],
"thinking_time_after_preceding_session": [
5400.0,
4200.0
],
"nb_jobs": 2,
"jobs": [
{
"id": 5,
"profile": "5400",
"res": 4,
"subtime": 0.0,
"walltime": 2592000.0
},
{
"id": 6,
"profile": "4200",
"res": 4,
"subtime": 600.0,
"walltime": 2592000.0
}
]
}
]
}
\ No newline at end of file
"""Checks the validity of the script on instances that have been manually checked"""
from conftest import run_script
import json, os
def load_expected_output(file_name):
with open(os.path.abspath(f"test/expected_output/{file_name}.SABjson"), "r") as fd:
return json.load(fd)
def load_output(file_name):
with open(os.path.abspath(f"test-out/{file_name}/1.SABjson"), "r") as fd:
return json.load(fd)
def test_example_workload():
test_name = "example_arrival_t60"
_, _ = run_script(delim="arrival", threshold=60, input_swf="workloads/example.swf")
obtained = load_output(test_name)
expected = load_expected_output(test_name)
assert obtained["nb_res"] == expected["nb_res"]
assert len(obtained["sessions"]) == len(expected["sessions"])
for s_e, s_o in zip(obtained["sessions"], expected["sessions"]):
assert int(s_e["id"]) == int(s_o["id"])
assert float(s_e["first_submit_time"]) == float(s_o["first_submit_time"])
assert s_e["preceding_sessions"] == s_o["preceding_sessions"]
assert s_e["thinking_time_after_preceding_session"] == s_o["thinking_time_after_preceding_session"]
assert int(s_e["nb_jobs"]) == int(s_o["nb_jobs"])
for j_o, j_e in zip(s_o["jobs"], s_e["jobs"]):
assert int( j_e["id"]) == int(j_o["id"])
assert float(j_e["profile"]) == float(j_o["profile"])
assert int( j_e["res"]) == int(j_o["res"])
assert float(j_e["subtime"]) == float(j_o["subtime"])
assert float(j_e["walltime"]) == float(j_o["walltime"])
import sys
from conftest import *
from conftest import run_script, KTH_WL, thresholds
import os, subprocess
import pytest
import json
def run_test(delim,
threshold,
input_swf=KTH_WL.swf_file,
dyn_red=True,
graph=False):
test_name = f"KTH_{delim}_t{threshold}"
out_dir = os.path.abspath(f'test-out/{test_name}')
create_dir_rec_if_needed(out_dir)
if not graph:
cp = subprocess.run(['python3', 'swf2sessions.py',
'-at', str(threshold), '-q', input_swf, out_dir],
check=True)
else:
cp = subprocess.run(['python3', 'swf2sessions.py',
'-at', str(threshold), '--graph', '-q', input_swf, out_dir],
check=True)
return out_dir, cp
def test_bad_input():
with pytest.raises(subprocess.CalledProcessError):
run_test('arrival', 30, input_swf="grumpf")
run_script('arrival', 30, input_swf="grumpf")
def test_bad_output():
......@@ -47,7 +26,7 @@ def test_kth_delim_last():
"""Launch swf2sessions with 'last' delimitation approach and several threshold values"""
for threshold in thresholds:
out_dir, _ = run_test('last', threshold)
out_dir, _ = run_script('last', threshold)
assert os.path.exists(out_dir)
assert len(os.listdir(out_dir)) > 0
......@@ -57,7 +36,7 @@ def test_kth_delim_arrival():
"""Launch swf2sessions with 'arrival' delimitation approach and several threshold values"""
for threshold in thresholds:
out_dir, _ = run_test('arrival', threshold)
out_dir, _ = run_script('arrival', threshold)
assert os.path.exists(out_dir)
assert len(os.listdir(out_dir)) > 0
......@@ -67,7 +46,7 @@ def test_kth_delim_max():
"""Launch swf2sessions with 'max' delimitation approach and several threshold values"""
for threshold in thresholds:
out_dir, _ = run_test('max', threshold)
out_dir, _ = run_script('max', threshold)
assert os.path.exists(out_dir)
assert len(os.listdir(out_dir)) > 0
......@@ -77,9 +56,9 @@ def test_graph():
"""Launch a couple of instances with the --graph option and checks if the graphs are output"""
instances = ["", "", ""]
instances[0], _ = run_test('arrival', 0, graph=True)
instances[1], _ = run_test('last', 30, graph=True)
instances[2], _ = run_test('max', 1440, graph=True)
instances[0], _ = run_script('arrival', 0, graph=True)
instances[1], _ = run_script('last', 30, graph=True)
instances[2], _ = run_script('max', 1440, graph=True)
for out_dir in instances:
assert os.path.exists(out_dir)
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment