test(rocprofiler-compute): Move tests under tests/unit and tests/integration

projects/rocprofiler-compute/tests/common.py

@@ -11,9 +11,6 @@
 from threading import Thread
 from unittest.mock import Mock
 
-import pandas as pd
-import pytest
-
 ROOT = os.path.dirname(os.path.dirname(__file__))
 src_candidate = os.path.join(ROOT, "src")
 SRC = src_candidate if os.path.isdir(src_candidate) else ROOT
@@ -99,41 +96,6 @@ def get_output_dir(suffix="_output", clean_existing=True, param_id=None):
     return output_dir
 
 
-def setup_workload_dir(input_dir, suffix="_tmp", clean_existing=True, param_id=None):
-    """Provides a unique input workload directory with contents of input_dir
-    based on the name of the calling test function. For parametrized tests,
-    pass param_id to ensure unique directory names and avoid NFS conflicts.
-
-    Creates a copy to avoid modifying source workload data.
-
-    Args:
-        input_dir (str): Source directory to copy from.
-        suffix (str, optional): suffix to append to output_dir.
-            Defaults to "_tmp".
-        clean_existing (bool, optional): Whether to remove existing directory if exists.
-            Defaults to True.
-        param_id (str, optional): Unique identifier for parametrized tests.
-            When provided, appended to the directory name to ensure uniqueness.
-            Defaults to None.
-    """
-
-    func_name = inspect.stack()[1].function
-
-    # Include param_id in directory name if provided
-    param_suffix = ""
-    if param_id:
-        # Sanitize param_id: replace special chars that may not be valid in paths
-        param_suffix = "_" + re.sub(r"[^\w\-]", "_", str(param_id))
-
-    output_dir = func_name + param_suffix + suffix
-    if clean_existing:
-        if Path(output_dir).exists():
-            shutil.rmtree(output_dir)
-
-    shutil.copytree(input_dir, output_dir)
-    return output_dir
-
-
 def clean_output_dir(cleanup, output_dir):
     """Remove output directory generated from rocprofiler-compute execution
 
@@ -152,100 +114,6 @@ def clean_output_dir(cleanup, output_dir):
     return
 
 
-def check_csv_files(output_dir, num_devices, num_kernels):
-    """Check profiling output csv files for expected
-    number of entries (based on kernel invocations)
-
-    Args:
-        output_dir (string): output directory containing csv files
-        num_kernels (int): number of kernels expected to have been profiled
-
-    Returns:
-        dict: dictionary housing file contents as pandas dataframe
-              (excludes PMC files - those are validated internally)
-    """
-    files_in_workload = os.listdir(output_dir)
-
-    # results_*.csv is written compressed, so accept either form. read_csv
-    # infers gzip from the .gz suffix.
-    def is_csv(name):
-        return name.endswith(".csv") or name.endswith(".csv.gz")
-
-    # Validate PMC data exists (profile creates pmc_perf_*.csv or results_*.csv)
-    has_separate = any(
-        f.startswith("pmc_perf_") and is_csv(f) for f in files_in_workload
-    )
-    has_results = any(f.startswith("results_") and is_csv(f) for f in files_in_workload)
-
-    assert has_separate or has_results, (
-        "Expected pmc_perf_*.csv or results_*.csv from profile mode"
-    )
-
-    # Validate row counts for PMC files (but don't add to return dict)
-    for file in files_in_workload:
-        is_pmc = file.startswith("pmc_perf_") or file.startswith("results_")
-        if is_pmc and is_csv(file):
-            df = pd.read_csv(output_dir + "/" + file)
-            err_msg = (
-                f"PMC file {file} has insufficient rows: "
-                f"{len(df.index)} < {num_kernels}"
-            )
-            assert len(df.index) >= num_kernels, err_msg
-
-    # Check and return non-PMC files
-    return check_non_pmc_files(output_dir, num_devices, num_kernels)
-
-
-def check_non_pmc_files(output_dir, num_devices, num_kernels):
-    """
-    Check profiling output non-PMC files and return them as a dictionary.
-
-    Args:
-        output_dir (string): output directory containing non-PMC files
-        num_devices (int): number of devices expected to have been profiled
-        num_kernels (int): number of kernels expected to have been profiled
-
-    Returns:
-        dict: dictionary housing file contents as pandas dataframe
-    """
-    file_dict = {}
-    files_in_workload = os.listdir(output_dir)
-
-    # Load non-PMC files into return dict
-    for file in files_in_workload:
-        # Skip PMC files (already validated above), compressed or not
-        if file.startswith("pmc_perf_") or file.startswith("results_"):
-            continue
-
-        if file.endswith(".csv"):
-            # Load other CSV files
-            file_dict[file] = pd.read_csv(output_dir + "/" + file)
-            if "roofline" in file:
-                assert len(file_dict[file].index) >= num_devices
-            elif "sysinfo" not in file and "ps_file" not in file:
-                assert len(file_dict[file].index) >= num_kernels
-        elif file.endswith(".html"):
-            file_dict[file] = "html"
-        elif file.endswith(".json"):
-            file_dict[file] = "json"
-
-    return file_dict
-
-
-def get_num_pmc_file(output_dir):
-    """
-    Returns:
-        int: number of pmc perf yaml files in perfmon dir
-    """
-
-    perfmon_path = Path(output_dir) / "perfmon"
-    return len([
-        f
-        for f in perfmon_path.iterdir()
-        if f.is_file() and f.name.startswith("pmc_perf_") and f.suffix == ".yaml"
-    ])
-
-
 def read_binary_file_tree(root: Path) -> dict[Path, bytes]:
     """Return binary contents keyed by each file's path relative to root."""
     return {
@@ -330,53 +198,3 @@ def patch_console(monkeypatch, module, *names, **overrides):
         monkeypatch.setattr(f"{module}.console_{name}", mock)
         mocks[name] = mock
     return mocks
-
-
-def gpu_soc():
-    """Return (arch, model) from rocminfo, e.g. ('gfx942', 'MI300').
-
-    Both are '' when no supported GPU is detected.
-    """
-    # decode with utf-8 to account for rocm-smi changes in latest rocm
-    rocminfo = (
-        subprocess
-        .run(["rocminfo"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        .stdout.decode("utf-8")
-        .split("\n")
-    )
-    soc_regex = re.compile(r"^\s*Name\s*:\s+ ([a-zA-Z0-9]+)\s*$", re.MULTILINE)
-    devices = list(filter(soc_regex.match, rocminfo))
-    if not devices:
-        return "", ""
-    arch = devices[0].split()[1]
-    if arch not in SUPPORTED_ARCHS:
-        return "", ""
-    model = list(SUPPORTED_ARCHS[arch].keys())[0].upper()
-    return arch, model
-
-
-def skip_unsupported_pc_sampling_soc(is_stochastic=False):
-    """Skip PC-sampling tests on SoCs that do not support the selected mode."""
-    _, soc = gpu_soc()
-
-    unsupported_socs = {
-        "MI100",
-        "RDNA35_POINT_1",
-        "RDNA35_HALO",
-        "RDNA35_POINT_2",
-        "RDNA35_GORGON_POINT",
-    }
-    if is_stochastic:
-        unsupported_socs.add("MI200")
-
-    if soc in unsupported_socs:
-        pytest.skip(f"PC sampling is not supported on {soc}")
-
-
-def require_pc_sampling_gpu(is_stochastic=False):
-    """Skip the test unless a GPU that supports the selected PC sampling mode is
-    present."""
-    _, soc = gpu_soc()
-    if not soc:
-        pytest.skip("GPU not supported")
-    skip_unsupported_pc_sampling_soc(is_stochastic=is_stochastic)