| ############################################################################ | |
| # Copyright (C) SchedMD LLC. | |
| ############################################################################ | |
| import pytest | |
| import atf | |
| check_iterations = 5 | |
| windows_used = 4 | |
| step_lifetime = 10 | |
| # Setup/Teardown | |
| @pytest.fixture(scope="module", autouse=True) | |
| def setup(): | |
| atf.require_config_parameter("SwitchType", None) | |
| atf.require_slurm_running() | |
| @pytest.fixture(scope="module") | |
| def salloc_noshell(): | |
| """Submit a backgrounded salloc job""" | |
| job_id = atf.submit_job_salloc("--verbose --no-shell", fatal=True) | |
| atf.wait_for_job_state(job_id, "RUNNING", fatal=True) | |
| return job_id | |
| @pytest.fixture(scope="module") | |
| def srun_background(salloc_noshell): | |
| """Submit an initial job step to claim some switch windows""" | |
| return atf.run_command( | |
| f"nohup srun --jobid={salloc_noshell} -O -n {windows_used} sleep {step_lifetime} >/dev/null 2>&1 &", | |
| fatal=True, | |
| ) | |
| def test_window_conflicts(salloc_noshell, srun_background): | |
| """Start more job steps to check see if any switch window conflicts occur""" | |
| for iteration in range(0, check_iterations): | |
| assert ( | |
| atf.run_job_exit(f"--jobid={salloc_noshell} -O -n {windows_used} true") == 0 | |
| ) |