From 24b8750f0fb553f089c8605c1a36fae50a508922 Mon Sep 17 00:00:00 2001 From: SikongJueluo Date: Fri, 15 May 2026 14:01:24 +0800 Subject: [PATCH] feat(sim): add CAM correctness runner with scenario matrix and contract tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce scripts/run_cam_correctness.py — a data-driven CAM correctness checker that runs cocotb and pytest scenarios across multiple noise modes and CAM submodules (cam_top, cam_core_banked, match_engine_pipeline). Outputs structured CSV results and a paper-ready Chinese summary. Add tests/test_run_cam_correctness.py with contract tests for scenario spec validation, command building, subprocess error handling, output file generation, CLI defaults, and environment preservation. --- scripts/run_cam_correctness.py | 642 ++++++++++++++++++++++++++++++ tests/test_run_cam_correctness.py | 585 +++++++++++++++++++++++++++ 2 files changed, 1227 insertions(+) create mode 100644 scripts/run_cam_correctness.py create mode 100644 tests/test_run_cam_correctness.py diff --git a/scripts/run_cam_correctness.py b/scripts/run_cam_correctness.py new file mode 100644 index 0000000..1ef913c --- /dev/null +++ b/scripts/run_cam_correctness.py @@ -0,0 +1,642 @@ +#!/usr/bin/env python3 +"""CAM correctness runner — data model, scenario matrix, and command builders.""" + +import csv +import os +import shutil +import subprocess +import time +from dataclasses import dataclass +from datetime import date as date_type +from datetime import datetime +from pathlib import Path +from typing import Annotated, Literal + +import typer + +# Timeout for each external command invocation (30 minutes). +DEFAULT_COMMAND_TIMEOUT_SEC = 1800 + +# --------------------------------------------------------------------------- +# Types +# --------------------------------------------------------------------------- + +ScenarioKind = Literal["cocotb", "pytest"] +Status = Literal["pass", "fail"] + + +# --------------------------------------------------------------------------- +# Data model +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Scenario: + """A single correctness-check scenario.""" + + name: str + kind: ScenarioKind + module: str | None + testcase: str | None + params: dict[str, str] + noise_mode: str + expected: str + + +@dataclass(frozen=True) +class ScenarioRun: + """Result of executing a single scenario.""" + + scenario: Scenario + status: Status + returncode: int + duration_sec: float + command: list[str] + log_path: Path + results_xml: str + + +# --------------------------------------------------------------------------- +# Default scenario matrix +# --------------------------------------------------------------------------- + +default_scenarios: list[Scenario] = [ + Scenario( + name="cam_top_basic", + kind="cocotb", + module="tests.test_cam_basic", + testcase=None, + params={}, + noise_mode="none", + expected="match", + ), + Scenario( + name="cam_top_write_noise", + kind="cocotb", + module="tests.test_cam_basic", + testcase=None, + params={ + "WRITE_NOISE_EN": "1", + "WRITE_NOISE_RATE_NUM": "1", + "WRITE_NOISE_RATE_DEN": "100", + "READ_NOISE_EN": "0", + }, + noise_mode="write_noise", + expected="bit_flip_injected", + ), + Scenario( + name="cam_top_read_noise", + kind="cocotb", + module="tests.test_cam_basic", + testcase=None, + params={ + "WRITE_NOISE_EN": "0", + "READ_NOISE_EN": "1", + "READ_NOISE_RATE_NUM": "1", + "READ_NOISE_RATE_DEN": "100", + }, + noise_mode="read_noise", + expected="read_perturbed", + ), + Scenario( + name="cam_core_banked", + kind="cocotb", + module="tests.test_cam_core_banked", + testcase=None, + params={}, + noise_mode="none", + expected="read_write_aligned", + ), + Scenario( + name="match_engine_pipeline", + kind="cocotb", + module="tests.test_match_engine_pipeline", + testcase=None, + params={}, + noise_mode="none", + expected="top1_hamming_correct", + ), + Scenario( + name="cam_write_noise", + kind="cocotb", + module="tests.test_cam_write_noise", + testcase=None, + params={}, + noise_mode="write_noise", + expected="bit_flip_injected", + ), + Scenario( + name="cam_read_noise", + kind="cocotb", + module="tests.test_cam_read_noise", + testcase=None, + params={}, + noise_mode="read_noise", + expected="read_perturbed", + ), + Scenario( + name="ref_model_noise", + kind="pytest", + module=None, + testcase=None, + params={}, + noise_mode="none", + expected="reference_noise_model_pass", + ), +] + + +# --------------------------------------------------------------------------- +# Command builders +# --------------------------------------------------------------------------- + +# Internal: map unit scenario names to their Verilog TOPLEVEL module. +# cam_top_* scenarios use the Makefile default (cam_top) and are omitted. +_UNIT_TOPLEVEL: dict[str, str] = { + "cam_core_banked": "cam_core_banked", + "match_engine_pipeline": "match_engine_pipeline", + "cam_write_noise": "cam_write_noise", + "cam_read_noise": "cam_read_noise", +} + + +def _infer_toplevel(scenario: Scenario) -> str | None: + """Return the Verilog TOPLEVEL for *scenario* or None to use Makefile default.""" + return _UNIT_TOPLEVEL.get(scenario.name) + + +def build_scenario_command( + scenario: Scenario, + *, + num_rows: int = 512, + hash_bits: int = 512, + lanes: int = 8, +) -> list[str]: + """Build the shell command to execute *scenario*.""" + if scenario.kind == "pytest": + return [ + "uv", + "run", + "pytest", + "hw/sim/tests/test_ref_model_noise.py", + "-q", + ] + + # cocotb + cmd = [ + "make", + "-C", + "hw/sim", + f"NUM_ROWS={num_rows}", + f"HASH_BITS={hash_bits}", + f"LANES={lanes}", + f"COCOTB_TEST_MODULES={scenario.module}", + ] + if scenario.testcase is not None: + cmd.append(f"COCOTB_TESTCASE={scenario.testcase}") + for key, value in scenario.params.items(): + cmd.append(f"{key}={value}") + toplevel = _infer_toplevel(scenario) + if toplevel is not None: + cmd.append(f"TOPLEVEL={toplevel}") + return cmd + + +def build_clean_command(scenario: Scenario) -> list[str] | None: + """Build a clean command or None if cleanup is not needed.""" + if scenario.kind == "pytest": + return None + return ["make", "-C", "hw/sim", "clean"] + + +# --------------------------------------------------------------------------- +# Execution helpers +# --------------------------------------------------------------------------- + + +def _run_subprocess( + cmd: list[str], + *, + cwd: Path, + env: dict[str, str], + timeout: int, +) -> tuple[subprocess.CompletedProcess | None, float, str | None]: + """Execute *cmd*, capture output, enforce *timeout*. + + Returns ``(result, duration_sec, error_msg)``. + On success *result* is a ``CompletedProcess`` and *error_msg* is ``None``. + On failure (timeout / OSError / SubprocessError) *result* is ``None`` + and *error_msg* describes what happened. + """ + start = time.monotonic() + try: + result = subprocess.run( + cmd, + cwd=cwd, + env=env, + text=True, + capture_output=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired: + end = time.monotonic() + return None, round(end - start, 3), f"timed out after {timeout}s" + except (OSError, subprocess.SubprocessError) as exc: + end = time.monotonic() + return None, round(end - start, 3), str(exc) + end = time.monotonic() + return result, round(end - start, 3), None + + +def run_one_scenario( + scenario: Scenario, + *, + num_rows: int = 512, + hash_bits: int = 512, + lanes: int = 8, + output_dir: Path, + repo_root: Path, +) -> ScenarioRun: + """Execute a single scenario and return the result.""" + logs_dir = output_dir / "logs" + logs_dir.mkdir(parents=True, exist_ok=True) + log_path = logs_dir / f"{scenario.name}.log" + + command = build_scenario_command( + scenario, num_rows=num_rows, hash_bits=hash_bits, lanes=lanes + ) + clean_command = build_clean_command(scenario) + + env = os.environ.copy() + if scenario.kind == "pytest": + hw_sim = str(repo_root / "hw" / "sim") + existing = env.get("PYTHONPATH", "") + env["PYTHONPATH"] = ( + hw_sim + (os.pathsep + existing) if existing else hw_sim + ) + + log_chunks: list[str] = [] + + # Run clean step if applicable + if clean_command is not None: + log_chunks.append(f"# Command: {' '.join(clean_command)}\n") + clean_result, clean_duration, clean_err = _run_subprocess( + clean_command, + cwd=repo_root, + env=env, + timeout=DEFAULT_COMMAND_TIMEOUT_SEC, + ) + if clean_result is not None: + log_chunks.append(clean_result.stdout) + log_chunks.append(clean_result.stderr) + else: + log_chunks.append(f"# ERROR: {clean_err}\n") + + if clean_err is not None or ( + clean_result is not None and clean_result.returncode != 0 + ): + returncode = clean_result.returncode if clean_result is not None else -1 + log_path.write_text("".join(log_chunks)) + return ScenarioRun( + scenario=scenario, + status="fail", + returncode=returncode, + duration_sec=clean_duration, + command=clean_command, + log_path=Path("logs") / f"{scenario.name}.log", + results_xml="", + ) + + # Run main command + log_chunks.append(f"# Command: {' '.join(command)}\n") + main_result, duration, main_err = _run_subprocess( + command, + cwd=repo_root, + env=env, + timeout=DEFAULT_COMMAND_TIMEOUT_SEC, + ) + + if main_result is not None: + log_chunks.append(main_result.stdout) + log_chunks.append(main_result.stderr) + else: + log_chunks.append(f"# ERROR: {main_err}\n") + + if main_err is not None: + status: Status = "fail" + returncode = -1 + else: + status = "pass" if main_result.returncode == 0 else "fail" + returncode = main_result.returncode + + log_path.write_text("".join(log_chunks)) + + # Stabilise cocotb results.xml when the file was produced + if scenario.kind == "cocotb": + src_xml = repo_root / "hw" / "sim" / "results.xml" + if src_xml.exists(): + dst_xml = logs_dir / f"{scenario.name}.results.xml" + shutil.copy2(str(src_xml), str(dst_xml)) + results_xml = f"logs/{scenario.name}.results.xml" + else: + results_xml = "" + else: + results_xml = "" + + return ScenarioRun( + scenario=scenario, + status=status, + returncode=returncode, + duration_sec=duration, + command=command, + log_path=Path("logs") / f"{scenario.name}.log", + results_xml=results_xml, + ) + + +def result_rows( + runs: list[ScenarioRun], + *, + num_rows: int = 512, + hash_bits: int = 512, + lanes: int = 8, +) -> list[dict[str, str]]: + """Convert runs to CSV-ready dict rows, inserting derived rows.""" + rows: list[dict[str, str]] = [] + for run in runs: + actual = run.scenario.expected if run.status == "pass" else "failed" + row: dict[str, str] = { + "test_name": run.scenario.name, + "status": run.status, + "num_rows": str(num_rows), + "hash_bits": str(hash_bits), + "lanes": str(lanes), + "noise_mode": run.scenario.noise_mode, + "expected": run.scenario.expected, + "actual": actual, + "log_path": str(run.log_path), + } + rows.append(row) + + # Insert derived popcount_pipeline_indirect row after match_engine_pipeline + if run.scenario.name == "match_engine_pipeline": + rows.append({ + "test_name": "popcount_pipeline_indirect", + "status": run.status, + "num_rows": str(num_rows), + "hash_bits": str(hash_bits), + "lanes": str(lanes), + "noise_mode": "none", + "expected": "hamming_score_correct", + "actual": "hamming_score_correct" if run.status == "pass" else "failed", + "log_path": str(run.log_path), + }) + return rows + + +def write_results_csv(output_dir: Path, rows: list[dict[str, str]]) -> None: + """Write results CSV with fixed header.""" + fieldnames = [ + "test_name", + "status", + "num_rows", + "hash_bits", + "lanes", + "noise_mode", + "expected", + "actual", + "log_path", + ] + path = output_dir / "results.csv" + with open(path, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(rows) + + +def write_raw_results_csv(output_dir: Path, runs: list[ScenarioRun]) -> None: + """Write raw results CSV with fixed header.""" + fieldnames = [ + "scenario", + "status", + "returncode", + "duration_sec", + "command", + "log_path", + "results_xml", + ] + path = output_dir / "raw_results.csv" + with open(path, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + for run in runs: + writer.writerow({ + "scenario": run.scenario.name, + "status": run.status, + "returncode": run.returncode, + "duration_sec": run.duration_sec, + "command": " ".join(run.command), + "log_path": str(run.log_path), + "results_xml": run.results_xml, + }) + + +def write_summary_md( + output_dir: Path, + rows: list[dict[str, str]], + *, + run_datetime: str, + num_rows: int = 512, + hash_bits: int = 512, + lanes: int = 8, +) -> None: + """Write a Chinese summary markdown file with scenario table, failure list, + coverage notes, and a paper-friendly conclusion draft.""" + total = len(rows) + failed = sum(1 for r in rows if r["status"] == "fail") + passed = total - failed + failed_rows = [r for r in rows if r["status"] == "fail"] + + lines: list[str] = [] + lines.append("# CAM 功能正确性验证摘要") + lines.append("") + lines.append(f"**运行日期：** {run_datetime}") + lines.append("") + lines.append("**参数：**") + lines.append(f"- NUM_ROWS={num_rows}") + lines.append(f"- HASH_BITS={hash_bits}") + lines.append(f"- LANES={lanes}") + lines.append("") + lines.append(f"**输出目录：** {output_dir}") + lines.append("") + lines.append(f"**总场景数：** {total}") + lines.append(f"**通过数：** {passed}") + lines.append(f"**失败数：** {failed}") + lines.append("") + lines.append("## 场景结果") + lines.append("") + lines.append("| 场景 | 状态 | 噪声模式 | 预期 | 实际 | 日志 |") + lines.append("|------|------|----------|------|------|------|") + for r in rows: + lines.append( + f"| {r['test_name']} | {r['status']} | {r['noise_mode']} " + f"| {r['expected']} | {r['actual']} | {r['log_path']} |" + ) + lines.append("") + lines.append("## 失败场景") + lines.append("") + if failed_rows: + for r in failed_rows: + lines.append(f"- **{r['test_name']}**: 日志路径 `{r['log_path']}`") + else: + lines.append("- 无") + lines.append("") + lines.append("## 覆盖说明") + lines.append("") + lines.append( + "- **cam_top**: 通过 `cam_top_basic`、`cam_top_write_noise`、" + "`cam_top_read_noise` 三个场景覆盖顶层 CAM 模块的基本功能、" + "写入噪声注入和读取噪声注入路径" + ) + lines.append( + "- **cam_core_banked**: 通过 `cam_core_banked` 场景覆盖" + "分块 CAM 核心的读写对齐验证" + ) + lines.append( + "- **match_engine_pipeline**: 通过 `match_engine_pipeline`" + " 场景覆盖流水线匹配引擎的 Top-1 汉明距离计算" + ) + lines.append( + "- **popcount_pipeline**: 通过 `match_engine_pipeline`" + " 的汉明匹配路径间接覆盖 popcount 流水线" + ) + lines.append( + "- **cam_write_noise**: 通过 `cam_write_noise` 场景独立覆盖" + "写入噪声注入路径" + ) + lines.append( + "- **cam_read_noise**: 通过 `cam_read_noise` 场景独立覆盖" + "读取噪声注入路径" + ) + lines.append("") + lines.append("## 论文可引用结论草稿") + lines.append("") + lines.append( + f"本测试套件在 {run_datetime} 对 CAM 设计进行了全面的功能正确性验证。" + f"测试参数为 NUM_ROWS={num_rows}、HASH_BITS={hash_bits}、" + f"LANES={lanes}。共执行 {total} 个测试场景，通过 {passed} 个，" + f"失败 {failed} 个。详细结果见 `results.csv` 和 `raw_results.csv`。" + f"日志文件位于 `logs/` 目录下，可用于深入调试与复现。" + ) + lines.append("") + + path = output_dir / "summary.md" + path.write_text("\n".join(lines)) + + +def run_all( + *, + scenarios: list[Scenario], + num_rows: int = 512, + hash_bits: int = 512, + lanes: int = 8, + output_root: str | Path = Path("outputs/cam_correctness"), + date: str | None = None, + repo_root: Path | None = None, +) -> int: + """Run all scenarios and produce all output files. + + Output is written under *output_root* / *date*. + Returns 0 if all scenario *runs* pass, otherwise 1. + """ + if repo_root is None: + repo_root = Path(__file__).resolve().parent.parent + output_root = Path(output_root) + if date is None: + date = datetime.now().strftime("%Y-%m-%d") + output_dir = output_root / date + output_dir.mkdir(parents=True, exist_ok=True) + + runs: list[ScenarioRun] = [] + for scenario in scenarios: + run = run_one_scenario( + scenario, + num_rows=num_rows, + hash_bits=hash_bits, + lanes=lanes, + output_dir=output_dir, + repo_root=repo_root, + ) + runs.append(run) + + rows = result_rows( + runs, + num_rows=num_rows, + hash_bits=hash_bits, + lanes=lanes, + ) + + write_results_csv(output_dir, rows) + write_raw_results_csv(output_dir, runs) + + run_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + write_summary_md( + output_dir, + rows, + run_datetime=run_datetime, + num_rows=num_rows, + hash_bits=hash_bits, + lanes=lanes, + ) + + all_pass = all(run.status == "pass" for run in runs) + return 0 if all_pass else 1 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def script_repo_root() -> Path: + """Return the repository root (two levels up from scripts/run_cam_correctness.py).""" + return Path(__file__).resolve().parents[1] + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +app = typer.Typer( + add_completion=False, + help=( + "Run CAM Cocotb/Verilator correctness checks " + "and collect paper-ready outputs." + ), +) + + +@app.command() +def main( + num_rows: Annotated[int, typer.Option('--num-rows', help='CAM row count passed to NUM_ROWS.', min=1)] = 512, + hash_bits: Annotated[int, typer.Option('--hash-bits', help='Hash width passed to HASH_BITS.', min=1)] = 512, + lanes: Annotated[int, typer.Option('--lanes', help='CAM lane count passed to LANES.', min=1)] = 8, + output_root: Annotated[Path, typer.Option('--output-root', help='Root directory for dated correctness outputs.')] = Path('outputs/cam_correctness'), + date: Annotated[str | None, typer.Option('--date', help='Output date directory. Defaults to today.')] = None, +) -> None: + """Run the full CAM correctness suite.""" + run_date = date or date_type.today().isoformat() + exit_code = run_all( + scenarios=default_scenarios, + num_rows=num_rows, + hash_bits=hash_bits, + lanes=lanes, + output_root=output_root, + date=run_date, + repo_root=script_repo_root(), + ) + raise typer.Exit(exit_code) + + +if __name__ == "__main__": + app() diff --git a/tests/test_run_cam_correctness.py b/tests/test_run_cam_correctness.py new file mode 100644 index 0000000..d0a030a --- /dev/null +++ b/tests/test_run_cam_correctness.py @@ -0,0 +1,585 @@ +#!/usr/bin/env python3 +"""Contract tests for scripts/run_cam_correctness.py.""" + +from __future__ import annotations + +import importlib.util +import subprocess +from pathlib import Path + +from typer.testing import CliRunner + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def load_runner(): + """Load scripts/run_cam_correctness via importlib.""" + path = str( + Path(__file__).resolve().parent.parent / "scripts" / "run_cam_correctness.py" + ) + spec = importlib.util.spec_from_file_location("run_cam_correctness", path) + assert spec is not None, f"Could not find spec for {path}" + assert spec.loader is not None, f"No loader available for {path}" + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class FakeCompletedProcess: + """Stand-in for subprocess.CompletedProcess.""" + + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class FakeSubprocess: + """Record subprocess calls; return pre-configured return codes. + + Parameters + ---------- + returncodes: + Sequence of return codes consumed in order. + exception_indices: + Call indices at which to raise ``subprocess.TimeoutExpired`` + instead of returning a result. + """ + + def __init__(self, returncodes=None, exception_indices=None): + self.returncodes = returncodes or [] + self.exception_indices = set(exception_indices or []) + self.calls = [] + self._index = 0 + + def run(self, command, cwd=None, env=None, text=None, capture_output=None, timeout=None): + self.calls.append({ + "command": command, + "cwd": cwd, + "env": dict(env or {}), + "text": text, + "capture_output": capture_output, + "timeout": timeout, + }) + idx = self._index + self._index += 1 + if idx in self.exception_indices: + raise subprocess.TimeoutExpired(cmd=command, timeout=timeout or 1800) + rc = ( + self.returncodes[idx] + if idx < len(self.returncodes) + else 0 + ) + stdout = "fake stdout\n" + stderr = "fake stderr\n" if rc != 0 else "" + return FakeCompletedProcess(returncode=rc, stdout=stdout, stderr=stderr) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestRunnerContract: + """Contract tests for scripts/run_cam_correctness.""" + + def test_default_scenarios_match_approved_spec(self): + runner = load_runner() + names = [s.name for s in runner.default_scenarios] + assert names == [ + "cam_top_basic", + "cam_top_write_noise", + "cam_top_read_noise", + "cam_core_banked", + "match_engine_pipeline", + "cam_write_noise", + "cam_read_noise", + "ref_model_noise", + ] + + sc = {s.name: s for s in runner.default_scenarios} + + # --- cam_top_basic --- + s = sc["cam_top_basic"] + assert s.name == "cam_top_basic" + assert s.kind == "cocotb" + assert s.module == "tests.test_cam_basic" + assert s.testcase is None + assert s.params == {} + assert s.noise_mode == "none" + assert s.expected == "match" + + # --- cam_top_write_noise --- + s = sc["cam_top_write_noise"] + assert s.name == "cam_top_write_noise" + assert s.kind == "cocotb" + assert s.module == "tests.test_cam_basic" + assert s.testcase is None + assert s.params == { + "WRITE_NOISE_EN": "1", + "WRITE_NOISE_RATE_NUM": "1", + "WRITE_NOISE_RATE_DEN": "100", + "READ_NOISE_EN": "0", + } + assert s.noise_mode == "write_noise" + assert s.expected == "bit_flip_injected" + + # --- cam_top_read_noise --- + s = sc["cam_top_read_noise"] + assert s.name == "cam_top_read_noise" + assert s.kind == "cocotb" + assert s.module == "tests.test_cam_basic" + assert s.testcase is None + assert s.params == { + "WRITE_NOISE_EN": "0", + "READ_NOISE_EN": "1", + "READ_NOISE_RATE_NUM": "1", + "READ_NOISE_RATE_DEN": "100", + } + assert s.noise_mode == "read_noise" + assert s.expected == "read_perturbed" + + # --- cam_core_banked --- + s = sc["cam_core_banked"] + assert s.name == "cam_core_banked" + assert s.kind == "cocotb" + assert s.module == "tests.test_cam_core_banked" + assert s.testcase is None + assert s.params == {} + assert s.noise_mode == "none" + assert s.expected == "read_write_aligned" + + # --- match_engine_pipeline --- + s = sc["match_engine_pipeline"] + assert s.name == "match_engine_pipeline" + assert s.kind == "cocotb" + assert s.module == "tests.test_match_engine_pipeline" + assert s.testcase is None + assert s.params == {} + assert s.noise_mode == "none" + assert s.expected == "top1_hamming_correct" + + # --- cam_write_noise --- + s = sc["cam_write_noise"] + assert s.name == "cam_write_noise" + assert s.kind == "cocotb" + assert s.module == "tests.test_cam_write_noise" + assert s.testcase is None + assert s.params == {} + assert s.noise_mode == "write_noise" + assert s.expected == "bit_flip_injected" + + # --- cam_read_noise --- + s = sc["cam_read_noise"] + assert s.name == "cam_read_noise" + assert s.kind == "cocotb" + assert s.module == "tests.test_cam_read_noise" + assert s.testcase is None + assert s.params == {} + assert s.noise_mode == "read_noise" + assert s.expected == "read_perturbed" + + # --- ref_model_noise --- + s = sc["ref_model_noise"] + assert s.name == "ref_model_noise" + assert s.kind == "pytest" + assert s.module is None + assert s.testcase is None + assert s.params == {} + assert s.noise_mode == "none" + assert s.expected == "reference_noise_model_pass" + + def test_build_cocotb_command_includes_make_params(self): + runner = load_runner() + scenario = runner.Scenario( + name="cam_top_write_noise", + kind="cocotb", + module="tests.test_cam_basic", + testcase="exact_noise_model_match", + params={"WRITE_NOISE_EN": "1", "WRITE_NOISE_RATE_NUM": "1"}, + noise_mode="write_noise", + expected="bit_flip_injected", + ) + cmd = runner.build_scenario_command(scenario) + assert cmd == [ + "make", + "-C", + "hw/sim", + "NUM_ROWS=512", + "HASH_BITS=512", + "LANES=8", + "COCOTB_TEST_MODULES=tests.test_cam_basic", + "COCOTB_TESTCASE=exact_noise_model_match", + "WRITE_NOISE_EN=1", + "WRITE_NOISE_RATE_NUM=1", + ] + + def test_build_cocotb_command_includes_toplevel_for_unit_scenarios(self): + """Unit-level cocotb scenarios get explicit TOPLEVEL override matching scenario name.""" + runner = load_runner() + sc = {s.name: s for s in runner.default_scenarios} + + # Unit scenarios should get TOPLEVEL override matching their scenario name + unit_names = [ + "cam_core_banked", + "match_engine_pipeline", + "cam_write_noise", + "cam_read_noise", + ] + for name in unit_names: + cmd = runner.build_scenario_command(sc[name]) + toplevel_kvp = f"TOPLEVEL={name}" + assert toplevel_kvp in cmd, ( + f"Expected {toplevel_kvp!r} in command for {name!r}, got {cmd}" + ) + + # cam_top_* scenarios should NOT get explicit TOPLEVEL (use Makefile default) + top_names = ["cam_top_basic", "cam_top_write_noise", "cam_top_read_noise"] + for name in top_names: + cmd = runner.build_scenario_command(sc[name]) + toplevel_flags = [arg for arg in cmd if arg.startswith("TOPLEVEL=")] + assert len(toplevel_flags) == 0, ( + f"Unexpected TOPLEVEL override in command for {name!r}: {cmd}" + ) + + def test_build_pytest_command_uses_reference_model_test(self): + runner = load_runner() + scenario = runner.Scenario( + name="ref_model_noise", + kind="pytest", + module="tests.test_ref_model_noise", + testcase=None, + params={}, + noise_mode="none", + expected="reference_noise_model_pass", + ) + cmd = runner.build_scenario_command(scenario) + assert cmd == [ + "uv", + "run", + "pytest", + "hw/sim/tests/test_ref_model_noise.py", + "-q", + ] + + def test_run_all_writes_outputs_and_continues_after_failure( + self, tmp_path, monkeypatch + ): + runner = load_runner() + + fake_subp = FakeSubprocess(returncodes=[0, 0, 0, 1, 0]) + monkeypatch.setattr(runner.subprocess, "run", fake_subp.run) + + time_values = iter([0.0, 1.0, 1.0, 2.5, 2.5, 3.0, 3.0, 4.0, 4.0, 5.0]) + monkeypatch.setattr(runner.time, "monotonic", lambda: next(time_values)) + + scenarios = [ + runner.default_scenarios[0], # cam_top_basic + runner.default_scenarios[4], # match_engine_pipeline + runner.default_scenarios[7], # ref_model_noise + ] + + date = "2026-05-14" + output_root = tmp_path + output_dir = output_root / date + log_dir = output_dir / "logs" + + exit_code = runner.run_all( + scenarios=scenarios, + num_rows=512, + hash_bits=512, + lanes=8, + output_root=str(tmp_path), + date=date, + ) + + assert exit_code == 1 + assert len(fake_subp.calls) == 5 + + # Exact file paths — no globs + results_csv = output_dir / "results.csv" + raw_csv = output_dir / "raw_results.csv" + summary_md = output_dir / "summary.md" + + assert results_csv.is_file(), f"Missing {results_csv}" + assert raw_csv.is_file(), f"Missing {raw_csv}" + assert summary_md.is_file(), f"Missing {summary_md}" + + # Output / log files exist + assert log_dir.is_dir() + for scenario in scenarios: + log_file = log_dir / f"{scenario.name}.log" + assert log_file.is_file(), f"Missing log: {log_file}" + assert "fake stdout\n" in log_file.read_text() + + # Results CSV content + results_text = results_csv.read_text() + assert ( + "cam_top_basic,pass,512,512,8,none,match,match,logs/cam_top_basic.log" + in results_text + ) + assert ( + "match_engine_pipeline,fail,512,512,8,none,top1_hamming_correct,failed," + "logs/match_engine_pipeline.log" in results_text + ) + assert ( + "popcount_pipeline_indirect,fail,512,512,8,none,hamming_score_correct," + "failed,logs/match_engine_pipeline.log" in results_text + ) + assert ( + "ref_model_noise,pass,512,512,8,none,reference_noise_model_pass," + "reference_noise_model_pass,logs/ref_model_noise.log" in results_text + ) + + # Raw CSV header and data + raw_text = raw_csv.read_text() + assert ( + "scenario,status,returncode,duration_sec,command,log_path,results_xml" + in raw_text + ) + assert "match_engine_pipeline,fail,1,1.0" in raw_text + + # Summary content checks + summary_text = summary_md.read_text() + assert "# CAM 功能正确性验证摘要" in summary_text + assert "**总场景数：** 4" in summary_text + assert "**失败数：** 2" in summary_text + assert "**通过数：** 2" in summary_text + assert "**运行日期：**" in summary_text + + # Config parameter lines + assert "NUM_ROWS=512" in summary_text + assert "HASH_BITS=512" in summary_text + assert "LANES=8" in summary_text + + # Output directory + assert "**输出目录：**" in summary_text + + # Scenario result table header + assert "| 场景 | 状态 | 噪声模式 | 预期 | 实际 | 日志 |" in summary_text + assert "|------|------|----------|------|------|------|" in summary_text + # Table contains at least one pass and one fail row + assert "cam_top_basic | pass | none | match | match" in summary_text + assert "match_engine_pipeline | fail | none | top1_hamming_correct | failed" in summary_text + assert "popcount_pipeline_indirect | fail | none | hamming_score_correct | failed" in summary_text + assert "ref_model_noise | pass | none | reference_noise_model_pass" in summary_text + + # Failed scenarios section — must include log path for failed rows + assert "## 失败场景" in summary_text + assert "**match_engine_pipeline**" in summary_text + assert "`logs/match_engine_pipeline.log`" in summary_text + assert "**popcount_pipeline_indirect**" in summary_text + + # Coverage section headings and key bullets + assert "## 覆盖说明" in summary_text + assert "**cam_top**" in summary_text + assert "**cam_core_banked**" in summary_text + assert "**match_engine_pipeline**" in summary_text + assert "**popcount_pipeline**" in summary_text + assert "**cam_write_noise**" in summary_text + assert "**cam_read_noise**" in summary_text + + # Conclusion section heading + assert "## 论文可引用结论草稿" in summary_text + assert "`results.csv`" in summary_text + assert "`raw_results.csv`" in summary_text + + def test_result_rows_popcount_derived_row_pass(self): + """Derived popcount_pipeline_indirect row shows hamming_score_correct when match_engine_pipeline passes.""" + runner = load_runner() + pass_run = runner.ScenarioRun( + scenario=runner.default_scenarios[4], # match_engine_pipeline + status="pass", + returncode=0, + duration_sec=1.0, + command=["make", "-C", "hw/sim"], + log_path=Path("logs/match_engine_pipeline.log"), + results_xml="", + ) + rows = runner.result_rows([pass_run]) + derived = [r for r in rows if r["test_name"] == "popcount_pipeline_indirect"] + assert len(derived) == 1 + assert derived[0]["expected"] == "hamming_score_correct" + assert derived[0]["actual"] == "hamming_score_correct" + assert derived[0]["status"] == "pass" + + def test_run_one_scenario_handles_subprocess_timeout(self, tmp_path, monkeypatch): + """Subprocess.TimeoutExpired is caught and recorded as a failure.""" + runner = load_runner() + fake_subp = FakeSubprocess(returncodes=[0], exception_indices={0}) + monkeypatch.setattr(runner.subprocess, "run", fake_subp.run) + monkeypatch.setattr(runner.time, "monotonic", lambda: 0.0) + + scenario = runner.default_scenarios[0] # cam_top_basic (cocotb) + output_dir = tmp_path / "2026-05-14" + (output_dir / "logs").mkdir(parents=True) + + run = runner.run_one_scenario( + scenario, + num_rows=512, + hash_bits=512, + lanes=8, + output_dir=output_dir, + repo_root=tmp_path, + ) + + assert run.status == "fail" + assert run.returncode == -1 + log_text = (output_dir / "logs" / "cam_top_basic.log").read_text() + assert "timed out" in log_text + assert len(fake_subp.calls) == 1 + assert fake_subp.calls[0]["timeout"] == 1800 + + def test_pytest_scenario_preserves_existing_pythonpath(self, tmp_path, monkeypatch): + """PYTHONPATH is prepended with absolute hw/sim and preserves prior value.""" + runner = load_runner() + fake_subp = FakeSubprocess(returncodes=[0]) + monkeypatch.setattr(runner.subprocess, "run", fake_subp.run) + monkeypatch.setattr(runner.time, "monotonic", lambda: 0.0) + monkeypatch.setenv("PYTHONPATH", "/existing/path") + + scenario = runner.default_scenarios[7] # ref_model_noise (pytest) + output_dir = tmp_path / "2026-05-14" + (output_dir / "logs").mkdir(parents=True) + + runner.run_one_scenario( + scenario, + num_rows=512, + hash_bits=512, + lanes=8, + output_dir=output_dir, + repo_root=tmp_path, + ) + + assert len(fake_subp.calls) == 1 + env = fake_subp.calls[0]["env"] + pp = env.get("PYTHONPATH", "") + assert str(tmp_path / "hw" / "sim") in pp + assert "/existing/path" in pp + + def test_cocotb_scenario_copies_results_xml(self, tmp_path, monkeypatch): + """Existing hw/sim/results.xml is copied to logs/.results.xml.""" + runner = load_runner() + fake_subp = FakeSubprocess(returncodes=[0]) + monkeypatch.setattr(runner.subprocess, "run", fake_subp.run) + monkeypatch.setattr(runner.time, "monotonic", lambda: 0.0) + + # Create the source XML that cocotb would have produced + src_xml = tmp_path / "hw" / "sim" / "results.xml" + src_xml.parent.mkdir(parents=True) + src_xml.write_text("") + + scenario = runner.default_scenarios[0] # cam_top_basic (cocotb) + output_dir = tmp_path / "2026-05-14" + (output_dir / "logs").mkdir(parents=True) + + run = runner.run_one_scenario( + scenario, + num_rows=512, + hash_bits=512, + lanes=8, + output_dir=output_dir, + repo_root=tmp_path, + ) + + # The source file should have been copied + dst_xml = output_dir / "logs" / "cam_top_basic.results.xml" + assert dst_xml.is_file(), f"Missing copied XML: {dst_xml}" + assert dst_xml.read_text() == "" + + # ScenarioRun points to the copied file + assert run.results_xml == "logs/cam_top_basic.results.xml" + + def test_cocotb_scenario_no_results_xml_means_empty_string(self, tmp_path, monkeypatch): + """When no hw/sim/results.xml exists, results_xml is empty string.""" + runner = load_runner() + fake_subp = FakeSubprocess(returncodes=[0]) + monkeypatch.setattr(runner.subprocess, "run", fake_subp.run) + monkeypatch.setattr(runner.time, "monotonic", lambda: 0.0) + + scenario = runner.default_scenarios[0] # cam_top_basic (cocotb) + output_dir = tmp_path / "2026-05-14" + (output_dir / "logs").mkdir(parents=True) + + run = runner.run_one_scenario( + scenario, + num_rows=512, + hash_bits=512, + lanes=8, + output_dir=output_dir, + repo_root=tmp_path, + ) + + assert run.results_xml == "" + # No .results.xml file should exist + assert not (output_dir / "logs" / "cam_top_basic.results.xml").exists() + + # ------------------------------------------------------------------ + # CLI / Typer tests + # ------------------------------------------------------------------ + + def test_cli_defaults_without_args(self, monkeypatch): + """Invoking the CLI without args uses all defaults and script-derived repo_root.""" + import datetime as dt + + runner = load_runner() + captured = {} + + def fake_run_all(**kwargs): + captured.update(kwargs) + return 0 + + monkeypatch.setattr(runner, "run_all", fake_run_all) + + # Replace the module-level date_type reference with a mock so today() is + # deterministic. We cannot monkeypatch datetime.date.today directly + # because datetime.date is an immutable C type. + class FakeDate: + @classmethod + def today(cls): + return dt.date(2026, 5, 15) + monkeypatch.setattr(runner, "date_type", FakeDate) + + cli_runner = CliRunner() + result = cli_runner.invoke(runner.app, []) + assert result.exit_code == 0, f"CLI failed: {result.output}" + + assert captured["num_rows"] == 512 + assert captured["hash_bits"] == 512 + assert captured["lanes"] == 8 + assert captured["output_root"] == Path("outputs/cam_correctness") + assert captured["date"] == "2026-05-15" + assert captured["scenarios"] == runner.default_scenarios + assert captured["repo_root"] == Path(__file__).resolve().parent.parent + + def test_typer_help_exposes_expected_options(self): + runner = load_runner() + cli_runner = CliRunner() + result = cli_runner.invoke(runner.app, ["--help"]) + assert result.exit_code == 0 + for opt in ["--num-rows", "--hash-bits", "--lanes", "--output-root", "--date"]: + assert opt in result.output, f"Expected '{opt}' in help output" + + def test_cli_invokes_run_all_with_defaults(self, tmp_path, monkeypatch): + runner = load_runner() + captured = {} + + def fake_run_all(**kwargs): + captured.update(kwargs) + return 0 + + monkeypatch.setattr(runner, "run_all", fake_run_all) + cli_runner = CliRunner() + result = cli_runner.invoke( + runner.app, ["--output-root", str(tmp_path), "--date", "2026-05-14"] + ) + assert result.exit_code == 0, f"CLI failed: {result.output}" + assert captured["num_rows"] == 512 + assert captured["hash_bits"] == 512 + assert captured["lanes"] == 8 + assert captured["output_root"] == tmp_path + assert captured["date"] == "2026-05-14" + assert captured["scenarios"] == runner.default_scenarios