| ############################################################################ | |
| # Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. | |
| ############################################################################ | |
| import re | |
| import pytest | |
| import atf | |
| pytestmark = pytest.mark.slow | |
| def setup_module(): | |
| atf.require_nodes(4, [("CPUs", 2)]) | |
| atf.require_slurm_running() | |
| def submit_running_job(): | |
| """Submit a job and wait for it to be running. Returns job_id.""" | |
| job_id = atf.submit_job_sbatch("-n1 --wrap 'srun sleep infinity'", fatal=True) | |
| atf.wait_for_job_state(job_id, "RUNNING", fatal=True) | |
| atf.wait_for_step(job_id, 0, fatal=True) | |
| return job_id | |
| def submit_multistep_job(): | |
| """Submit a batch job that launches 2 steps in parallel which sleep | |
| indefinitely. Returns jobid. | |
| """ | |
| script = atf.module_tmp_path / "multistep.sh" | |
| atf.make_bash_script( | |
| script, | |
| """srun -n1 --mem=0 --overlap sleep infinity & | |
| srun -n1 --mem=0 --overlap sleep infinity & | |
| wait""", | |
| ) | |
| job_id = atf.submit_job_sbatch(f"-N2 -O {script}", fatal=True) | |
| atf.wait_for_job_state(job_id, "RUNNING", fatal=True) | |
| atf.wait_for_step(job_id, 0, fatal=True) | |
| atf.wait_for_step(job_id, 1, fatal=True) | |
| return job_id | |
| def get_running_user_steps(job_id): | |
| """Return set of running user step keys like {'<job_id>.0', '<job_id>.1'}.""" | |
| steps = atf.get_steps(job_id, quiet=True) | |
| return { | |
| k | |
| for k, v in steps.items() | |
| if v.get("State", "") == "RUNNING" and ".batch" not in k and ".extern" not in k | |
| } | |
| @pytest.mark.parametrize( | |
| "id_type,step", | |
| [ | |
| ("JobId", False), | |
| ("JobId", True), | |
| ("SLUID", False), | |
| ("SLUID", True), | |
| ], | |
| ) | |
| def test_verbose_signal(id_type, step): | |
| """Verify scancel --verbose shows correct 'Signal N to job/step <id>' message.""" | |
| if id_type == "SLUID": | |
| atf.require_version( | |
| (26, 5), | |
| "bin/scancel", | |
| reason="Ticket 22180: SLUID availability added in 26.05+", | |
| ) | |
| job_id = submit_running_job() | |
| identifier = str(atf.get_job_parameter(job_id, id_type)) | |
| target = f"{identifier}.0" if step else identifier | |
| entity = "step" if step else "job" | |
| result = atf.run_command(f"scancel --signal SIGHUP {target} --verbose") | |
| assert re.search( | |
| rf"Signal 1 to {entity} {re.escape(target)}", result["stderr"] | |
| ), f"Expected 'Signal 1 to {entity} {target}' in stderr: {result['stderr']}" | |
| atf.wait_for_job_state(job_id, "FAILED", fatal=True) | |
| @pytest.mark.skipif( | |
| atf.get_version("bin/scancel") < (26, 5), | |
| reason="Ticket 22180: SLUID availability added in 26.05+", | |
| ) | |
| def test_verbose_mixed_ids(): | |
| """Verify scancel --verbose output with a mix of job IDs, step IDs, SLUIDs, and SLUID steps. | |
| Equivalent to: | |
| scancel --signal SIGHUP <job1>,<job2>.0,<sluid3>,<sluid4>.0 --verbose | |
| Expected output lines (in any order): | |
| scancel: Signal 1 to job <job1> | |
| scancel: Signal 1 to step <job2>.0 | |
| scancel: Signal 1 to job <sluid3> | |
| scancel: Signal 1 to step <sluid4>.0 | |
| """ | |
| job1 = submit_running_job() | |
| job2 = submit_running_job() | |
| job3 = submit_running_job() | |
| job4 = submit_running_job() | |
| sluid3 = atf.get_job_parameter(job3, "SLUID") | |
| sluid4 = atf.get_job_parameter(job4, "SLUID") | |
| ids = f"{job1},{job2}.0,{sluid3},{sluid4}.0" | |
| result = atf.run_command(f"scancel --signal SIGHUP {ids} --verbose") | |
| stderr = result["stderr"] | |
| assert re.search( | |
| rf"Signal 1 to job {job1}", stderr | |
| ), f"Missing 'Signal 1 to job {job1}' in: {stderr}" | |
| assert re.search( | |
| rf"Signal 1 to step {job2}\.0", stderr | |
| ), f"Missing 'Signal 1 to step {job2}.0' in: {stderr}" | |
| assert re.search( | |
| rf"Signal 1 to job {re.escape(sluid3)}", stderr | |
| ), f"Missing 'Signal 1 to job {sluid3}' in: {stderr}" | |
| assert re.search( | |
| rf"Signal 1 to step {re.escape(sluid4)}\.0", stderr | |
| ), f"Missing 'Signal 1 to step {sluid4}.0' in: {stderr}" | |
| # Verify all four jobs are actually terminated | |
| for jid in [job1, job2, job3, job4]: | |
| atf.wait_for_job_state(jid, "FAILED", fatal=True) | |
| @pytest.mark.parametrize( | |
| "id_type,step,expected_state", | |
| [ | |
| ("JobId", False, "CANCELLED"), | |
| ("JobId", True, "FAILED"), | |
| ("SLUID", False, "CANCELLED"), | |
| ("SLUID", True, "FAILED"), | |
| ], | |
| ) | |
| def test_cancel_terminates(id_type, step, expected_state): | |
| """Verify scancel with job/step ID or SLUID actually terminates the job.""" | |
| if id_type == "SLUID": | |
| atf.require_version( | |
| (26, 5), | |
| "bin/scancel", | |
| reason="Ticket 22180: SLUID availability added in 26.05+", | |
| ) | |
| job_id = submit_running_job() | |
| identifier = str(atf.get_job_parameter(job_id, id_type)) | |
| target = f"{identifier}.0" if step else identifier | |
| atf.run_command(f"scancel {target}", fatal=True) | |
| atf.wait_for_job_state(job_id, expected_state, fatal=True) | |
| @pytest.mark.parametrize("id_type", ["JobId", "SLUID"]) | |
| def test_signal_delivers(id_type): | |
| """Verify scancel --signal with a job ID or SLUID delivers SIGSTOP/SIGCONT.""" | |
| if id_type == "SLUID": | |
| atf.require_version( | |
| (26, 5), | |
| "bin/scancel", | |
| reason="Ticket 22180: SLUID availability added in 26.05+", | |
| ) | |
| job_id = submit_running_job() | |
| identifier = str(atf.get_job_parameter(job_id, id_type)) | |
| atf.run_command(f"scancel --signal SIGSTOP {identifier}", fatal=True) | |
| atf.wait_for_job_state(job_id, "STOPPED", fatal=True) | |
| atf.run_command(f"scancel --signal SIGCONT {identifier}", fatal=True) | |
| atf.wait_for_job_state(job_id, "RUNNING", fatal=True) | |
| atf.run_command(f"scancel {identifier}", fatal=True) | |
| atf.wait_for_job_state(job_id, "CANCELLED", fatal=True) | |
| @pytest.mark.skipif( | |
| atf.get_version("bin/scancel") < (26, 5), | |
| reason="Ticket 22180: SLUID availability added in 26.05+", | |
| ) | |
| def test_cancel_mixed_terminates(): | |
| """Verify scancel with mixed job IDs, step IDs, SLUIDs, and SLUID steps terminates all jobs.""" | |
| job1 = submit_running_job() | |
| job2 = submit_running_job() | |
| job3 = submit_running_job() | |
| job4 = submit_running_job() | |
| sluid3 = atf.get_job_parameter(job3, "SLUID") | |
| sluid4 = atf.get_job_parameter(job4, "SLUID") | |
| ids = f"{job1},{job2}.0,{sluid3},{sluid4}.0" | |
| atf.run_command(f"scancel {ids}", fatal=True) | |
| # job cancel -> CANCELLED, step cancel -> FAILED | |
| atf.wait_for_job_state(job1, "CANCELLED", fatal=True) | |
| atf.wait_for_job_state(job2, "FAILED", fatal=True) | |
| atf.wait_for_job_state(job3, "CANCELLED", fatal=True) | |
| atf.wait_for_job_state(job4, "FAILED", fatal=True) | |
| @pytest.mark.parametrize("id_type", ["JobId", "SLUID"]) | |
| def test_cancel_one_step(id_type): | |
| """Cancel only step 0 in a multi-step job; step 1 must survive.""" | |
| if id_type == "SLUID": | |
| atf.require_version( | |
| (26, 5), | |
| "bin/scancel", | |
| reason="Ticket 22180: SLUID availability added in 26.05+", | |
| ) | |
| job_id = submit_multistep_job() | |
| identifier = str(atf.get_job_parameter(job_id, id_type)) | |
| result = atf.run_command(f"scancel {identifier}.0 --verbose") | |
| assert re.search( | |
| rf"Terminating step {re.escape(identifier)}\.0", result["stderr"] | |
| ), f"Expected 'Terminating step {identifier}.0' in stderr: {result['stderr']}" | |
| # Step 1 should still be running | |
| assert atf.repeat_until( | |
| lambda: get_running_user_steps(job_id), | |
| lambda steps: f"{job_id}.0" not in steps and f"{job_id}.1" in steps, | |
| poll_interval=1, | |
| ), "Step 1 should still be running after cancelling step 0" |