Files
TheRock/build_tools/github_actions/tests/fetch_test_configurations_test.py
derek.holloway 7502bf64a4
CodeQL / Analyze (actions) (push) Canceled after 0s
CodeQL / Analyze (python) (push) Canceled after 0s
pre-commit / pre-commit (push) Canceled after 0s
Multi-Arch CI / setup (push) Canceled after 0s
Publish build_manylinux_x86_64 images / publish_build_manylinux_x86_64 (push) Canceled after 0s
Setup (Multi-Arch) / setup (push) Canceled after 0s
Multi-Arch CI / Linux::${{ fromJSON(needs.setup.outputs.linux_build_config || '{}').build_variant_label || 'skip' }} (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Publish no_rocm_image_ubi10 image / publish_no_rocm_image_ubi10 (push) Canceled after 0s
Multi-Arch CI / Windows::${{ fromJSON(needs.setup.outputs.windows_build_config || '{}').build_variant_label || 'skip' }} (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Multi-Arch CI / Manifest Diff (push) Canceled after 0s
TheRock Manifest Diff Report / Generate Manifest Diff Report (push) Canceled after 0s
Multi-Arch CI / CI Summary (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04 (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04_ocl_rt images / publish_no_rocm_image_ubuntu24_04_ocl_rt (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04_rocgdb (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04_rocgdb images / publish_no_rocm_image_ubuntu24_04_rocgdb (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04_ocl_rt (push) Canceled after 0s
Publish TheRock Docker image / build-and-push-image (push) Canceled after 0s
Unit Tests / Unit Tests :: ubuntu-24.04 (push) Canceled after 0s
Unit Tests / Unit Tests :: windows-2022 (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04_media (push) Canceled after 0s
Unit Tests / Unit Tests Summary (push) Canceled after 0s
Publish no_rocm_image_ubuntu24_04 images / publish_no_rocm_image_ubuntu24_04_openmpi (push) Canceled after 0s
Flatten subprojects into a single base monorepo
2026-09-02 16:34:47 -07:00

548 lines
20 KiB
Python

# Copyright Advanced Micro Devices, Inc.
# SPDX-License-Identifier: MIT
from pathlib import Path
import os
import sys
import json
import unittest
# Add repo root to PYTHONPATH
sys.path.insert(0, os.fspath(Path(__file__).parent.parent))
import fetch_test_configurations
class FetchTestConfigurationsTest(unittest.TestCase):
def setUp(self):
# Save environment so tests don't leak state
self._orig_env = os.environ.copy()
# Save sys.argv so tests don't leak state
self._orig_argv = sys.argv.copy()
# Save module-level attributes that tests may change
self._orig_functional_matrix = fetch_test_configurations.functional_matrix
self._orig_benchmark_matrix = fetch_test_configurations.benchmark_matrix
self._orig_get_all_families = (
fetch_test_configurations.get_all_families_for_trigger_types
)
os.environ["AMDGPU_FAMILIES"] = "gfx94X-dcgpu"
os.environ["TEST_TYPE"] = "full"
os.environ["TEST_LABELS"] = "[]"
os.environ["PROJECTS_TO_TEST"] = "*"
# Default to linux platform
sys.argv = ["fetch_test_configurations.py", "--platform=linux"]
# Capture gha_set_output instead of writing to GitHub
self.gha_output = {}
def fake_gha_set_output(payload):
self.gha_output.update(payload)
fetch_test_configurations.gha_set_output = fake_gha_set_output
def tearDown(self):
os.environ.clear()
os.environ.update(self._orig_env)
sys.argv = self._orig_argv
# Restore module-level attributes
fetch_test_configurations.functional_matrix = self._orig_functional_matrix
fetch_test_configurations.benchmark_matrix = self._orig_benchmark_matrix
fetch_test_configurations.get_all_families_for_trigger_types = (
self._orig_get_all_families
)
def _get_components(self):
self.assertIn("components", self.gha_output)
return json.loads(self.gha_output["components"])
# -----------------------
# Basic selection tests
# -----------------------
def test_linux_jobs_selected(self):
fetch_test_configurations.run()
components = self._get_components()
self.assertGreater(len(components), 0)
for job in components:
self.assertIn("linux", job["platform"])
def test_windows_jobs_selected(self):
sys.argv = ["fetch_test_configurations.py", "--platform=windows"]
fetch_test_configurations.run()
components = self._get_components()
self.assertGreater(len(components), 0)
for job in components:
self.assertIn("windows", job["platform"])
def test_single_project_filter(self):
os.environ["PROJECTS_TO_TEST"] = "hipblas"
fetch_test_configurations.run()
components = self._get_components()
self.assertEqual(len(components), 1)
self.assertEqual(components[0]["job_name"], "hipblas")
def test_test_labels_filter(self):
os.environ["TEST_LABELS"] = json.dumps(["rocblas", "hipblas"])
fetch_test_configurations.run()
components = self._get_components()
names = {job["job_name"] for job in components}
self.assertEqual(names, {"rocblas", "hipblas"})
# -----------------------
# TEST_LABELS handling
# -----------------------
def test_empty_test_labels_env_is_handled(self):
# Regression test: json.loads("") used to crash
os.environ["TEST_LABELS"] = ""
# Should not raise
fetch_test_configurations.run()
components = self._get_components()
self.assertGreater(len(components), 0)
def test_missing_test_labels_env_is_handled(self):
# Regression test: missing TEST_LABELS should behave like []
if "TEST_LABELS" in os.environ:
del os.environ["TEST_LABELS"]
# Should not raise
fetch_test_configurations.run()
components = self._get_components()
self.assertGreater(len(components), 0)
# -----------------------
# Sharding behavior
# -----------------------
def test_full_test_uses_all_shards(self):
fetch_test_configurations.run()
components = self._get_components()
hipblaslt = next(j for j in components if j["job_name"] == "hipblaslt")
self.assertEqual(hipblaslt["total_shards"], 6)
self.assertEqual(hipblaslt["shard_arr"], [1, 2, 3, 4, 5, 6])
def test_quick_test_forces_single_shard(self):
os.environ["TEST_TYPE"] = "quick"
fetch_test_configurations.run()
components = self._get_components()
for job in components:
self.assertEqual(job["total_shards"], 1)
self.assertEqual(job["shard_arr"], [1])
def test_platform_specific_shards(self):
os.environ["PROJECTS_TO_TEST"] = "hipblaslt"
fetch_test_configurations.run()
components = self._get_components()
hipblaslt_linux = components[0]
sys.argv = ["fetch_test_configurations.py", "--platform=windows"]
fetch_test_configurations.run()
components = self._get_components()
hipblaslt_windows = components[0]
self.assertNotEqual(
hipblaslt_linux["total_shards"], hipblaslt_windows["total_shards"]
)
# -----------------------
# Exclude-family logic
# -----------------------
def test_exclude_family_skips_job(self):
os.environ["AMDGPU_FAMILIES"] = "gfx1150"
fetch_test_configurations.run()
components = self._get_components()
names = {job["job_name"] for job in components}
self.assertNotIn("rocroller", names)
# -----------------------
# Functional test merging via run_extended_tests
# -----------------------
def _setup_functional_test(self):
"""Common setup for functional tests: fake matrix + isolate from other components."""
os.environ["PROJECTS_TO_TEST"] = "func1"
fetch_test_configurations.functional_matrix = {
"func1": {
"job_name": "func1",
"platform": ["linux"],
"total_shards": 1,
}
}
def test_functional_merged_when_enabled(self):
os.environ["RUN_EXTENDED_TESTS"] = "true"
self._setup_functional_test()
fetch_test_configurations.run()
components = self._get_components()
self.assertEqual(len(components), 1)
self.assertEqual(components[0]["job_name"], "func1")
def test_functional_not_merged_when_disabled(self):
os.environ["RUN_EXTENDED_TESTS"] = "false"
self._setup_functional_test()
fetch_test_configurations.run()
components = self._get_components()
names = {job["job_name"] for job in components}
self.assertNotIn("func1", names)
# -----------------------
# Benchmark merging via run_extended_tests
# -----------------------
def _setup_benchmark_test(self):
"""Common setup for benchmark tests: fake matrix + isolate from other components."""
os.environ["PROJECTS_TO_TEST"] = "bench1"
fetch_test_configurations.benchmark_matrix = {
"bench1": {
"job_name": "bench1",
"platform": ["linux"],
"total_shards_dict": {"linux": 1},
}
}
def test_benchmarks_merged_when_extended_tests_enabled(self):
os.environ["RUN_EXTENDED_TESTS"] = "true"
self._setup_benchmark_test()
fetch_test_configurations.run()
components = self._get_components()
self.assertEqual(len(components), 1)
self.assertEqual(components[0]["job_name"], "bench1")
self.assertTrue(components[0]["is_benchmark"])
self.assertEqual(components[0]["test_type"], "full")
def test_benchmarks_not_merged_when_extended_tests_disabled(self):
os.environ["RUN_EXTENDED_TESTS"] = "false"
self._setup_benchmark_test()
fetch_test_configurations.run()
components = self._get_components()
names = {job["job_name"] for job in components}
self.assertNotIn("bench1", names)
# -----------------------
# Multi-GPU logic (RCCL)
# -----------------------
def test_multi_gpu_job_included_when_supported(self):
def fake_get_all_families(_):
return {"gfx94x": {"linux": {"test-runs-on-multi-gpu": "linux-mi300-mgpu"}}}
fetch_test_configurations.get_all_families_for_trigger_types = (
fake_get_all_families
)
fetch_test_configurations.run()
components = self._get_components()
rccl = next(j for j in components if j["job_name"] == "rccl")
self.assertEqual(rccl["multi_gpu_runner"], "linux-mi300-mgpu")
def test_multi_gpu_job_uses_weighted_labels_when_available(self):
"""When test-runs-on-multi-gpu-labels is present, select_weighted_label is used."""
def fake_get_all_families(_):
return {
"gfx94x": {
"linux": {
"test-runs-on": "linux-gfx942-default",
"test-runs-on-labels": [
{"label": "linux-gfx942-a", "weight": 0.5},
{"label": "linux-gfx942-b", "weight": 0.5},
],
"test-runs-on-multi-gpu": "linux-mi300-mgpu-default",
"test-runs-on-multi-gpu-labels": [
{"label": "linux-mi300-mgpu-a", "weight": 0.5},
{"label": "linux-mi300-mgpu-b", "weight": 0.5},
],
}
}
}
fetch_test_configurations.get_all_families_for_trigger_types = (
fake_get_all_families
)
# Mock select_weighted_label to verify it's called and return known labels
original_select_weighted_label = fetch_test_configurations.select_weighted_label
selected_labels = []
def fake_select_weighted_label(labels_config, context_name):
selected_labels.append((labels_config, context_name))
# Return different labels based on whether it's multi-gpu
if "multi-gpu" in context_name:
return "linux-mi300-mgpu-a"
return "linux-gfx942-a"
fetch_test_configurations.select_weighted_label = fake_select_weighted_label
try:
fetch_test_configurations.run()
components = self._get_components()
rccl = next(j for j in components if j["job_name"] == "rccl")
self.assertEqual(rccl["multi_gpu_runner"], "linux-mi300-mgpu-a")
# Verify select_weighted_label was called for the multi-gpu jobs.
# With rocshmem added there is more than one multi-GPU job (rccl and
# rocshmem), each using a "<job_name>-multi-gpu" context.
multi_gpu_calls = [c for c in selected_labels if "multi-gpu" in c[1]]
multi_gpu_contexts = {c[1] for c in multi_gpu_calls}
self.assertIn("rccl-multi-gpu", multi_gpu_contexts)
self.assertIn("rocshmem-multi-gpu", multi_gpu_contexts)
finally:
fetch_test_configurations.select_weighted_label = (
original_select_weighted_label
)
def test_multi_gpu_job_excluded_when_not_supported(self):
os.environ["AMDGPU_FAMILIES"] = "gfx90a"
def fake_get_all_families(_):
return {}
fetch_test_configurations.get_all_families_for_trigger_types = (
fake_get_all_families
)
fetch_test_configurations.run()
components = self._get_components()
names = {job["job_name"] for job in components}
self.assertNotIn("rccl", names)
# -----------------------
# Output contract
# -----------------------
def test_windows_hip_tests_default_emits_pal_only(self):
"""On Windows, hip-tests emits only PAL by default (WINDOWS_HIP_ROCR_TESTS off)."""
sys.argv = ["fetch_test_configurations.py", "--platform=windows"]
os.environ["TEST_LABELS"] = json.dumps(["hip-tests"])
fetch_test_configurations.run()
components = self._get_components()
hip_jobs = [j for j in components if "hip-tests" in j["job_name"]]
self.assertEqual(len(hip_jobs), 1, "Expected only hip-tests (PAL)")
self.assertEqual(hip_jobs[0]["job_name"], "hip-tests (PAL)")
self.assertNotIn("expect_failure", hip_jobs[0])
self.assertEqual(hip_jobs[0]["total_shards"], 4)
self.assertEqual(hip_jobs[0]["shard_arr"], [1, 2, 3, 4])
def test_windows_hip_tests_emits_pal_and_rocr_entries(self):
"""On Windows with WINDOWS_HIP_ROCR_TESTS=true, hip-tests runs PAL and ROCR."""
sys.argv = ["fetch_test_configurations.py", "--platform=windows"]
os.environ["TEST_LABELS"] = json.dumps(["hip-tests"])
os.environ["WINDOWS_HIP_ROCR_TESTS"] = "true"
fetch_test_configurations.run()
components = self._get_components()
hip_jobs = [j for j in components if "hip-tests" in j["job_name"]]
self.assertEqual(
len(hip_jobs), 2, "Expected hip-tests (PAL) and hip-tests (ROCR)"
)
names = {j["job_name"] for j in hip_jobs}
self.assertEqual(names, {"hip-tests (PAL)", "hip-tests (ROCR)"})
pal = next(j for j in hip_jobs if j["job_name"] == "hip-tests (PAL)")
self.assertNotIn("expect_failure", pal)
self.assertEqual(pal["total_shards"], 4)
self.assertEqual(pal["shard_arr"], [1, 2, 3, 4])
rocr = next(j for j in hip_jobs if j["job_name"] == "hip-tests (ROCR)")
self.assertTrue(rocr["expect_failure"])
self.assertEqual(rocr["total_shards"], 4)
self.assertEqual(rocr["shard_arr"], [1, 2, 3, 4])
def test_windows_hip_tests_quick_uses_single_shard(self):
"""On Windows with test_type=quick and ROCR enabled, PAL/ROCR each use 1 shard."""
sys.argv = ["fetch_test_configurations.py", "--platform=windows"]
os.environ["TEST_LABELS"] = json.dumps(["hip-tests"])
os.environ["TEST_TYPE"] = "quick"
os.environ["WINDOWS_HIP_ROCR_TESTS"] = "true"
fetch_test_configurations.run()
components = self._get_components()
hip_jobs = [j for j in components if "hip-tests" in j["job_name"]]
for job in hip_jobs:
self.assertEqual(job["total_shards"], 1)
self.assertEqual(job["shard_arr"], [1])
def test_platform_is_emitted(self):
fetch_test_configurations.run()
self.assertEqual(self.gha_output["platform"], "linux")
def test_container_options_on_windows_is_string_not_list(self):
# Regression: a list value here caused
# `options: ${{ fromJSON(...).container_options }}` in test_component.yml
# to evaluate to a YAML sequence, failing template parsing.
job = {
"container_options": [
"--cap-add SYS_MODULE",
"-v /lib/modules:/lib/modules",
]
}
out = fetch_test_configurations._build_container_options(job, "windows")
self.assertIsInstance(out["container_options"], str)
def test_container_options_on_linux_is_joined_string(self):
job = {"container_options": ["--cap-add=SYS_PTRACE"]}
out = fetch_test_configurations._build_container_options(job, "linux")
self.assertIsInstance(out["container_options"], str)
self.assertIn("--cap-add=SYS_PTRACE", out["container_options"])
# -----------------------
# ASAN sandbox runner selection
# -----------------------
def test_asan_build_uses_sandbox_runner(self):
"""ASAN builds should use test-runs-on-sandbox when available."""
os.environ["BUILD_VARIANT"] = "asan"
os.environ["PROJECTS_TO_TEST"] = "rocblas"
def fake_get_all_families(_):
return {
"gfx94x": {
"linux": {
"test-runs-on": "linux-gfx942-prod",
"test-runs-on-labels": [
{"label": "linux-gfx942-a", "weight": 0.5},
{"label": "linux-gfx942-b", "weight": 0.5},
],
"test-runs-on-sandbox": "linux-mi325-gpu-rocm-cpu-sandbox",
}
}
}
fetch_test_configurations.get_all_families_for_trigger_types = (
fake_get_all_families
)
fetch_test_configurations.run()
components = self._get_components()
rocblas = next(j for j in components if j["job_name"] == "rocblas")
self.assertEqual(rocblas["test_runner"], "linux-mi325-gpu-rocm-cpu-sandbox")
def test_host_asan_build_uses_sandbox_runner(self):
"""host-asan builds should also use test-runs-on-sandbox."""
os.environ["BUILD_VARIANT"] = "host-asan"
os.environ["PROJECTS_TO_TEST"] = "hipblas"
def fake_get_all_families(_):
return {
"gfx94x": {
"linux": {
"test-runs-on": "linux-gfx942-prod",
"test-runs-on-sandbox": "linux-sandbox-runner",
}
}
}
fetch_test_configurations.get_all_families_for_trigger_types = (
fake_get_all_families
)
fetch_test_configurations.run()
components = self._get_components()
hipblas = next(j for j in components if j["job_name"] == "hipblas")
self.assertEqual(hipblas["test_runner"], "linux-sandbox-runner")
def test_release_build_uses_weighted_runner(self):
"""Release builds should use weighted runner labels, not sandbox."""
os.environ["BUILD_VARIANT"] = "release"
os.environ["PROJECTS_TO_TEST"] = "rocblas"
def fake_get_all_families(_):
return {
"gfx94x": {
"linux": {
"test-runs-on": "linux-gfx942-default",
"test-runs-on-labels": [
{"label": "linux-gfx942-weighted", "weight": 1.0},
],
"test-runs-on-sandbox": "linux-sandbox-runner",
}
}
}
fetch_test_configurations.get_all_families_for_trigger_types = (
fake_get_all_families
)
# Mock select_weighted_label to return a known value
original_select_weighted_label = fetch_test_configurations.select_weighted_label
def fake_select_weighted_label(labels_config, context_name):
return "linux-gfx942-weighted"
fetch_test_configurations.select_weighted_label = fake_select_weighted_label
try:
fetch_test_configurations.run()
components = self._get_components()
rocblas = next(j for j in components if j["job_name"] == "rocblas")
self.assertEqual(rocblas["test_runner"], "linux-gfx942-weighted")
finally:
fetch_test_configurations.select_weighted_label = (
original_select_weighted_label
)
def test_asan_build_without_sandbox_uses_default_runner(self):
"""ASAN builds without sandbox config should fall back to default runner."""
os.environ["BUILD_VARIANT"] = "asan"
os.environ["PROJECTS_TO_TEST"] = "hipblas"
def fake_get_all_families(_):
return {
"gfx94x": {
"linux": {
"test-runs-on": "linux-gfx942-default",
# No test-runs-on-sandbox defined
}
}
}
fetch_test_configurations.get_all_families_for_trigger_types = (
fake_get_all_families
)
fetch_test_configurations.run()
components = self._get_components()
hipblas = next(j for j in components if j["job_name"] == "hipblas")
self.assertEqual(hipblas["test_runner"], "linux-gfx942-default")
if __name__ == "__main__":
unittest.main()