Mini-Nav/tests/test_sweep_cam_perf.py

#!/usr/bin/env python3
"""Fast unit tests for scripts/sweep_cam_perf.py (sweep helpers)."""

from __future__ import annotations

import subprocess
from pathlib import Path

from scripts.sweep_cam_perf import (
    CSV_FIELDNAMES,
    PerfRun,
    SweepConfig,
    _run_subprocess,
    build_make_command,
    default_sweep_configs,
    logical_sweeps_for_config,
    parse_perf_result,
    plot_figures,
    run_all,
)


# ---------------------------------------------------------------------------
# parse_perf_result
# ---------------------------------------------------------------------------


def test_parse_perf_result_extracts_metrics():
    log = (
        "PERF_RESULT latency_cycles=148 total_cycles=158 "
        "accepted_queries=1 completed_queries=1 "
        "queries_per_cycle=0.006329 status=pass"
    )
    result = parse_perf_result(log)
    assert result == {
        "latency_cycles": "148",
        "total_cycles": "158",
        "accepted_queries": "1",
        "completed_queries": "1",
        "queries_per_cycle": "0.006329",
        "status": "pass",
    }


def test_parse_perf_result_returns_empty_when_missing():
    log = "some random simulation output without the marker"
    assert parse_perf_result(log) == {}


# ---------------------------------------------------------------------------
# default_sweep_configs
# ---------------------------------------------------------------------------


def test_default_sweep_configs_are_paper_supporting_dense_matrix():
    configs = default_sweep_configs()
    assert len(configs) == 19

    tuples = {(c.num_rows, c.hash_bits, c.lanes) for c in configs}

    # NUM_ROWS sweep (lanes fixed at 16)
    for nr in (64, 128, 192, 256, 384, 512, 768, 1024):
        assert (nr, 512, 16) in tuples, f"Missing NUM_ROWS config {(nr, 512, 16)}"

    # HASH_BITS sweep (num_rows fixed at 512, lanes fixed at 16)
    for hb in (64, 128, 192, 256, 320, 384, 448, 512):
        assert (512, hb, 16) in tuples, f"Missing HASH_BITS config {(512, hb, 16)}"

    # LANES sweep (num_rows and hash_bits fixed at 512)
    for la in (2, 4, 8, 16, 32):
        assert (512, 512, la) in tuples, f"Missing LANES config {(512, 512, la)}"

    assert len(tuples) == 19
    # All configs must satisfy num_rows % lanes == 0
    for c in configs:
        assert c.num_rows % c.lanes == 0, (
            f"num_rows={c.num_rows} not divisible by lanes={c.lanes} for {c}"
        )


# ---------------------------------------------------------------------------
# logical_sweeps_for_config
# ---------------------------------------------------------------------------


def test_logical_sweeps_baseline_belongs_to_all_three_groups():
    """The baseline (512,512,16) config belongs to all three sweep groups,
    even though its CSV sweep_name is 'num_rows' (first-occurrence dedup)."""
    configs = default_sweep_configs()
    # Locate the shared baseline config.
    baseline = next(c for c in configs if (c.num_rows, c.hash_bits, c.lanes) == (512, 512, 16))
    groups = logical_sweeps_for_config(baseline)
    assert set(groups) == {"num_rows", "hash_bits", "lanes"}, (
        f"Baseline should belong to all three groups, got {groups}"
    )
    assert len(configs) == 19, "Sweep config count must remain 19"


# ---------------------------------------------------------------------------
# build_make_command
# ---------------------------------------------------------------------------


def test_build_make_command_uses_perf_test_module():
    cfg = SweepConfig("smoke", 128, 128, 4)
    cmd = build_make_command(cfg)
    assert cmd == [
        "make",
        "-C",
        "hw/sim",
        "NUM_ROWS=128",
        "HASH_BITS=128",
        "LANES=4",
        "COCOTB_TEST_MODULES=tests.test_cam_perf",
    ]


# ---------------------------------------------------------------------------
# CSV_FIELDNAMES
# ---------------------------------------------------------------------------


def test_csv_fieldnames_match_spec():
    assert CSV_FIELDNAMES == [
        "sweep_name",
        "num_rows",
        "hash_bits",
        "lanes",
        "latency_cycles",
        "total_cycles",
        "accepted_queries",
        "completed_queries",
        "queries_per_cycle",
        "sim_time_sec",
        "status",
        "log_path",
    ]


# ---------------------------------------------------------------------------
# _run_subprocess — timeout preserves partial output
# ---------------------------------------------------------------------------


