diff --git a/docs/compatibility.md b/docs/compatibility.md new file mode 100644 index 00000000..0ff80991 --- /dev/null +++ b/docs/compatibility.md @@ -0,0 +1,90 @@ +# CUDA Compatibility Tests + +The compatibility adapter compiles and runs CUDA Samples on NVIDIA and +CUDA-compatible accelerator toolchains. It reports compilation, execution, +failure, and waived-sample counts without treating a partial pass rate as an +adapter execution error. + +## Test Input + +```json +{ + "run_id": "cuda_samples.nvidia.quick", + "testcase": "compatibility.CudaSamples.PassRate", + "config": { + "platform": "nvidia", + "cuda_samples_dir": "/workspace/cuda-samples", + "build_system": "make", + "sample_filter": ["vectorAdd", "matrixMul", "clock"], + "timeout_per_sample": 180, + "jobs": 4 + } +} +``` + +`cuda_samples_dir` must contain a `Samples` directory. The adapter recursively +discovers `Samples//` directories containing a Makefile or a +standalone CMake project. CMake grouping manifests that only aggregate child +directories are excluded. A requested sample name that is not found, or an +empty `sample_filter`, is a configuration error instead of a successful +zero-sample result. + +The InfiniPerf cuda-samples submodule pins the CMake-based `master` revision. +Its `batch_test` branch provides the Makefiles used by the original InfiniPerf +compatibility workflow. The adapter supports both layouts; use `build_system` +to select `cmake`, `make`, or the default `auto` detection. + +CMake samples are configured through a temporary wrapper project. The wrapper +sets `CUDA_ARCHITECTURES` on every generated target after the sample manifest +has been evaluated, so manifests that set their own default architecture list +cannot override the requested `sms` value. + +## Platform Toolchains + +Platform aliases and compiler candidates are shared with the hardware adapter +through `infinimetrics.hardware.constants`. Compatibility-only architecture +values and Make arguments remain in `infinimetrics.compatibility.constants`. +This keeps hardware detection and compiler discovery consistent without +coupling CUDA Samples to the hardware benchmark build script. + +| Platform | Default compiler | Default architecture | +| --- | --- | --- | +| NVIDIA | `nvcc` | `80` | +| MetaX | `cucc` (falls back to `mxcc`) | `70` | +| Iluvatar CoreX | `/usr/local/corex/bin/clang++` | `ivcore20` | + +The supported canonical platform names are `cuda`, `metax`, and `corex`. +The existing aliases `nvidia` and `iluvatar` are also accepted. + +Set `compiler`, `sms`, or `make_args` in the input when the installed vendor +SDK uses a wrapper or different target. Arguments are passed directly as an +argument list; shell expansion is not performed. The resolved default compiler +and architecture are added to the result config when they were not explicit in +the input. For MetaX, the adapter also infers `MACA_PATH` from the resolved +`cucc` or `mxcc` location when the variable is unset. An explicit `MACA_PATH` +is preserved. + +For non-NVIDIA Makefile builds, the default arguments remove NVIDIA-only +`--threads`, `-gencode`, and `-m64` flags. Platform support is declared only +after its compile and runtime workflow has been validated on target hardware. + +## Metrics + +- `compile_passed` and `compile_failed` cover all discovered samples. +- `run_passed` and `run_failed` cover samples that produced an executable. +- `run_skipped` counts CUDA Samples that explicitly return a waived result. +- A sample that does not run because compilation failed has `run_result: + "not_run"` and is not included in `run_skipped`. +- `run_pass_rate` keeps the original end-to-end definition: run passes divided + by all selected samples. +- `details` records each sample path and the final compiler or runtime error. + +`result_code: 0` means the compatibility test completed and produced valid +measurements. It does not mean every sample passed; use the pass-rate metrics +for that decision. + +Generate a readable report and optional CSV with: + +```bash +python scripts/generate_compat_report.py result.json --csv result.csv +``` diff --git a/infinimetrics/common/constants.py b/infinimetrics/common/constants.py index d3c5a4e4..218d6f7b 100644 --- a/infinimetrics/common/constants.py +++ b/infinimetrics/common/constants.py @@ -31,6 +31,7 @@ class TestCategory(str, Enum): INFER = "infer" COMM = "comm" TRAIN = "train" + COMPATIBILITY = "compatibility" # Valid test categories (derived from TestCategory enum) diff --git a/infinimetrics/compatibility/__init__.py b/infinimetrics/compatibility/__init__.py new file mode 100644 index 00000000..5e03ceba --- /dev/null +++ b/infinimetrics/compatibility/__init__.py @@ -0,0 +1,2 @@ +#!/usr/bin/env python3 +"""Compatibility testing module for CUDA Samples compilation and execution.""" diff --git a/infinimetrics/compatibility/compatibility_adapter.py b/infinimetrics/compatibility/compatibility_adapter.py new file mode 100644 index 00000000..ec0e6b47 --- /dev/null +++ b/infinimetrics/compatibility/compatibility_adapter.py @@ -0,0 +1,511 @@ +#!/usr/bin/env python3 +"""CUDA Samples compatibility test adapter. + +Compiles and runs CUDA Samples on supported CUDA-compatible platforms and +reports compile/run pass rates. + +Testcase format: + compatibility.CudaSamples.PassRate +""" + +import logging +import os +import re +import shutil +import subprocess +import tempfile +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from infinimetrics.adapter import BaseAdapter +from infinimetrics.common.constants import InfiniMetricsJson, ErrorCode +from infinimetrics.compatibility.constants import CUDA_SAMPLE_CONFIGS +from infinimetrics.hardware.constants import PLATFORM_ALIASES, PLATFORM_CONFIGS +from infinimetrics.utils.time_utils import get_timestamp + +logger = logging.getLogger(__name__) +_METRIC_PREFIX = "compatibility.cuda_samples" + +# Repo root for finding cuda-samples +_REPO_ROOT = Path(__file__).resolve().parents[3] +_CUDA_SAMPLES_DIR = ( + _REPO_ROOT / "InfiniPerf" / "benchmarks" / "compatibility" / "cuda-samples" +) + + +class CompatibilityAdapter(BaseAdapter): + """Adapter for CUDA Samples compatibility tests.""" + + def process(self, test_input: Any) -> Dict[str, Any]: + """Process compatibility test.""" + test_dict = self._normalize_test_input(test_input) + if not test_dict: + return self._create_error_response( + "Invalid test input format", result_code=ErrorCode.CONFIG + ) + + testcase = test_dict.get(InfiniMetricsJson.TESTCASE, "unknown") + config = test_dict.get(InfiniMetricsJson.CONFIG, {}) + run_id = test_dict.get(InfiniMetricsJson.RUN_ID, "unknown") + + logger.info(f"CompatibilityAdapter: Processing {testcase}") + + # Extract sub-test type from testcase (third component) + parts = testcase.split(".") + if len(parts) < 3: + return self._create_error_response( + f"Invalid testcase format: {testcase}. " + f"Expected: compatibility..", + test_dict, + result_code=ErrorCode.CONFIG, + ) + + if parts[1].lower() != "cudasamples": + return self._create_error_response( + f"Unknown compatibility sub-test: {parts[1].lower()}", + test_dict, + result_code=ErrorCode.CONFIG, + ) + + return { + InfiniMetricsJson.RESULT_CODE: 0, + InfiniMetricsJson.TIME: get_timestamp(), + InfiniMetricsJson.RUN_ID: run_id, + InfiniMetricsJson.TESTCASE: testcase, + InfiniMetricsJson.CONFIG: config, + InfiniMetricsJson.METRICS: self._run_cuda_samples_test(config), + } + + # ------------------------------------------------------------------ + # cuda-samples + # ------------------------------------------------------------------ + + def _run_cuda_samples_test(self, config: Dict[str, Any]) -> List[Dict]: + """Compile and run cuda-samples, collect pass rate.""" + requested_platform = str(config.get("platform", "cuda")).lower().strip() + platform = PLATFORM_ALIASES.get(requested_platform) + if not platform: + raise ValueError( + f"Unsupported CUDA-compatible platform: {requested_platform}" + ) + + samples_dir = config.get("cuda_samples_dir", str(_CUDA_SAMPLES_DIR)) + platform_config = CUDA_SAMPLE_CONFIGS.get(platform) + if platform_config is None: + raise KeyError( + f"CUDA Samples configuration not found for platform: {platform}" + ) + sms = self._validate_architectures(config.get("sms", platform_config["sms"])) + timeout_per_sample = self._positive_int( + config.get("timeout_per_sample", 60), "timeout_per_sample" + ) + jobs = self._positive_int(config.get("jobs", os.cpu_count() or 1), "jobs") + sample_filter = config.get("sample_filter") + build_system = str(config.get("build_system", "auto")).lower() + make_args = config.get("make_args") + if make_args is None: + make_args = list(platform_config["make_args"]) + elif not isinstance(make_args, list) or not all( + isinstance(argument, str) for argument in make_args + ): + raise ValueError("make_args must be a list of strings") + + samples_root = Path(samples_dir) / "Samples" + if not samples_root.exists(): + raise FileNotFoundError(f"CUDA samples directory not found: {samples_root}") + + sample_dirs = self._discover_sample_dirs(samples_root, sample_filter) + env = self._build_compile_env(platform, sms, config.get("compiler")) + config.setdefault("compiler", env["CUDACXX"]) + config.setdefault("sms", sms) + + details = [ + self._test_sample( + sample_dir, + samples_root, + env, + timeout_per_sample, + jobs, + build_system, + make_args, + ) + for sample_dir in sample_dirs + ] + return self._build_metrics(details) + + def _test_sample( + self, + sample_dir: Path, + samples_root: Path, + env: Dict[str, str], + timeout: int, + jobs: int, + build_system: str, + make_args: List[str], + ) -> Dict[str, Any]: + selected_build_system = self._select_build_system(sample_dir, build_system) + compile_result, run_result, error = "fail", "not_run", "" + + with tempfile.TemporaryDirectory( + prefix=f"infinibench-{sample_dir.name}-" + ) as temp_dir: + try: + binary = self._compile_sample( + sample_dir, + Path(temp_dir), + env, + timeout, + jobs, + selected_build_system, + make_args, + ) + compile_result = "pass" + run_result, error = self._run_sample(binary, timeout, env) + except subprocess.TimeoutExpired: + error = f"Compilation timed out after {timeout} seconds" + except Exception as exc: + if compile_result == "pass": + run_result = "fail" + error = str(exc) + finally: + if selected_build_system == "make": + self._clean_make_sample(sample_dir, env, timeout) + + return { + "name": sample_dir.name, + "path": sample_dir.relative_to(samples_root).as_posix(), + "compile_result": compile_result, + "run_result": run_result, + "error": error[-2000:] if error else "", + } + + @staticmethod + def _build_metrics(details: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + total = len(details) + if total == 0: + raise ValueError("No CUDA samples selected") + compile_passed = sum(d["compile_result"] == "pass" for d in details) + run_passed = sum(d["run_result"] == "pass" for d in details) + values = { + "total": total, + "compile_passed": compile_passed, + "compile_failed": total - compile_passed, + "compile_pass_rate": round(compile_passed / total * 100, 2), + "run_passed": run_passed, + "run_failed": sum(d["run_result"] == "fail" for d in details), + "run_skipped": sum(d["run_result"] == "skip" for d in details), + "run_pass_rate": round(run_passed / total * 100, 2), + } + metrics = [ + { + "name": f"{_METRIC_PREFIX}.{name}", + "value": value, + "type": "scalar", + "unit": "%" if name.endswith("pass_rate") else "", + } + for name, value in values.items() + ] + metrics.append( + { + "name": f"{_METRIC_PREFIX}.details", + "value": details, + "type": "detail", + "unit": "", + } + ) + return metrics + + @staticmethod + def _positive_int(value: Any, name: str) -> int: + try: + parsed = int(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{name} must be a positive integer") from exc + if parsed <= 0: + raise ValueError(f"{name} must be a positive integer") + return parsed + + @staticmethod + def _validate_architectures(value: Any) -> str: + architectures = str(value).strip() + if not architectures or not re.fullmatch(r"[A-Za-z0-9_.+; -]+", architectures): + raise ValueError("sms contains invalid architecture characters") + return architectures + + @staticmethod + def _discover_sample_dirs( + samples_root: Path, sample_filter: Optional[List[str]] + ) -> List[Path]: + sample_dirs = sorted( + { + manifest.parent + for manifest_name in ("CMakeLists.txt", "Makefile") + for manifest in samples_root.rglob(manifest_name) + if len(manifest.parent.relative_to(samples_root).parts) >= 2 + and CompatibilityAdapter._is_standalone_manifest(manifest) + } + ) + if not sample_dirs: + raise ValueError(f"No CUDA samples found under: {samples_root}") + if sample_filter is None: + return sample_dirs + if ( + not isinstance(sample_filter, list) + or not sample_filter + or not all(isinstance(name, str) and name for name in sample_filter) + ): + raise ValueError("sample_filter must be a non-empty list of names") + + requested = set(sample_filter) + selected = [sample for sample in sample_dirs if sample.name in requested] + found = {sample.name for sample in selected} + missing = sorted(requested - found) + if missing: + raise ValueError(f"CUDA sample filters not found: {', '.join(missing)}") + return selected + + @staticmethod + def _is_standalone_manifest(manifest: Path) -> bool: + if manifest.name == "Makefile": + return True + try: + content = manifest.read_text(encoding="utf-8", errors="replace") + except OSError: + return False + return ( + re.search(r"^\s*project\s*\(", content, re.IGNORECASE | re.MULTILINE) + is not None + ) + + @staticmethod + def _select_build_system(sample_dir: Path, requested: str) -> str: + if requested not in {"auto", "cmake", "make"}: + raise ValueError("build_system must be one of: auto, cmake, make") + if requested == "auto": + if (sample_dir / "CMakeLists.txt").exists(): + return "cmake" + if (sample_dir / "Makefile").exists(): + return "make" + raise ValueError(f"No supported build manifest in: {sample_dir}") + + manifest = "CMakeLists.txt" if requested == "cmake" else "Makefile" + if not (sample_dir / manifest).exists(): + raise ValueError(f"{manifest} not found in CUDA sample: {sample_dir}") + return requested + + def _build_compile_env( + self, platform: str, sms: str, compiler: Optional[str] = None + ) -> Dict[str, str]: + """Build environment variables for compilation.""" + env = os.environ.copy() + env["SMS"] = sms + + candidates = ( + (compiler,) if compiler else PLATFORM_CONFIGS[platform]["compilers"] + ) + compiler_path = next( + ( + resolved + for candidate in candidates + if (resolved := shutil.which(candidate, path=env.get("PATH"))) + ), + None, + ) + if not compiler_path: + raise FileNotFoundError( + f"No compiler found for platform {platform}; checked: " + + ", ".join(candidates) + ) + + env["CUDACXX"] = compiler_path + toolkit_root = str(Path(compiler_path).resolve().parent.parent) + env["CUDA_HOME"] = toolkit_root + env["CUDA_PATH"] = toolkit_root + if platform == "metax" and not env.get("MACA_PATH"): + maca_path = self._infer_metax_root(compiler_path) + if maca_path: + env["MACA_PATH"] = maca_path + + return env + + @staticmethod + def _infer_metax_root(compiler_path: str) -> Optional[str]: + path = Path(compiler_path).resolve() + if path.parts[-4:] == ("tools", "cu-bridge", "bin", "cucc"): + return str(path.parents[3]) + if path.parts[-3:] == ("mxgpu_llvm", "bin", "mxcc"): + return str(path.parents[2]) + return None + + def _compile_sample( + self, + sample_dir: Path, + build_dir: Path, + env: Dict[str, str], + timeout: int, + jobs: int, + build_system: str, + make_args: Optional[List[str]] = None, + ) -> Path: + """Compile one CUDA sample and return its executable.""" + sms = env.get("SMS", "80") + if build_system == "cmake": + wrapper_dir = build_dir / "source" + cmake_build_dir = build_dir / "build" + self._write_cmake_wrapper(wrapper_dir, sample_dir, sms) + cmake_build_dir.mkdir(parents=True, exist_ok=True) + configure_result = self._run_build_command( + [ + "cmake", + "-S", + str(wrapper_dir), + "-B", + str(cmake_build_dir), + f"-DCMAKE_CUDA_ARCHITECTURES={sms}", + f"-DCMAKE_CUDA_COMPILER={env['CUDACXX']}", + ], + sample_dir, + env, + timeout, + ) + if configure_result.returncode: + raise RuntimeError(self._command_error(configure_result)) + command = [ + "cmake", + "--build", + str(cmake_build_dir), + "--parallel", + str(jobs), + ] + search_root = cmake_build_dir + else: + self._run_build_command(["make", "clean"], sample_dir, env, timeout) + command = [ + "make", + f"-j{jobs}", + f"SMS={sms}", + f"NVCC={env['CUDACXX']}", + *(make_args or []), + ] + search_root = sample_dir + + build_result = self._run_build_command(command, sample_dir, env, timeout) + if build_result.returncode: + raise RuntimeError(self._command_error(build_result)) + + binary = self._find_sample_binary(search_root, sample_dir.name) + if not binary: + raise RuntimeError( + f"Build succeeded but no executable was found for {sample_dir.name}" + ) + return binary + + @staticmethod + def _write_cmake_wrapper(wrapper_dir: Path, sample_dir: Path, sms: str) -> None: + """Create an out-of-tree wrapper that owns the target architecture.""" + wrapper_dir.mkdir(parents=True, exist_ok=True) + sample_path = sample_dir.resolve().as_posix().replace('"', '\\"') + architecture = sms.replace('"', '\\"') + content = f"""cmake_minimum_required(VERSION 3.20) +project(InfiniBenchCudaSample LANGUAGES C CXX CUDA) + +add_subdirectory("{sample_path}" sample) + +function(infinibench_set_cuda_architectures directory) + get_property(targets DIRECTORY "${{directory}}" PROPERTY BUILDSYSTEM_TARGETS) + foreach(target IN LISTS targets) + get_target_property(target_type "${{target}}" TYPE) + if(NOT target_type STREQUAL "UTILITY" AND + NOT target_type STREQUAL "INTERFACE_LIBRARY") + set_property(TARGET "${{target}}" PROPERTY CUDA_ARCHITECTURES "{architecture}") + endif() + endforeach() + get_property(subdirectories DIRECTORY "${{directory}}" PROPERTY SUBDIRECTORIES) + foreach(subdirectory IN LISTS subdirectories) + infinibench_set_cuda_architectures("${{subdirectory}}") + endforeach() +endfunction() + +infinibench_set_cuda_architectures("{sample_path}") +""" + (wrapper_dir / "CMakeLists.txt").write_text(content, encoding="utf-8") + + @staticmethod + def _run_build_command( + command: List[str], cwd: Path, env: Dict[str, str], timeout: int + ) -> subprocess.CompletedProcess: + return subprocess.run( + command, + cwd=str(cwd), + capture_output=True, + text=True, + errors="replace", + env=env, + timeout=timeout, + ) + + @staticmethod + def _command_error(result: subprocess.CompletedProcess) -> str: + output = "\n".join(part for part in (result.stdout, result.stderr) if part) + return output[-2000:] or f"Command exited with code {result.returncode}" + + @staticmethod + def _find_sample_binary(search_root: Path, sample_name: str) -> Optional[Path]: + for name in (sample_name, sample_name.replace("_", "")): + direct = search_root / name + if direct.is_file() and os.access(direct, os.X_OK): + return direct + matches = sorted( + path + for path in search_root.rglob(name) + if path.is_file() and os.access(path, os.X_OK) + ) + if matches: + return matches[0] + return None + + def _run_sample( + self, binary: Path, timeout: int, env: Optional[Dict[str, str]] = None + ) -> Tuple[str, str]: + """Run a compiled CUDA sample.""" + try: + result = subprocess.run( + [str(binary)], + cwd=str(binary.parent), + capture_output=True, + text=True, + errors="replace", + env=env, + timeout=timeout, + ) + return self._classify_run_result(result) + except subprocess.TimeoutExpired: + return "fail", f"Execution timed out after {timeout} seconds" + except FileNotFoundError as exc: + return "fail", str(exc) + + @staticmethod + def _classify_run_result( + result: subprocess.CompletedProcess, + ) -> Tuple[str, str]: + output = "\n".join(part for part in (result.stdout, result.stderr) if part) + normalized = output.lower() + if result.returncode == 2 or any( + marker in normalized + for marker in ("sample waived", "waiving sample", "result = waived") + ): + return "skip", output[-2000:] or "Sample waived" + if result.returncode == 0 and not re.search(r"result\s*=\s*fail", normalized): + return "pass", "" + return ( + "fail", + output[-2000:] or f"Executable exited with code {result.returncode}", + ) + + def _clean_make_sample( + self, sample_dir: Path, env: Dict[str, str], timeout: int + ) -> None: + try: + self._run_build_command(["make", "clean"], sample_dir, env, timeout) + except (FileNotFoundError, subprocess.TimeoutExpired): + logger.warning("Failed to clean CUDA sample build: %s", sample_dir) diff --git a/infinimetrics/compatibility/constants.py b/infinimetrics/compatibility/constants.py new file mode 100644 index 00000000..39fc7997 --- /dev/null +++ b/infinimetrics/compatibility/constants.py @@ -0,0 +1,27 @@ +"""CUDA Samples compatibility build configuration.""" + +CUDA_SAMPLE_CONFIGS = { + "cuda": { + "sms": "80", + "make_args": (), + }, + "metax": { + "sms": "70", + "make_args": ( + "ALL_CCFLAGS=--std=c++11", + "ALL_LDFLAGS=", + "GENCODE_FLAGS=", + ), + }, + "corex": { + "sms": "ivcore20", + "make_args": ( + "ALL_CCFLAGS=-x ivcore --cuda-gpu-arch=ivcore20 " + "--cuda-path=/usr/local/corex --std=c++11", + "ALL_LDFLAGS=--cuda-gpu-arch=ivcore20 " + "--cuda-path=/usr/local/corex -L/usr/local/corex/lib " + "-Wl,-rpath,/usr/local/corex/lib -lcudart", + "GENCODE_FLAGS=", + ), + }, +} diff --git a/infinimetrics/dispatcher.py b/infinimetrics/dispatcher.py index 75856cc1..ba3b1b43 100644 --- a/infinimetrics/dispatcher.py +++ b/infinimetrics/dispatcher.py @@ -23,6 +23,10 @@ (TestCategory.INFER, "vllm"): lambda: _create_inference_adapter(), (TestCategory.TRAIN, "megatron"): lambda: _create_training_adapter(), (TestCategory.TRAIN, "infinitrain"): lambda: _create_training_adapter(), + ( + TestCategory.COMPATIBILITY, + "cudasamples", + ): lambda: _create_compatibility_adapter(), } @@ -61,6 +65,13 @@ def _create_training_adapter(): return TrainingAdapter() +def _create_compatibility_adapter(): + """Create compatibility adapter (lazy import).""" + from infinimetrics.compatibility.compatibility_adapter import CompatibilityAdapter + + return CompatibilityAdapter() + + class Dispatcher: """Test orchestration dispatcher for managing test executions.""" diff --git a/infinimetrics/hardware/constants.py b/infinimetrics/hardware/constants.py index e2b117f8..7f41cf24 100644 --- a/infinimetrics/hardware/constants.py +++ b/infinimetrics/hardware/constants.py @@ -19,29 +19,58 @@ "benchmark_subdir": "cuda-memory-benchmark", "build_platform": "cuda", "cache_parser": "cuda", + "compilers": ("nvcc", "/usr/local/cuda/bin/nvcc"), + "detection_tools": ("nvcc", "nvidia-smi"), }, "metax": { "binary_name": "cuda_perf_suite", "benchmark_subdir": "cuda-memory-benchmark", "build_platform": "metax", "cache_parser": "cuda", + "compilers": ( + "/opt/maca/tools/cu-bridge/bin/cucc", + "cucc", + "/opt/maca/mxgpu_llvm/bin/mxcc", + "mxcc", + ), + "detection_tools": ( + "/opt/maca/tools/cu-bridge/bin/cucc", + "cucc", + "mxcc", + ), }, "corex": { "binary_name": "cuda_perf_suite", "benchmark_subdir": "cuda-memory-benchmark", "build_platform": "corex", "cache_parser": "cuda", + "compilers": ( + "/usr/local/corex/bin/clang++", + "/usr/local/corex/bin/nvcc", + "nvcc", + ), + "detection_tools": ( + "/usr/local/corex/bin/ixsmi", + "/usr/local/corex/bin/clang++", + ), }, "hygon": { "binary_name": "cuda_perf_suite", "benchmark_subdir": "cuda-memory-benchmark", "build_platform": "hygon", "cache_parser": "cuda", + "compilers": ("/opt/dtk/bin/hipcc", "hipcc"), + "detection_tools": ("/opt/dtk/bin/hy-smi", "hy-smi"), + "conditional_detection_tools": (("/opt/dtk", "hipcc"),), }, "moore": { "binary_name": "cuda_perf_suite", "benchmark_subdir": "cuda-memory-benchmark", "build_platform": "moore", "cache_parser": "cuda", + "compilers": ("/usr/local/musa/bin/mcc", "mcc"), + "detection_tools": ("mcc", "mthreads-gmi"), }, } + +PLATFORM_DETECTION_ORDER = ("moore", "metax", "hygon", "corex") diff --git a/infinimetrics/hardware/hardware_adapter.py b/infinimetrics/hardware/hardware_adapter.py index 1a86a457..b530a9dd 100644 --- a/infinimetrics/hardware/hardware_adapter.py +++ b/infinimetrics/hardware/hardware_adapter.py @@ -25,7 +25,11 @@ InfiniMetricsJson, ) from infinimetrics.common.csv_utils import create_timeseries_metric -from infinimetrics.hardware.constants import PLATFORM_ALIASES, PLATFORM_CONFIGS +from infinimetrics.hardware.constants import ( + PLATFORM_ALIASES, + PLATFORM_CONFIGS, + PLATFORM_DETECTION_ORDER, +) from infinimetrics.utils.time_utils import get_timestamp logger = logging.getLogger(__name__) @@ -33,31 +37,31 @@ def detect_platform() -> str: """Detect the installed accelerator toolchain.""" - if shutil.which("mcc") or shutil.which("mthreads-gmi"): - return "moore" - - maca_path = Path("/opt/maca") - if ( - (maca_path / "tools" / "cu-bridge" / "bin" / "cucc").exists() - or shutil.which("cucc") - or shutil.which("mxcc") - ): - return "metax" + for platform in PLATFORM_DETECTION_ORDER: + platform_config = PLATFORM_CONFIGS[platform] + tools = platform_config["detection_tools"] + if any(_tool_exists(tool) for tool in tools): + return platform + conditional_tools = platform_config.get("conditional_detection_tools", ()) + if any( + _path_exists(root) and _tool_exists(tool) + for root, tool in conditional_tools + ): + return platform + return "cuda" - dtk_path = Path("/opt/dtk") - if ( - (dtk_path / "bin" / "hy-smi").exists() - or shutil.which("hy-smi") - or (dtk_path.exists() and shutil.which("hipcc")) - ): - return "hygon" - corex_path = Path("/usr/local/corex") - if (corex_path / "bin" / "ixsmi").exists() or ( - corex_path / "bin" / "clang++" - ).exists(): - return "corex" - return "cuda" +def _tool_exists(tool: str) -> bool: + """Return whether a command or absolute tool path is available.""" + path = Path(tool) + if path.is_absolute(): + return path.is_file() + return shutil.which(tool) is not None + + +def _path_exists(path: str) -> bool: + """Return whether a platform-specific installation path exists.""" + return Path(path).exists() class HardwareTestAdapter(BaseAdapter): diff --git a/scripts/generate_compat_report.py b/scripts/generate_compat_report.py new file mode 100644 index 00000000..f6783e13 --- /dev/null +++ b/scripts/generate_compat_report.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +"""Generate a human-readable CUDA compatibility report.""" + +import argparse +import csv +import json +from pathlib import Path +from typing import Any, Dict, List + + +def _load_result(result_file: str) -> Dict[str, Any]: + with Path(result_file).open(encoding="utf-8") as input_file: + data = json.load(input_file) + if not isinstance(data, dict): + raise ValueError("Compatibility result must be a JSON object") + return data + + +def _metric_values(data: Dict[str, Any]) -> Dict[str, Any]: + return { + metric["name"]: metric.get("value") + for metric in data.get("metrics", []) + if isinstance(metric, dict) and "name" in metric + } + + +def _error_summary(detail: Dict[str, Any]) -> str: + error = str(detail.get("error", "")) + meaningful = [line.strip() for line in error.splitlines() if line.strip()] + if not meaningful: + return "No error output captured" + preferred = next( + ( + line + for line in meaningful + if "error" in line.lower() or "fatal" in line.lower() + ), + meaningful[0], + ) + return preferred[:120] + + +def _append_detail_section( + lines: List[str], title: str, details: List[Dict[str, Any]], total: int +) -> None: + lines.extend(["", "-" * 80, f" {title} ({len(details)}/{total})", "-" * 80]) + if not details: + lines.append(" (none)") + return + for detail in details: + lines.append(f" {detail.get('name', '')}") + lines.append(f" {_error_summary(detail)}") + + +def generate_report(result_file: str) -> str: + data = _load_result(result_file) + metrics = _metric_values(data) + config = data.get("config") + if not isinstance(config, dict): + config = {} + raw_details = metrics.get("compatibility.cuda_samples.details", []) + details = raw_details if isinstance(raw_details, list) else [] + + total = int(metrics.get("compatibility.cuda_samples.total", 0) or 0) + compile_passed = int( + metrics.get("compatibility.cuda_samples.compile_passed", 0) or 0 + ) + compile_failed = int( + metrics.get("compatibility.cuda_samples.compile_failed", 0) or 0 + ) + compile_rate = metrics.get("compatibility.cuda_samples.compile_pass_rate", 0) + run_passed = int(metrics.get("compatibility.cuda_samples.run_passed", 0) or 0) + run_failed = int(metrics.get("compatibility.cuda_samples.run_failed", 0) or 0) + run_skipped = metrics.get("compatibility.cuda_samples.run_skipped") + if run_skipped is None: + run_skipped = max(compile_passed - run_passed - run_failed, 0) + run_skipped = int(run_skipped) + run_rate = metrics.get("compatibility.cuda_samples.run_pass_rate", 0) + + lines = [ + "=" * 80, + " CUDA Compatibility Test Report", + "=" * 80, + f"Run ID: {data.get('run_id', 'N/A')}", + f"Time: {data.get('time', 'N/A')}", + f"Platform: {config.get('platform', 'N/A')}", + f"Result code: {data.get('result_code', 'N/A')}", + "", + "-" * 80, + " Summary", + "-" * 80, + f" Samples: {total}", + f" Compile passed: {compile_passed} ({compile_rate}%)", + f" Compile failed: {compile_failed}", + f" Run passed: {run_passed} ({run_rate}%)", + f" Run failed: {run_failed}", + f" Run skipped: {run_skipped}", + ] + + all_passed = [ + detail + for detail in details + if detail.get("compile_result") == "pass" and detail.get("run_result") == "pass" + ] + compile_failures = [ + detail for detail in details if detail.get("compile_result") == "fail" + ] + run_failures = [ + detail + for detail in details + if detail.get("compile_result") == "pass" and detail.get("run_result") == "fail" + ] + skipped = [ + detail + for detail in details + if detail.get("compile_result") == "pass" and detail.get("run_result") == "skip" + ] + + _append_detail_section(lines, "Compile and run passed", all_passed, total) + _append_detail_section(lines, "Compile failures", compile_failures, total) + _append_detail_section(lines, "Run failures", run_failures, total) + _append_detail_section(lines, "Skipped or waived", skipped, total) + lines.extend(["", "=" * 80]) + return "\n".join(lines) + + +def write_csv(result_file: str, csv_file: str) -> None: + data = _load_result(result_file) + details = _metric_values(data).get("compatibility.cuda_samples.details", []) + if not isinstance(details, list): + raise ValueError("compatibility.cuda_samples.details must be a list") + + with Path(csv_file).open("w", newline="", encoding="utf-8") as output_file: + writer = csv.DictWriter( + output_file, + fieldnames=("sample_name", "compile_result", "run_result", "error"), + ) + writer.writeheader() + for detail in details: + writer.writerow( + { + "sample_name": detail.get("name", ""), + "compile_result": detail.get("compile_result", ""), + "run_result": detail.get("run_result", ""), + "error": detail.get("error", ""), + } + ) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("result_file") + parser.add_argument("--csv", dest="csv_file") + args = parser.parse_args() + + print(generate_report(args.result_file)) + if args.csv_file: + write_csv(args.result_file, args.csv_file) + print(f"\nCSV saved to: {args.csv_file}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_compatibility_adapter.py b/tests/test_compatibility_adapter.py new file mode 100644 index 00000000..44ae4661 --- /dev/null +++ b/tests/test_compatibility_adapter.py @@ -0,0 +1,329 @@ +import os +import subprocess +from pathlib import Path + +import pytest + +from infinimetrics.compatibility.compatibility_adapter import CompatibilityAdapter +from infinimetrics.compatibility.constants import CUDA_SAMPLE_CONFIGS +from infinimetrics.dispatcher import Dispatcher + + +def _sample(root: Path, category: str, name: str, manifest: str) -> Path: + sample_dir = root / "Samples" / category / name + sample_dir.mkdir(parents=True) + content = f"project({name})\n" if manifest == "CMakeLists.txt" else "# test\n" + (sample_dir / manifest).write_text(content, encoding="utf-8") + return sample_dir + + +@pytest.fixture +def compiler_available(monkeypatch): + monkeypatch.setattr( + "infinimetrics.compatibility.compatibility_adapter.shutil.which", + lambda candidate, **_: candidate, + ) + + +def test_discovers_nested_cmake_and_make_samples(tmp_path): + (tmp_path / "Samples").mkdir() + (tmp_path / "Samples" / "CMakeLists.txt").write_text("# aggregate\n") + category = tmp_path / "Samples" / "0_Introduction" + category.mkdir() + (category / "CMakeLists.txt").write_text("# aggregate\n") + vector_add = _sample(tmp_path, "0_Introduction", "vectorAdd", "CMakeLists.txt") + clock = _sample(tmp_path, "0_Introduction", "clock", "Makefile") + + discovered = CompatibilityAdapter()._discover_sample_dirs( + tmp_path / "Samples", None + ) + + assert discovered == [clock, vector_add] + + +def test_sample_filter_rejects_unknown_names(tmp_path): + _sample(tmp_path, "0_Introduction", "vectorAdd", "CMakeLists.txt") + + with pytest.raises(ValueError, match="missingSample"): + CompatibilityAdapter()._discover_sample_dirs( + tmp_path / "Samples", ["vectorAdd", "missingSample"] + ) + + +def test_discovery_excludes_nested_cmake_group_manifests(tmp_path): + group = tmp_path / "Samples" / "8_Platform_Specific" / "Tegra" + group.mkdir(parents=True) + (group / "CMakeLists.txt").write_text( + "add_subdirectory(simpleGLES)\n", encoding="utf-8" + ) + sample = _sample( + tmp_path, + "8_Platform_Specific/Tegra", + "simpleGLES", + "CMakeLists.txt", + ) + + discovered = CompatibilityAdapter()._discover_sample_dirs( + tmp_path / "Samples", None + ) + + assert discovered == [sample] + + +def test_sample_filter_rejects_an_empty_list(tmp_path): + _sample(tmp_path, "0_Introduction", "vectorAdd", "CMakeLists.txt") + + with pytest.raises(ValueError, match="non-empty"): + CompatibilityAdapter()._discover_sample_dirs(tmp_path / "Samples", []) + + +@pytest.mark.parametrize("sms", ["", '80")\nmessage(FATAL_ERROR injected)']) +def test_architecture_rejects_empty_or_cmake_control_characters(sms): + with pytest.raises(ValueError, match="invalid architecture"): + CompatibilityAdapter._validate_architectures(sms) + + +@pytest.mark.parametrize( + ("build_system", "manifest", "sms", "compiler", "jobs"), + [ + ("cmake", "CMakeLists.txt", "80", "/usr/local/cuda/bin/nvcc", 7), + ("make", "Makefile", "70", "/usr/local/musa/bin/mcc", 3), + ], +) +def test_build_commands_include_compiler_arch_and_jobs( + tmp_path, monkeypatch, build_system, manifest, sms, compiler, jobs +): + sample_dir = _sample(tmp_path, "0_Introduction", "vectorAdd", manifest) + build_dir = tmp_path / "build" + expected_binary = ( + build_dir / "build" if build_system == "cmake" else sample_dir + ) / "vectorAdd" + calls = [] + + def fake_run(command, **kwargs): + calls.append(command) + is_build = command[:2] == ["cmake", "--build"] or ( + command[0] == "make" and "clean" not in command + ) + if is_build: + expected_binary.parent.mkdir(parents=True, exist_ok=True) + expected_binary.write_text("binary", encoding="utf-8") + expected_binary.chmod(0o755) + return subprocess.CompletedProcess(command, 0, "", "") + + monkeypatch.setattr(subprocess, "run", fake_run) + binary = CompatibilityAdapter()._compile_sample( + sample_dir, build_dir, {"SMS": sms, "CUDACXX": compiler}, 30, jobs, build_system + ) + + expected = ( + [ + "cmake", + "--build", + str(build_dir / "build"), + "--parallel", + str(jobs), + ] + if build_system == "cmake" + else ["make", f"-j{jobs}", f"SMS={sms}", f"NVCC={compiler}"] + ) + assert binary == expected_binary + assert expected in calls + + +def test_cmake_wrapper_overrides_sample_target_architectures(tmp_path): + sample_dir = _sample(tmp_path, "0_Introduction", "vectorAdd", "CMakeLists.txt") + wrapper_dir = tmp_path / "wrapper" + + CompatibilityAdapter._write_cmake_wrapper(wrapper_dir, sample_dir, "ivcore20") + + wrapper = (wrapper_dir / "CMakeLists.txt").read_text(encoding="utf-8") + assert f'add_subdirectory("{sample_dir.resolve().as_posix()}" sample)' in wrapper + assert 'PROPERTY CUDA_ARCHITECTURES "ivcore20"' in wrapper + + +@pytest.mark.parametrize("platform", ["metax", "corex"]) +def test_vendor_make_args_remove_nvidia_only_flags(platform): + args = " ".join(CUDA_SAMPLE_CONFIGS[platform]["make_args"]) + + assert all(flag not in args for flag in ("--threads", "-gencode", "-m64")) + + +def test_corex_make_args_keep_source_language_out_of_link_step(): + compile_args, link_args, _ = CUDA_SAMPLE_CONFIGS["corex"]["make_args"] + + assert "-x ivcore" in compile_args + assert "-x ivcore" not in link_args + + +def test_compile_env_matches_selected_vendor_toolkit(monkeypatch, compiler_available): + compiler = "/usr/local/corex/bin/clang++" + monkeypatch.setenv("CUDA_HOME", "/usr/local/cuda") + monkeypatch.setenv("CUDA_PATH", "/usr/local/cuda") + env = CompatibilityAdapter()._build_compile_env("corex", "ivcore20", compiler) + + expected_root = str(Path(compiler).resolve().parent.parent) + assert env["CUDACXX"] == compiler + assert env["CUDA_HOME"] == expected_root + assert env["CUDA_PATH"] == expected_root + + +@pytest.mark.parametrize( + ("compiler", "root_parent_index"), + [ + ("/opt/maca/tools/cu-bridge/bin/cucc", 3), + ("/opt/maca/mxgpu_llvm/bin/mxcc", 2), + ], +) +def test_metax_compile_env_infers_maca_path( + monkeypatch, compiler_available, compiler, root_parent_index +): + monkeypatch.delenv("MACA_PATH", raising=False) + + env = CompatibilityAdapter()._build_compile_env("metax", "70", compiler) + + assert env["MACA_PATH"] == str(Path(compiler).resolve().parents[root_parent_index]) + + +def test_metax_compile_env_preserves_explicit_maca_path( + monkeypatch, compiler_available +): + compiler = "/opt/maca/tools/cu-bridge/bin/cucc" + monkeypatch.setenv("MACA_PATH", "/custom/maca") + env = CompatibilityAdapter()._build_compile_env("metax", "70", compiler) + + assert env["MACA_PATH"] == "/custom/maca" + + +def test_cuda_sample_metrics_keep_skips_separate(tmp_path, monkeypatch): + samples = [ + _sample(tmp_path, "0_Introduction", name, "CMakeLists.txt") + for name in ("passes", "fails", "waived") + ] + binaries = {sample.name: tmp_path / f"{sample.name}.bin" for sample in samples} + adapter = CompatibilityAdapter() + + monkeypatch.setattr(adapter, "_discover_sample_dirs", lambda *_: samples) + monkeypatch.setattr( + adapter, + "_build_compile_env", + lambda *_: {"SMS": "80", "CUDACXX": "/usr/local/cuda/bin/nvcc"}, + ) + monkeypatch.setattr( + adapter, + "_compile_sample", + lambda sample_dir, *_: binaries[sample_dir.name], + ) + outcomes = { + "passes.bin": ("pass", ""), + "fails.bin": ("fail", "kernel failed"), + "waived.bin": ("skip", "sample waived"), + } + monkeypatch.setattr( + adapter, "_run_sample", lambda binary, *_: outcomes[binary.name] + ) + + config = {"platform": "nvidia", "cuda_samples_dir": str(tmp_path)} + metrics = adapter._run_cuda_samples_test(config) + values = {metric["name"]: metric["value"] for metric in metrics} + + assert config["compiler"] == "/usr/local/cuda/bin/nvcc" + assert config["sms"] == "80" + assert values["compatibility.cuda_samples.run_passed"] == 1 + assert values["compatibility.cuda_samples.run_failed"] == 1 + assert values["compatibility.cuda_samples.run_skipped"] == 1 + assert values["compatibility.cuda_samples.run_pass_rate"] == 33.33 + + +@pytest.mark.parametrize("failure_stage", ["compile", "run"]) +def test_sample_failures_are_accounted_for(tmp_path, monkeypatch, failure_stage): + sample = _sample(tmp_path, "0_Introduction", "vectorAdd", "CMakeLists.txt") + adapter = CompatibilityAdapter() + monkeypatch.setattr(adapter, "_discover_sample_dirs", lambda *_: [sample]) + monkeypatch.setattr( + adapter, + "_build_compile_env", + lambda *_: {"SMS": "80", "CUDACXX": "/usr/local/cuda/bin/nvcc"}, + ) + + def compile_sample(*_): + if failure_stage == "compile": + raise RuntimeError("compiler failed") + return tmp_path / "vectorAdd" + + def run_sample(*_): + raise RuntimeError("runtime failed") + + monkeypatch.setattr(adapter, "_compile_sample", compile_sample) + monkeypatch.setattr(adapter, "_run_sample", run_sample) + + metrics = adapter._run_cuda_samples_test( + {"platform": "nvidia", "cuda_samples_dir": str(tmp_path)} + ) + values = {metric["name"]: metric["value"] for metric in metrics} + details = values["compatibility.cuda_samples.details"] + + compile_passed = int(failure_stage == "run") + assert values["compatibility.cuda_samples.compile_passed"] == compile_passed + assert values["compatibility.cuda_samples.run_failed"] == compile_passed + assert values["compatibility.cuda_samples.run_skipped"] == 0 + assert details[0]["compile_result"] == ("pass" if compile_passed else "fail") + assert details[0]["run_result"] == ("fail" if compile_passed else "not_run") + + +def test_run_sample_classifies_cuda_waiver(tmp_path): + binary = tmp_path / "sample" + binary.write_text("binary", encoding="utf-8") + binary.chmod(0o755) + adapter = CompatibilityAdapter() + + result, error = adapter._classify_run_result( + subprocess.CompletedProcess([str(binary)], 2, "Sample waived", "") + ) + + assert result == "skip" + assert "waived" in error.lower() + + +def test_run_sample_uses_the_binary_directory(tmp_path, monkeypatch): + binary = tmp_path / "build" / "sample" + binary.parent.mkdir() + calls = [] + + def fake_run(command, **kwargs): + calls.append((command, kwargs)) + return subprocess.CompletedProcess(command, 0, "Result = PASS", "") + + monkeypatch.setattr(subprocess, "run", fake_run) + + result, _ = CompatibilityAdapter()._run_sample(binary, 30) + + assert result == "pass" + assert calls == [ + ( + [str(binary)], + { + "cwd": str(binary.parent), + "capture_output": True, + "text": True, + "errors": "replace", + "env": None, + "timeout": 30, + }, + ) + ] + + +def test_metrics_reject_an_empty_sample_set(): + with pytest.raises(ValueError, match="No CUDA samples selected"): + CompatibilityAdapter._build_metrics([]) + + +def test_dispatcher_only_registers_cuda_samples_compatibility(): + assert isinstance( + Dispatcher()._create_adapter("compatibility", "cudasamples"), + CompatibilityAdapter, + ) + for framework in ("megatron", "vllm", "infinilm"): + with pytest.raises(ValueError, match="Adapter not registered"): + Dispatcher()._create_adapter("compatibility", framework) diff --git a/tests/test_generate_compat_report.py b/tests/test_generate_compat_report.py new file mode 100644 index 00000000..f2faf3e9 --- /dev/null +++ b/tests/test_generate_compat_report.py @@ -0,0 +1,72 @@ +import csv +import importlib.util +import json +from pathlib import Path + +SCRIPT = Path(__file__).parents[1] / "scripts" / "generate_compat_report.py" +SPEC = importlib.util.spec_from_file_location("generate_compat_report", SCRIPT) +REPORT_MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(REPORT_MODULE) + + +def _result_file(tmp_path: Path) -> Path: + result = { + "run_id": "compat-test", + "time": "2026-08-05 12:00:00", + "result_code": 0, + "config": {"platform": "nvidia"}, + "metrics": [ + {"name": "compatibility.cuda_samples.total", "value": 3}, + {"name": "compatibility.cuda_samples.compile_passed", "value": 3}, + {"name": "compatibility.cuda_samples.compile_failed", "value": 0}, + {"name": "compatibility.cuda_samples.compile_pass_rate", "value": 100.0}, + {"name": "compatibility.cuda_samples.run_passed", "value": 1}, + {"name": "compatibility.cuda_samples.run_failed", "value": 1}, + {"name": "compatibility.cuda_samples.run_skipped", "value": 1}, + {"name": "compatibility.cuda_samples.run_pass_rate", "value": 33.33}, + { + "name": "compatibility.cuda_samples.details", + "value": [ + { + "name": "passes", + "compile_result": "pass", + "run_result": "pass", + "error": "", + }, + { + "name": "fails", + "compile_result": "pass", + "run_result": "fail", + "error": "kernel, failed", + }, + { + "name": "waived", + "compile_result": "pass", + "run_result": "skip", + "error": "waived", + }, + ], + }, + ], + } + path = tmp_path / "result.json" + path.write_text(json.dumps(result), encoding="utf-8") + return path + + +def test_report_renders_pass_fail_and_skip_sections(tmp_path): + report = REPORT_MODULE.generate_report(str(_result_file(tmp_path))) + + assert "CUDA Compatibility Test Report" in report + assert "Run failures (1/3)" in report + assert "Skipped or waived (1/3)" in report + + +def test_csv_writer_quotes_errors(tmp_path): + csv_path = tmp_path / "report.csv" + + REPORT_MODULE.write_csv(str(_result_file(tmp_path)), str(csv_path)) + + with csv_path.open(newline="", encoding="utf-8") as csv_file: + rows = list(csv.DictReader(csv_file)) + assert rows[1]["error"] == "kernel, failed" diff --git a/tests/test_hardware_adapter.py b/tests/test_hardware_adapter.py index 4262fab2..adaa1783 100644 --- a/tests/test_hardware_adapter.py +++ b/tests/test_hardware_adapter.py @@ -4,8 +4,7 @@ from infinimetrics.dispatcher import Dispatcher from infinimetrics.hardware import hardware_adapter -from infinimetrics.hardware.hardware_adapter import HardwareTestAdapter - +from infinimetrics.hardware.hardware_adapter import HardwareTestAdapter, detect_platform CUDA_OUTPUT = """ Direction: Host to Device @@ -175,6 +174,32 @@ def test_runtime_detection_ignores_testcase_framework(tmp_path, monkeypatch): assert adapter._get_device_type({}) == "moore" +def test_runtime_detection_uses_shared_detection_tools(monkeypatch): + monkeypatch.setattr( + hardware_adapter, + "_tool_exists", + lambda tool: tool == "/opt/maca/tools/cu-bridge/bin/cucc", + ) + + assert detect_platform() == "metax" + + +def test_runtime_detection_does_not_treat_rocm_hipcc_as_hygon(monkeypatch): + monkeypatch.setattr(hardware_adapter, "_tool_exists", lambda tool: tool == "hipcc") + monkeypatch.setattr(hardware_adapter, "_path_exists", lambda *_: False) + + assert detect_platform() == "cuda" + + +def test_runtime_detection_accepts_hipcc_inside_dtk(monkeypatch): + monkeypatch.setattr(hardware_adapter, "_tool_exists", lambda tool: tool == "hipcc") + monkeypatch.setattr( + hardware_adapter, "_path_exists", lambda path: path == "/opt/dtk" + ) + + assert detect_platform() == "hygon" + + @pytest.mark.parametrize("device", ["cuda", "metax", "corex", "hygon", "moore"]) def test_cuda_compatible_platforms_share_binary(tmp_path, device): cuda_binary = tmp_path / "cuda_perf_suite"