FallenSigh
8fdf944555
- 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
244 lines
8.5 KiB
Python
244 lines
8.5 KiB
Python
"""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'
|
|
}
|