feat: init mlkem project with Verilator test framework

- sync_rtl/common/: skid_buffer, pipeline_reg, defines (valid/ready) - sync_rtl/mod_add/: modular adder example with Verilator C++ TB - test_framework/: Python-driven Verilator compile/sim/compare pipeline - test_framework/modules/mod_add/: 50-vector test plan, full鏈路 PASS - .trellis/spec/: RTL and test_framework conventions documented
2026-06-24 19:43:29 +08:00
commit 8fdf944555
216 changed files with 24140 additions and 0 deletions
--- a/test_framework/lib/init.py
+++ b/test_framework/lib/init.py
--- a/test_framework/lib/pycache/init.cpython-313.pyc
+++ b/test_framework/lib/pycache/init.cpython-313.pyc
--- a/test_framework/lib/pycache/reporter.cpython-313.pyc
+++ b/test_framework/lib/pycache/reporter.cpython-313.pyc
--- a/test_framework/lib/pycache/result_checker.cpython-313.pyc
+++ b/test_framework/lib/pycache/result_checker.cpython-313.pyc
--- a/test_framework/lib/pycache/sim_controller.cpython-313.pyc
+++ b/test_framework/lib/pycache/sim_controller.cpython-313.pyc
--- a/test_framework/lib/pycache/test_runner.cpython-313.pyc
+++ b/test_framework/lib/pycache/test_runner.cpython-313.pyc
--- a/test_framework/lib/pycache/vector_gen.cpython-313.pyc
+++ b/test_framework/lib/pycache/vector_gen.cpython-313.pyc
--- a/test_framework/lib/reporter.py
+++ b/test_framework/lib/reporter.py
@@ -0,0 +1,130 @@
+"""reporter.py - Formats and displays test results."""
+
+import os
+import time
+from datetime import datetime
+
+
+class Reporter:
+    """Formats test results for terminal and HTML output."""
+
+    # ANSI color codes
+    GREEN = '\033[92m'
+    RED = '\033[91m'
+    YELLOW = '\033[93m'
+    CYAN = '\033[96m'
+    RESET = '\033[0m'
+    BOLD = '\033[1m'
+
+    def __init__(self, report_dir: str = 'test_framework/reports'):
+        self.report_dir = report_dir
+        os.makedirs(report_dir, exist_ok=True)
+
+    def print_summary(self, results: dict, total_elapsed: float = 0.0) -> bool:
+        """Print formatted test summary to terminal.
+
+        Args:
+            results: Dict of module_name -> module_results.
+            total_elapsed: Total elapsed time in seconds.
+
+        Returns:
+            bool: True if all tests passed.
+        """
+        total_pass = 0
+        total_fail = 0
+        all_pass = True
+
+        print()
+        print(f"{self.BOLD}{'='*60}{self.RESET}")
+        print(f"{self.BOLD}  TEST RESULTS{self.RESET}")
+        print(f"{self.BOLD}{'='*60}{self.RESET}")
+
+        for module_name, module_result in results.items():
+            status = module_result.get('status', 'UNKNOWN')
+            num_pass = module_result.get('pass', 0)
+            num_fail = module_result.get('fail', 0)
+            elapsed = module_result.get('elapsed', 0.0)
+            total_pass += num_pass
+            total_fail += num_fail
+
+            if status == 'PASS':
+                color = self.GREEN
+                icon = '✓'
+            elif status == 'FAIL':
+                color = self.RED
+                icon = '✗'
+                all_pass = False
+            else:
+                color = self.YELLOW
+                icon = '?'
+                all_pass = False
+
+            print(f"  {color}{icon} {module_name}{self.RESET}: "
+                  f"{num_pass} pass, {num_fail} fail, "
+                  f"{elapsed:.2f}s")
+
+        print(f"{self.BOLD}{'='*60}{self.RESET}")
+        total = total_pass + total_fail
+        if all_pass:
+            print(f"  {self.GREEN}{self.BOLD}ALL TESTS PASSED{self.RESET} "
+                  f"({total_pass}/{total} vectors, {total_elapsed:.2f}s)")
+        else:
+            print(f"  {self.RED}{self.BOLD}SOME TESTS FAILED{self.RESET} "
+                  f"({total_pass}/{total} pass, {total_fail}/{total} fail, "
+                  f"{total_elapsed:.2f}s)")
+        print()
+
+        # Write HTML report
+        self._write_html_report(results, total_elapsed)
+        return all_pass
+
+    def _write_html_report(self, results: dict, total_elapsed: float) -> None:
+        """Write an HTML test report."""
+        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+        report_path = os.path.join(self.report_dir, f'report_{timestamp}.html')
+
+        total_pass = sum(r.get('pass', 0) for r in results.values())
+        total_fail = sum(r.get('fail', 0) for r in results.values())
+        all_pass = total_fail == 0
+
+        rows = ''
+        for module_name, module_result in results.items():
+            status = module_result.get('status', 'UNKNOWN')
+            color = '#4CAF50' if status == 'PASS' else '#F44336'
+            rows += f'''
+            <tr>
+                <td>{module_name}</td>
+                <td style="color:{color}">{status}</td>
+                <td>{module_result.get('pass', 0)}</td>
+                <td>{module_result.get('fail', 0)}</td>
+                <td>{module_result.get('elapsed', 0):.2f}s</td>
+            </tr>'''
+
+        html = f'''<!DOCTYPE html>
+<html>
+<head>
+    <title>ML-KEM Test Report</title>
+    <style>
+        body {{ font-family: monospace; background: #1e1e1e; color: #d4d4d4;
+               padding: 20px; }}
+        h1 {{ color: {'#4CAF50' if all_pass else '#F44336'}; }}
+        table {{ border-collapse: collapse; width: 100%; }}
+        th, td {{ border: 1px solid #444; padding: 8px; text-align: left; }}
+        th {{ background: #333; }}
+    </style>
+</head>
+<body>
+    <h1>{'ALL TESTS PASSED' if all_pass else 'SOME TESTS FAILED'}</h1>
+    <p>Generated: {timestamp}<br>
+    Total: {total_pass + total_fail} vectors ({total_pass} pass, {total_fail} fail)<br>
+    Elapsed: {total_elapsed:.2f}s</p>
+    <table>
+        <tr><th>Module</th><th>Status</th><th>Pass</th><th>Fail</th><th>Time</th></tr>
+        {rows}
+    </table>
+</body>
+</html>'''
+
+        with open(report_path, 'w') as f:
+            f.write(html)
+        print(f"  Report saved to: {report_path}")
--- a/test_framework/lib/result_checker.py
+++ b/test_framework/lib/result_checker.py
@@ -0,0 +1,45 @@
+"""result_checker.py - Compares simulation output against expected values."""
+
+
+def check_results(got: list[str], expected_file: str) -> tuple[int, int]:
+    """Compare Verilator output against expected values.
+
+    Args:
+        got: List of result strings from simulation (hex values).
+        expected_file: Path to file containing expected hex values, one per line.
+
+    Returns:
+        tuple[int, int]: (num_pass, num_fail).
+    """
+    # Read expected values
+    expected = []
+    try:
+        with open(expected_file, 'r') as f:
+            for line in f:
+                line = line.strip()
+                if line and not line.startswith('#'):
+                    expected.append(line)
+    except FileNotFoundError:
+        print(f"[ERROR] Expected file not found: {expected_file}")
+        return (0, len(got))
+
+    num_pass = 0
+    num_fail = 0
+
+    max_len = max(len(got), len(expected))
+
+    for i in range(max_len):
+        got_val = got[i].upper() if i < len(got) else 'MISSING'
+        exp_val = expected[i].upper() if i < len(expected) else 'MISSING'
+
+        if got_val == exp_val:
+            num_pass += 1
+        else:
+            num_fail += 1
+            if num_fail <= 5:  # Only show first 5 failures
+                print(f"  MISMATCH[{i}]: got={got_val}, expected={exp_val}")
+
+    if num_fail > 5:
+        print(f"  ... and {num_fail - 5} more mismatches")
+
+    return (num_pass, num_fail)
--- a/test_framework/lib/sim_controller.py
+++ b/test_framework/lib/sim_controller.py
@@ -0,0 +1,134 @@
+"""sim_controller.py - Verilator simulation controller.
+
+Handles Verilator compilation and execution of RTL testbenches.
+"""
+
+import subprocess
+import os
+import time
+
+
+class SimController:
+    """Controls Verilator compilation and simulation."""
+
+    def __init__(self, config: dict):
+        """Initialize with test framework configuration.
+
+        Args:
+            config: The loaded config.json dict.
+        """
+        self.config = config
+        self.verilator_path = config.get('verilator', {}).get('path', 'verilator')
+        self.compile_args = config.get('verilator', {}).get('compile_args', [])
+
+    def compile(self, top_module: str, rtl_files: list[str], tb_cpp: str,
+                work_dir: str, inc_dirs: list[str] = None) -> str:
+        """Compile RTL with Verilator and return path to executable.
+
+        Args:
+            top_module: Name of the top-level Verilog module.
+            rtl_files: List of Verilog source file paths.
+            tb_cpp: Path to C++ testbench file.
+            work_dir: Working directory for compilation output (-Mdir).
+            inc_dirs: List of include directories for `include directives.
+
+        Returns:
+            Path to the compiled executable.
+
+        Raises:
+            RuntimeError: If Verilator is not found or compilation fails.
+        """
+        os.makedirs(work_dir, exist_ok=True)
+
+        # Build verilator command
+        cmd = [self.verilator_path]
+        cmd.extend(self.compile_args)
+        cmd.extend(['--top-module', top_module])
+        # No explicit -std= flag; let Verilator use its default (C++14+)
+        if inc_dirs:
+            for d in inc_dirs:
+                cmd.append(f'+incdir+{d}')
+        cmd.extend(rtl_files)
+        cmd.append(tb_cpp)
+        cmd.extend(['-Mdir', work_dir])
+
+        try:
+            result = subprocess.run(
+                cmd,
+                capture_output=True,
+                text=True,
+                timeout=120,
+                cwd=os.getcwd()
+            )
+        except subprocess.TimeoutExpired:
+            raise RuntimeError("Verilator compilation timed out.")
+        except FileNotFoundError:
+            raise RuntimeError(
+                f"Verilator not found at '{self.verilator_path}'."
+            )
+
+        if result.returncode != 0:
+            tail = result.stderr[-800:] if result.stderr else '(no stderr)'
+            raise RuntimeError(f"Verilator compilation failed:\n{tail}")
+
+        # Executable is at <work_dir>/V<top_module>
+        executable = os.path.join(work_dir, f'V{top_module}')
+        if not os.path.exists(executable):
+            raise RuntimeError(
+                f"Compilation succeeded but executable not found: {executable}"
+            )
+        return executable
+
+    def run(self, executable: str, vector_file: str,
+            timeout_s: int = 30) -> dict:
+        """Run the compiled simulation executable.
+
+        Args:
+            executable: Path to the compiled executable.
+            vector_file: Path to the hex vector file.
+            timeout_s: Timeout in seconds.
+
+        Returns:
+            dict with keys: success, results, elapsed, stderr.
+        """
+        if not os.path.exists(executable):
+            return {
+                'success': False,
+                'results': [],
+                'elapsed': 0.0,
+                'stderr': f'Executable not found: {executable}'
+            }
+
+        start_time = time.time()
+        try:
+            result = subprocess.run(
+                [executable, f'+VECTOR_FILE={vector_file}'],
+                capture_output=True,
+                text=True,
+                timeout=timeout_s
+            )
+            elapsed = time.time() - start_time
+
+            # Parse stdout for "RESULT: XXX" lines
+            results = []
+            for line in result.stdout.splitlines():
+                line = line.strip()
+                if line.startswith('RESULT:'):
+                    hex_val = line.split(':', 1)[1].strip()
+                    results.append(hex_val)
+
+            success = result.returncode == 0 and len(results) > 0
+            return {
+                'success': success,
+                'results': results,
+                'elapsed': elapsed,
+                'stderr': result.stderr
+            }
+        except subprocess.TimeoutExpired:
+            elapsed = time.time() - start_time
+            return {
+                'success': False,
+                'results': [],
+                'elapsed': elapsed,
+                'stderr': 'Simulation timed out.'
+            }
--- a/test_framework/lib/test_runner.py
+++ b/test_framework/lib/test_runner.py
@@ -0,0 +1,243 @@
+"""test_runner.py - Main test engine.
+
+Discovers modules, loads test plans, generates vectors, runs simulations,
+and collects results.
+"""
+
+import os
+import json
+import time
+import importlib.util
+import sys
+from pathlib import Path
+
+from .sim_controller import SimController
+from .result_checker import check_results
+
+
+class TestRunner:
+    """Main test engine for the ML-KEM RTL test framework."""
+
+    def __init__(self, config_path: str = 'test_framework/config.json'):
+        """Initialize with configuration.
+
+        Args:
+            config_path: Path to config.json.
+        """
+        self.project_root = os.path.dirname(os.path.dirname(
+            os.path.dirname(os.path.abspath(__file__))))
+        self.config_path = os.path.join(self.project_root, config_path)
+        with open(self.config_path) as f:
+            self.config = json.load(f)
+        self.sim = SimController(self.config)
+        self.modules_dir = os.path.join(self.project_root,
+                                        'test_framework', 'modules')
+        self.rtl_root = os.path.join(self.project_root,
+                                     self.config.get('rtl_root', 'sync_rtl'))
+
+    def discover_modules(self) -> dict[str, dict]:
+        """Scan test_framework/modules/*/test_plan.json and return plans.
+
+        Returns:
+            dict of module_name -> test_plan.
+        """
+        modules = {}
+        if not os.path.isdir(self.modules_dir):
+            return modules
+
+        for entry in os.listdir(self.modules_dir):
+            mod_dir = os.path.join(self.modules_dir, entry)
+            plan_path = os.path.join(mod_dir, 'test_plan.json')
+            if os.path.isdir(mod_dir) and os.path.exists(plan_path):
+                with open(plan_path) as f:
+                    plan = json.load(f)
+                modules[entry] = plan
+        return modules
+
+    def _get_rtl_files(self, plan: dict) -> list[str]:
+        """Resolve RTL source file paths from a test plan.
+
+        Args:
+            plan: Test plan dict with 'rtl_top' and optional 'rtl_deps'.
+
+        Returns:
+            List of absolute paths: deps first, then top.
+        """
+        rtl_top = os.path.join(self.project_root, plan['rtl_top'])
+        rtl_deps = [os.path.join(self.project_root, d)
+                    for d in plan.get('rtl_deps', [])]
+        return rtl_deps + [rtl_top]
+
+    def run_module(self, module_name: str, quick: bool = False,
+                   single_case: str = None) -> dict:
+        """Run all test cases for a module.
+
+        Args:
+            module_name: Name of the module to test.
+            quick: If True, reduce vector count for faster testing.
+            single_case: If set, only run this case ID.
+
+        Returns:
+            dict with keys: pass, fail, cases, elapsed, status.
+        """
+        modules = self.discover_modules()
+        if module_name not in modules:
+            return {
+                'pass': 0, 'fail': 0, 'cases': {},
+                'elapsed': 0.0, 'status': 'NOT_FOUND',
+                'error': f'Module "{module_name}" not found'
+            }
+
+        plan = modules[module_name]
+        mod_dir = os.path.join(self.modules_dir, module_name)
+        vectors_dir = os.path.join(mod_dir, 'vectors')
+        work_dir = os.path.join(mod_dir, 'obj_dir')
+        os.makedirs(vectors_dir, exist_ok=True)
+
+        # Import module's gen_vectors.py
+        gen_path = os.path.join(mod_dir, 'gen_vectors.py')
+        if not os.path.exists(gen_path):
+            return {
+                'pass': 0, 'fail': 0, 'cases': {},
+                'elapsed': 0.0, 'status': 'ERROR',
+                'error': f'gen_vectors.py not found for {module_name}'
+            }
+
+        spec = importlib.util.spec_from_file_location(
+            f'{module_name}_gen', gen_path)
+        gen_module = importlib.util.module_from_spec(spec)
+        sys.modules[f'{module_name}_gen'] = gen_module
+        spec.loader.exec_module(gen_module)
+
+        # Find generator class
+        generator = None
+        for name in dir(gen_module):
+            obj = getattr(gen_module, name)
+            if (isinstance(obj, type) and
+                    hasattr(obj, 'generate_one') and
+                    hasattr(obj, 'write_hex_file') and
+                    name != 'VectorGenerator'):
+                generator = obj()
+                break
+
+        if generator is None:
+            return {
+                'pass': 0, 'fail': 0, 'cases': {},
+                'elapsed': 0.0, 'status': 'ERROR',
+                'error': f'No generator class found in {gen_path}'
+            }
+
+        # Resolve RTL files and top module name
+        rtl_files = self._get_rtl_files(plan)
+        tb_cpp = os.path.join(self.project_root, plan['tb_cpp'])
+
+        # Top module name derived from rtl_top filename (strip .v)
+        top_module = os.path.basename(plan['rtl_top']).replace('.v', '')
+
+        # Compile once for the module
+        print(f"\n  Compiling {module_name} with Verilator...")
+        start_time = time.time()
+
+        try:
+            executable = self.sim.compile(
+                top_module=top_module,
+                rtl_files=rtl_files,
+                tb_cpp=tb_cpp,
+                work_dir=work_dir,
+                inc_dirs=[self.project_root]
+            )
+        except RuntimeError as e:
+            return {
+                'pass': 0, 'fail': 0, 'cases': {},
+                'elapsed': time.time() - start_time,
+                'status': 'COMPILE_ERROR',
+                'error': str(e)
+            }
+
+        case_results = {}
+        total_pass = 0
+        total_fail = 0
+
+        for case in plan.get('cases', []):
+            case_id = case['id']
+
+            # Filter by single case
+            if single_case and case_id != single_case:
+                continue
+
+            num_vectors = case.get('num_vectors', 10)
+            if quick:
+                num_vectors = max(1, num_vectors // 5)
+
+            params = case.get('params', {})
+            print(f"  Case [{case_id}]: {case.get('description', '')} "
+                  f"({num_vectors} vectors)")
+
+            # Generate vectors
+            vectors = []
+            for i in range(num_vectors):
+                vec = generator.generate_one(params)
+                vectors.append(vec)
+
+            # Write input vectors hex file
+            input_file = os.path.join(vectors_dir, f'{case_id}_input.hex')
+            generator.write_hex_file(vectors, input_file)
+
+            # Write expected output hex file
+            expected_file = os.path.join(vectors_dir, f'{case_id}_expected.hex')
+            generator.write_expected_file(vectors, expected_file)
+
+            # Run simulation
+            timeout_s = plan.get('timeout_s', 30)
+            run_result = self.sim.run(executable, input_file, timeout_s)
+
+            if not run_result['success']:
+                case_results[case_id] = {
+                    'pass': 0, 'fail': num_vectors,
+                    'elapsed': run_result['elapsed'],
+                    'error': run_result.get('stderr', 'Simulation failed')
+                }
+                total_fail += num_vectors
+                continue
+
+            # Compare results
+            tolerance = case.get('tolerance', 'bit_exact')
+            if tolerance == 'bit_exact':
+                num_pass, num_fail = check_results(
+                    run_result['results'], expected_file)
+            else:
+                # Delegate to generator for custom comparison
+                if hasattr(generator, 'compare_results'):
+                    compare_ok = generator.compare_results(
+                        run_result['results'], expected_file)
+                    num_pass = num_vectors if compare_ok else 0
+                    num_fail = 0 if compare_ok else num_vectors
+                else:
+                    num_pass, num_fail = check_results(
+                        run_result['results'], expected_file)
+
+            case_results[case_id] = {
+                'pass': num_pass,
+                'fail': num_fail,
+                'elapsed': run_result['elapsed']
+            }
+            total_pass += num_pass
+            total_fail += num_fail
+
+            # Clean up vectors unless keep_vectors is true
+            if not self.config.get('keep_vectors', False):
+                try:
+                    os.remove(input_file)
+                    os.remove(expected_file)
+                except OSError:
+                    pass
+
+        elapsed = time.time() - start_time
+
+        return {
+            'pass': total_pass,
+            'fail': total_fail,
+            'cases': case_results,
+            'elapsed': elapsed,
+            'status': 'PASS' if total_fail == 0 else 'FAIL'
+        }
--- a/test_framework/lib/vector_gen.py
+++ b/test_framework/lib/vector_gen.py
@@ -0,0 +1,61 @@
+"""vector_gen.py - Base class for vector generators.
+
+Each module provides a gen_vectors.py that subclasses VectorGenerator.
+The generate_one() method is abstract and must be overridden.
+The write_hex_file() method writes vectors as hex-formatted files.
+"""
+
+import os
+from abc import ABC, abstractmethod
+
+
+class VectorGenerator(ABC):
+    """Base class for test vector generators."""
+
+    @abstractmethod
+    def generate_one(self, params: dict) -> dict:
+        """Generate a single test vector.
+
+        Args:
+            params: Module-specific parameters for vector generation.
+
+        Returns:
+            dict with 'input' and 'expected' keys, each containing a dict
+            of signal_name -> value.
+        """
+        ...
+
+    def write_hex_file(self, vectors: list[dict], filepath: str) -> None:
+        """Write vectors to a hex file.
+
+        Subclasses should override this to match their module's input format.
+        Default: writes each vector on one line, hex values separated by spaces.
+
+        Args:
+            vectors: List of result dicts from generate_one().
+            filepath: Path to write the hex file.
+        """
+        os.makedirs(os.path.dirname(filepath), exist_ok=True)
+        with open(filepath, 'w') as f:
+            for v in vectors:
+                # Default: write input values as space-separated hex
+                inputs = v.get('input', {})
+                hex_vals = [format(val, 'X') for val in inputs.values()]
+                f.write(' '.join(hex_vals) + '\n')
+
+    def write_expected_file(self, vectors: list[dict], filepath: str) -> None:
+        """Write expected results to a hex file.
+
+        Subclasses should override this to match their module's expected format.
+        Default: writes each expected value on one line.
+
+        Args:
+            vectors: List of result dicts from generate_one().
+            filepath: Path to write the expected hex file.
+        """
+        os.makedirs(os.path.dirname(filepath), exist_ok=True)
+        with open(filepath, 'w') as f:
+            for v in vectors:
+                expected = v.get('expected', {})
+                hex_vals = [format(val, 'X') for val in expected.values()]
+                f.write(' '.join(hex_vals) + '\n')