def test_run_subprocess_timeout_preserves_partial_output(monkeypatch):
    """Timeout error string includes partial stdout/stderr from the process."""

    def mock_run(*args, **kwargs):
        raise subprocess.TimeoutExpired(
            cmd=["mock"],
            timeout=1,
            output="partial stdout data",
            stderr="partial stderr data",
        )

    monkeypatch.setattr(subprocess, "run", mock_run)

    result, duration, error = _run_subprocess(
        ["mock"], cwd=Path("/tmp"), env={}, timeout=1,
    )

    assert result is None
    assert duration >= 0
    assert "timed out after 1s" in error
    assert "[partial stdout before timeout]" in error
    assert "partial stdout data" in error
    assert "[partial stderr before timeout]" in error
    assert "partial stderr data" in error


def test_run_subprocess_timeout_without_partial_output(monkeypatch):
    """Timeout error is clean when the subprocess produced no output."""

    def mock_run(*args, **kwargs):
        raise subprocess.TimeoutExpired(cmd=["mock"], timeout=5)

    monkeypatch.setattr(subprocess, "run", mock_run)

    result, duration, error = _run_subprocess(
        ["mock"], cwd=Path("/tmp"), env={}, timeout=5,
    )

    assert result is None
    assert "timed out after 5s" in error
    assert "[partial stdout" not in error


# ---------------------------------------------------------------------------
# run_all — output_root anchoring
# ---------------------------------------------------------------------------


def _make_mock_run_one_config(status="pass"):
    """Return a ``run_one_config`` stand-in that never invokes a subprocess."""

    def mock(config, *, output_dir, repo_root, timeout=1800):
        return PerfRun(
            config=config,
            status=status,
            returncode=0 if status == "pass" else 1,
            duration_sec=0.0,
            log_path=Path("logs/test.log"),
            metrics={"status": status},
        )

    return mock


def test_run_all_anchors_relative_output_root_to_repo_root(tmp_path, monkeypatch):
    """A relative ``output_root`` is resolved under ``repo_root``."""
    monkeypatch.setattr(
        "scripts.sweep_cam_perf.run_one_config",
        _make_mock_run_one_config("pass"),
    )

    configs = [SweepConfig("test", 512, 512, 16)]
    exit_code = run_all(
        configs,
        output_root="outputs/cam_perf",
        date="2026-06-01",
        repo_root=tmp_path,
    )

    expected_csv = tmp_path / "outputs" / "cam_perf" / "2026-06-01" / "sweep.csv"
    assert expected_csv.exists(), f"sweep.csv not found at {expected_csv}"
    assert exit_code == 0


def test_run_all_uses_absolute_output_root_unmodified(tmp_path, monkeypatch):
    """An absolute ``output_root`` is used as-is, not anchored under repo_root."""
    monkeypatch.setattr(
        "scripts.sweep_cam_perf.run_one_config",
        _make_mock_run_one_config("pass"),
    )

    configs = [SweepConfig("test", 512, 512, 16)]
    abs_output = tmp_path / "custom_dir"
    repo_root = Path("/nonexistent/repo")

    exit_code = run_all(
        configs,
        output_root=abs_output,
        date="2026-06-01",
        repo_root=repo_root,
    )

    expected_csv = abs_output / "2026-06-01" / "sweep.csv"
    assert expected_csv.exists(), f"sweep.csv not found at {expected_csv}"
    assert exit_code == 0


# ---------------------------------------------------------------------------
# plot_figures — required PNG outputs
# ---------------------------------------------------------------------------


def test_plot_figures_creates_required_pngs(tmp_path):
    """plot_figures creates four PNGs under figures/ when enough pass runs exist
    across all three logical sweep groups.  No exception with Agg backend."""
    configs = [
        SweepConfig("num_rows", 128, 512, 16),
        SweepConfig("num_rows", 256, 512, 16),
        SweepConfig("num_rows", 512, 512, 16),   # baseline: all three groups
        SweepConfig("hash_bits", 512, 128, 16),
        SweepConfig("hash_bits", 512, 256, 16),
        SweepConfig("lanes", 512, 512, 4),
        SweepConfig("lanes", 512, 512, 8),
    ]
    runs = [
        PerfRun(
            config=cfg,
            status="pass",
            returncode=0,
            duration_sec=1.0,
            log_path=Path("logs/test.log"),
            metrics={
                "latency_cycles": str(50 + cfg.num_rows % 100),
                "total_cycles": str(100 + cfg.num_rows % 100),
                "accepted_queries": "10",
                "completed_queries": "10",
                "queries_per_cycle": "0.05",
                "status": "pass",
            },
        )
        for cfg in configs
    ]

    plot_figures(tmp_path, runs)

    fig_dir = tmp_path / "figures"
    assert fig_dir.is_dir()
    assert (fig_dir / "cam_num_rows_vs_latency.png").exists()
    assert (fig_dir / "cam_hash_bits_vs_latency.png").exists()
    assert (fig_dir / "cam_lanes_vs_throughput.png").exists()
    assert (fig_dir / "cam_perf_multipanel.png").exists()