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

test: added tests based on KTH workload

parent 9a38c687
No related branches found
No related tags found
No related merge requests found
import os
from main import main
import os, sys
import subprocess
import pytest
KTH_WL = "workloads/KTH-SP2-1996-2.1-cln.swf"
sys.path.append('../user_session_decompo')
from swf2sessions import swf2sessions
class KTH_WL:
swf_file = "workloads/KTH-SP2-1996-2.1-cln.swf"
nb_users = 214
nb_jobs = 28476
thresholds = [0, 30, 1440]
def create_dir_rec_if_needed(dirname):
if not os.path.exists(dirname):
os.makedirs(dirname)
os.makedirs(dirname)
def run_test(delim, threshold, input_swf=KTH_WL, dyn_red=True, graph=False):
def run_test(delim, threshold, input_swf=KTH_WL.swf_file, dyn_red=True, graph=False):
test_name = f"KTH_{delim}_t{threshold}"
out_folder = "test-out/" + test_name
create_dir_rec_if_needed(out_folder)
out_dir = os.path.abspath(f'test-out/{test_name}')
create_dir_rec_if_needed(out_dir)
cp = subprocess.run(['python3', 'swf2sessions.py', '-at', '30', input_swf, out_dir], check=True)
swf2sessions(input_swf, out_dir, delim, threshold, dyn_red, graph, False)
return out_dir, cp
def test_bad_input():
with pytest.raises(subprocess.CalledProcessError):
run_test('arrival', 30, input_swf="grumpf")
def test_bad_output():
bad_output_name = "/grumpf"
with pytest.raises(subprocess.CalledProcessError):
subprocess.run(['python3', 'swf2sessions.py', '-at', '30', KTH_WL.swf_file, bad_output_name], check=True)
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)
assert os.path.exists(out_dir)
assert len(os.listdir(out_dir)) > 0
kth_sanity_check(out_dir)
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)
assert os.path.exists(out_dir)
assert len(os.listdir(out_dir)) > 0
kth_sanity_check(out_dir)
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)
assert os.path.exists(out_dir)
assert len(os.listdir(out_dir)) > 0
kth_sanity_check(out_dir)
main(input_swf, out_folder, delim, threshold, dyn_red, graph, True)
def nb_users(out_dir):
"""Return the number of users for which a SABjson file has been created"""
assert os.path.exists(out_dir)
SABjsons = [file for file in os.listdir(out_dir) if file[-8:]==".SABjson"]
return len(SABjsons)
def test_delim_approaches():
"""Launch TODO(name of the parser) with different delimitation approaches"""
delims = ['last', 'max', 'arrival']
thresholds = [0, 30, 1440]
def kth_sanity_check(out_dir):
"""Perform some sanity checks on the output files"""
assert nb_users(out_dir) == KTH_WL.nb_users
for delim in delims:
for threshold in thresholds:
run_test(delim, threshold)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment