from __future__ import annotations import argparse import csv import hashlib from pathlib import Path from scripts import validate_mlir_ir_graphs as validator def digest(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def test_primary_graph_validator_accepts_exact_42_image_set(tmp_path: Path) -> None: rows: list[dict[str, str]] = [] report_dir = tmp_path / "reports/graphs/mlir" for model_index in range(21): model_id = f"M{model_index:02d}" for variant in ("fp32", "public_quantized"): source = tmp_path / "models" / model_id / variant / "onnx.mlir" source.parent.mkdir(parents=True) source.write_text("module {}\n", encoding="utf-8") rows.append( { "model_id": model_id, "variant": variant, "onnx_status": "PASS", "onnx_artifact": str(source.relative_to(tmp_path)), "onnx_sha256": digest(source), } ) graph_dir = report_dir / "graphs" / model_id / variant / "onnx" graph_dir.mkdir(parents=True) graph_id = f"{model_id}:{variant}:ONNX" (graph_dir / "execution_dependency_graph.svg").write_text( f"{graph_id} STATIC_MLIR_PROGRAM_ORDER\n", encoding="utf-8", ) (graph_dir / "execution_dependency_graph.png").write_bytes( b"\x89PNG\r\n\x1a\nfixture" ) matrix = tmp_path / "reports/conversion/ir_stage_coverage.csv" matrix.parent.mkdir(parents=True) with matrix.open("w", newline="", encoding="utf-8") as stream: writer = csv.DictWriter(stream, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) result, exit_code = validator.validate( argparse.Namespace( repo_root=tmp_path, report_dir=Path("reports/graphs/mlir"), coverage_matrix=Path("reports/conversion/ir_stage_coverage.csv"), output=Path("logs/validation.json"), ) ) assert exit_code == 0 assert result["status"] == "PASS" assert result["counts"]["models"] == 21 assert result["counts"]["variants"] == 42 assert result["counts"]["svg"] == 42 assert result["counts"]["png"] == 42 assert result["counts"]["failed"] == 0