testsuite/python/tests/test_168_1.py

############################################################################
# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
############################################################################
"""Basic heterogeneous job sbatch test.

Submit a 3-component hetjob via sbatch and verify that the batch script
sees the expected per-component environment (SLURM_HET_SIZE, the
SLURM_JOB_ID_HET_GROUP_<N> / SLURM_NTASKS_HET_GROUP_<N> /
SLURM_CPUS_PER_TASK_HET_GROUP_<N> / SLURM_MEM_PER_CPU_HET_GROUP_<N> vars,
and a per-component partition var for each group).
"""

import re

import pytest

import atf

# (cpus-per-task, mem-per-cpu, ntasks) per het group. The values differ
# between components so that broadcasting one component's value across
# every _HET_GROUP_<N> suffix would be caught.
COMPONENTS = [(4, 10, 1), (2, 2, 2), (1, 6, 3)]

# Echoed after the env dump. The env output order is not deterministic, so
# this marker is what tells us the whole dump has been flushed.
OUTPUT_MARKER = "HETJOB_ENV_DUMPED"


@pytest.fixture(scope="module", autouse=True)
def setup():
    atf.require_nodes(3, [("CPUs", 4), ("RealMemory", 1024)])
    # Het jobs are only started by backfill; don't wait for its full cycle
    atf.require_config_parameter_includes("SchedulerParameters", ("bf_interval", 1))
    atf.require_config_parameter("SchedulerType", "sched/backfill")
    atf.require_slurm_running()


def test_hetjob_sbatch_env():
    """Verify hetjob env vars in the batch script."""

    file_in = "hetjob.in"
    file_out = "hetjob.out"
    directives = "\n#SBATCH hetjob\n".join(
        f"#SBATCH --cpus-per-task={cpus} --mem-per-cpu={mem} --ntasks={ntasks} -t1"
        for cpus, mem, ntasks in COMPONENTS
    )
    atf.make_bash_script(file_in, f"{directives}\n\nenv\necho {OUTPUT_MARKER}\n")

    leader_job_id = atf.submit_job_sbatch(f"-o {file_out} {file_in}", fatal=True)

    jobs = atf.get_jobs(leader_job_id, fatal=True)
    component_ids = sorted(
        atf.range_to_list(jobs[leader_job_id]["HetJobIdSet"]),
        key=lambda job_id: jobs[job_id]["HetJobOffset"],
    )

    atf.wait_for_job_state(leader_job_id, "DONE", fatal=True)
    atf.assert_file_contents(file_out, OUTPUT_MARKER, contains=True)

    output = atf.run_command_output(f"cat {file_out}", fatal=True)

    assert (
        re.search(r"^SLURM_HET_SIZE=3$", output, re.MULTILINE) is not None
    ), "Missing SLURM_HET_SIZE=3 in batch environment"

    for offset, (cpus, mem, ntasks) in enumerate(COMPONENTS):
        expected = {
            "SLURM_JOB_ID": component_ids[offset],
            "SLURM_CPUS_PER_TASK": cpus,
            "SLURM_MEM_PER_CPU": mem,
            "SLURM_NTASKS": ntasks,
        }
        for base, value in expected.items():
            var = f"{base}_HET_GROUP_{offset}"
            assert (
                re.search(rf"^{var}={value}$", output, re.MULTILINE) is not None
            ), f"Missing {var}={value} in batch environment"

    partition_offsets = sorted(
        int(offset)
        for offset in re.findall(
            r"^SLURM_JOB_PARTITION_HET_GROUP_(\d+)=", output, re.MULTILINE
        )
    )
    assert partition_offsets == [0, 1, 2], (
        "Expected SLURM_JOB_PARTITION_HET_GROUP_ vars for offsets [0, 1, 2],"
        f" got {partition_offsets}"
    )