diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 30dc1b7..8899f12 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -38,3 +38,9 @@ jobs: pip install -e '.[dev]' python3 -c 'import kerneldb' " + - name: Run pytest + run: | + bash .github/scripts/container_exec.sh " + pip install -e '.[dev]' + pytest tests/ -v + " diff --git a/pyproject.toml b/pyproject.toml index ad167e0..be2068f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,11 @@ kerneldb = ["*.so", "*.so.*"] [tool.setuptools.dynamic] version = {attr = "setuptools_scm.get_version"} +# ---- pytest configuration ---- +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-v" + # ---- setuptools-scm versioning ---- [tool.setuptools_scm] version_scheme = "post-release" # .postN after last tag diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..361b094 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,27 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved. + +import shutil + +import pytest + + +def _hipcc_available(): + return shutil.which("hipcc") is not None + + +def _gpu_available(): + try: + from kerneldb import _kerneldb + + if _kerneldb.hsa_init() != 0: + return False + return _kerneldb.get_first_gpu_agent().handle != 0 + except Exception: + return False + + +requires_rocm = pytest.mark.skipif( + not (_hipcc_available() and _gpu_available()), + reason="Requires hipcc and a GPU", +) diff --git a/tests/test_arguments.py b/tests/test_arguments.py new file mode 100644 index 0000000..3121734 --- /dev/null +++ b/tests/test_arguments.py @@ -0,0 +1,218 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved. + +""" +Tests for KernelDB kernel argument extraction: basic metadata, nested structs, +typedef resolution, and template instantiations. + +All tests require a ROCm environment with hipcc and a GPU. +""" + +import subprocess +import tempfile +from pathlib import Path + +import pytest + +from conftest import requires_rocm + +kerneldb = pytest.importorskip("kerneldb", reason="kernelDB C++ extension not available") +KernelDB = kerneldb.KernelDB + + +def _compile(source, name): + tmp = Path(tempfile.mkdtemp(prefix=f"kerneldb_{name}_")) + src = tmp / f"{name}.cpp" + exe = tmp / name + src.write_text(source) + r = subprocess.run(["hipcc", "-g", str(src), "-o", str(exe)], + capture_output=True, text=True) + if r.returncode != 0: + pytest.skip(f"hipcc compilation failed:\n{r.stderr}") + return str(exe) + + +def _find_kernel(kernels, fragment): + matches = [k for k in kernels if fragment in k] + if not matches: + pytest.skip(f"No kernel containing {fragment!r} found") + return matches[0] + + +def _arg_by_name(arguments, name): + return next((a for a in arguments if a.name == name), None) + + +_ARGS_SOURCE = r""" +#include + +__global__ void kernel_with_args(double* a, double* b, double* c, int n) { + int idx = blockIdx.x * blockDim.x + threadIdx.x; + if (idx < n) { + c[idx] = a[idx] + b[idx]; + } +} + +int main() { return 0; } +""" + +_NESTED_SOURCE = r""" +#include + +struct Point3D { + float x; + float y; + float z; +}; + +struct BoundingBox { + Point3D min_pt; + Point3D max_pt; +}; + +__global__ void update_bounds(BoundingBox* boxes, int count) { + int idx = blockIdx.x * blockDim.x + threadIdx.x; + if (idx < count) { + boxes[idx].min_pt.x = 0.0f; + } +} + +int main() { return 0; } +""" + +_TYPEDEF_SOURCE = r""" +#include + +using MyInt = int; +using MyFloat = float; +typedef double MyDouble; + +__global__ void typedef_kernel(MyInt a, MyFloat b, MyDouble c, int* d) { + int idx = blockIdx.x * blockDim.x + threadIdx.x; + if (idx == 0) { + d[0] = a + static_cast(b) + static_cast(c); + } +} + +int main() { return 0; } +""" + +_TEMPLATE_SOURCE = r""" +#include + +template +__global__ void scale_values(T* input, T* output, T factor, int n) { + int idx = blockIdx.x * blockDim.x + threadIdx.x; + if (idx < n) { + output[idx] = input[idx] * factor; + } +} + +template __global__ void scale_values(float*, float*, float, int); +template __global__ void scale_values(double*, double*, double, int); + +int main() { return 0; } +""" + +_cache = {} + + +def _get_binary(source, name): + if name not in _cache: + _cache[name] = _compile(source, name) + return _cache[name] + + +@requires_rocm +def test_kernel_arguments(): + kdb = KernelDB(_get_binary(_ARGS_SOURCE, "args")) + name = _find_kernel(kdb.get_kernels(), "kernel_with_args") + args = kdb.get_kernel_arguments(name) + + assert isinstance(args, list) and len(args) == 4 + for arg in args: + assert isinstance(arg.name, str) and arg.name + assert isinstance(arg.type_name, str) and arg.type_name + assert isinstance(arg.size, int) and arg.size > 0 + assert isinstance(arg.alignment, int) and arg.alignment > 0 + + for expected in ("a", "b", "c", "n"): + assert expected in [a.name for a in args] + + assert _arg_by_name(args, "a").size == 8 # pointer = 8 bytes + n_arg = _arg_by_name(args, "n") + assert n_arg.size == 4 + assert "int" in n_arg.type_name.lower() + + +@requires_rocm +def test_kernel_wrapper_arguments(): + kdb = KernelDB(_get_binary(_ARGS_SOURCE, "args")) + name = _find_kernel(kdb.get_kernels(), "kernel_with_args") + kernel = kdb.get_kernel(name) + + assert kernel.has_arguments() is True + assert isinstance(kernel.arguments, list) and kernel.arguments + + +@requires_rocm +def test_nested_struct_arguments(): + kdb = KernelDB(_get_binary(_NESTED_SOURCE, "nested")) + name = _find_kernel(kdb.get_kernels(), "update_bounds") + args = kdb.get_kernel_arguments(name) + + for arg in args: + assert hasattr(arg, "members") + + def find_type(arg_list, fragment): + for arg in arg_list: + if fragment in arg.type_name: + return arg + if arg.members: + found = find_type(arg.members, fragment) + if found: + return found + return None + + point = find_type(args, "Point3D") + assert point is not None, "Point3D not found in argument tree" + assert len(point.members) == 3 + + +@requires_rocm +def test_typedef_resolution(): + kdb = KernelDB(_get_binary(_TYPEDEF_SOURCE, "typedef")) + name = _find_kernel(kdb.get_kernels(), "typedef_kernel") + + args_raw = kdb.get_kernel_arguments(name, resolve_typedefs=False) + assert len(args_raw) == 4 + raw_types = [a.type_name for a in args_raw] + assert any(n in ("MyInt", "MyFloat", "MyDouble") for n in raw_types) + + args_resolved = kdb.get_kernel_arguments(name, resolve_typedefs=True) + resolved_types = [a.type_name for a in args_resolved] + for alias in ("MyInt", "MyFloat", "MyDouble"): + assert alias not in resolved_types + + primitives = {"int", "float", "double", "unsigned int"} + for t in resolved_types: + base = t.replace("*", "").strip() + assert any(p in base for p in primitives), f"{t!r} doesn't look primitive" + + +@requires_rocm +def test_template_kernel_arguments(): + kdb = KernelDB(_get_binary(_TEMPLATE_SOURCE, "template")) + kernels = kdb.get_kernels() + + scale_kernels = [k for k in kernels if "scale_values" in k] + assert len(scale_kernels) >= 2 + + for type_frag in ("float", "double"): + matching = [k for k in scale_kernels if type_frag in k] + if not matching: + pytest.skip(f"scale_values<{type_frag}> not found") + args = kdb.get_kernel_arguments(matching[0]) + assert len(args) == 4 + non_int = [a for a in args if a.name != "n"] + assert all(type_frag in a.type_name.lower() for a in non_int) diff --git a/tests/test_imports.py b/tests/test_imports.py new file mode 100644 index 0000000..0b57196 --- /dev/null +++ b/tests/test_imports.py @@ -0,0 +1,24 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved. + +"""Tests that the public Python API of kernelDB can be imported correctly.""" + +import pytest + +kerneldb = pytest.importorskip("kerneldb", reason="kernelDB C++ extension not available") + + +def test_package_metadata(): + assert isinstance(kerneldb.__version__, str) and kerneldb.__version__ + assert isinstance(kerneldb.__all__, list) and kerneldb.__all__ + + +def test_public_api_accessible(): + for name in ("KernelDB", "Kernel", "Instruction", "BasicBlock", + "CDNAKernel", "ArchDescriptor", "HsaAgent", "KernelArgument"): + assert getattr(kerneldb, name, None) is not None, f"kerneldb.{name} not accessible" + + +def test_KernelDB_and_Kernel_are_classes(): + assert isinstance(kerneldb.KernelDB, type) + assert isinstance(kerneldb.Kernel, type) diff --git a/tests/test_kernels.py b/tests/test_kernels.py new file mode 100644 index 0000000..2cb73c4 --- /dev/null +++ b/tests/test_kernels.py @@ -0,0 +1,147 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved. + +""" +Tests for KernelDB kernel-level operations: discovery, source-line mapping, +instruction extraction, filtering, basic blocks, and the Kernel wrapper. + +All tests require a ROCm environment with hipcc and a GPU. +""" + +import re +import subprocess +import tempfile +from pathlib import Path + +import pytest + +from conftest import requires_rocm + +kerneldb = pytest.importorskip("kerneldb", reason="kernelDB C++ extension not available") +KernelDB = kerneldb.KernelDB + +_HIP_SOURCE = r""" +#include + +__global__ void vector_add(float* a, float* b, float* c, int n) { + int idx = blockIdx.x * blockDim.x + threadIdx.x; + if (idx < n) { + c[idx] = a[idx] + b[idx]; + } +} + +__global__ void vector_multiply(float* a, float* b, float* c, int n) { + int idx = blockIdx.x * blockDim.x + threadIdx.x; + if (idx < n) { + c[idx] = a[idx] * b[idx]; + } +} + +int main() { return 0; } +""" + +_binary_path = None + + +def _get_binary(): + global _binary_path + if _binary_path is not None: + return _binary_path + tmp = Path(tempfile.mkdtemp(prefix="kerneldb_test_")) + src = tmp / "simple.cpp" + exe = tmp / "simple" + src.write_text(_HIP_SOURCE) + r = subprocess.run(["hipcc", "-g", str(src), "-o", str(exe)], + capture_output=True, text=True) + if r.returncode != 0: + pytest.skip(f"hipcc compilation failed:\n{r.stderr}") + _binary_path = str(exe) + return _binary_path + + +def _find_kernel(kernels, fragment): + matches = [k for k in kernels if fragment in k] + if not matches: + pytest.skip(f"No kernel containing {fragment!r} found") + return matches[0] + + +@requires_rocm +def test_kernel_discovery(): + kdb = KernelDB(_get_binary()) + kernels = kdb.get_kernels() + + assert isinstance(kernels, list) + assert len(kernels) == 2 + assert all(isinstance(k, str) and k for k in kernels) + assert any("vector_add" in k for k in kernels) + assert any("vector_multiply" in k for k in kernels) + + +@requires_rocm +def test_kernel_lines(): + kdb = KernelDB(_get_binary()) + kernels = kdb.get_kernels() + + for name in kernels: + lines = kdb.get_kernel_lines(name) + assert isinstance(lines, list) and lines, f"No lines for {name}" + assert all(isinstance(ln, int) and ln >= 0 for ln in lines) + + add = _find_kernel(kernels, "vector_add") + mul = _find_kernel(kernels, "vector_multiply") + assert set(kdb.get_kernel_lines(add)) != set(kdb.get_kernel_lines(mul)) + + +@requires_rocm +def test_instructions(): + kdb = KernelDB(_get_binary()) + name = _find_kernel(kdb.get_kernels(), "vector_add") + + for line in kdb.get_kernel_lines(name): + for inst in kdb.get_instructions_for_line(name, line): + assert isinstance(inst.disassembly, str) and inst.disassembly + assert isinstance(inst.line, int) and inst.line == line + assert isinstance(inst.column, int) + assert hasattr(inst, "file_name") + + +@requires_rocm +def test_instruction_filtering(): + kdb = KernelDB(_get_binary()) + name = _find_kernel(kdb.get_kernels(), "vector_add") + pattern = ".*(load|store).*" + + all_insts, mem_ops = [], [] + for line in kdb.get_kernel_lines(name): + all_insts.extend(kdb.get_instructions_for_line(name, line)) + mem_ops.extend(kdb.get_instructions_for_line(name, line, pattern)) + + assert len(mem_ops) > 0, "vector_add must have at least one load/store" + assert len(mem_ops) <= len(all_insts) + assert all(re.search(pattern, i.disassembly, re.IGNORECASE) for i in mem_ops) + + for line in kdb.get_kernel_lines(name): + assert kdb.get_instructions_for_line(name, line, "NOMATCH__xyz") == [] + + +@requires_rocm +def test_basic_blocks_and_kernel_wrapper(): + kdb = KernelDB(_get_binary()) + name = _find_kernel(kdb.get_kernels(), "vector_add") + kernel = kdb.get_kernel(name) + + assert isinstance(kernel.get_basic_blocks(), list) and kernel.get_basic_blocks() + assert kernel.name == name + assert isinstance(kernel.signature, str) and kernel.signature + assert isinstance(kernel.lines, list) and all(ln >= 0 for ln in kernel.lines) + assert isinstance(kernel.assembly, list) and all(isinstance(s, str) for s in kernel.assembly) + assert isinstance(kernel.files, list) and all(isinstance(f, str) for f in kernel.files) + + first_line = kernel.lines[0] + insts = kernel.get_instructions_for_line(first_line) + assert isinstance(insts, list) and insts + filtered = kernel.get_instructions_for_line(first_line, ".*load.*") + assert len(filtered) <= len(insts) + + assert isinstance(kdb.get_file_name(name, 0), str)