tests/test_mlir_graph_builder_resilience.py

from __future__ import annotations

import csv
import hashlib
import json
import sys
from pathlib import Path
from types import SimpleNamespace

from scripts import build_mlir_ir_graphs as builder
from scripts import mlir_graph_common as common


PNG = b"\x89PNG\r\n\x1a\nfixture-png"


def digest(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def test_render_requires_a_new_temporary_png(
    tmp_path: Path, monkeypatch
) -> None:
    svg = tmp_path / "graph.svg"
    png = tmp_path / "graph.png"
    svg.write_text("<svg/>\n", encoding="utf-8")
    png.write_bytes(PNG + b"-stale")
    stale_digest = digest(png)

    monkeypatch.setattr(
        builder.subprocess,
        "run",
        lambda *args, **kwargs: SimpleNamespace(returncode=0),
    )
    result = builder.render_one(
        svg=svg,
        png=png,
        log_dir=tmp_path / "logs",
        inkscape="/fixture/inkscape",
        root=tmp_path,
    )

    assert result["status"] == "FAIL"
    assert result["failure_code"] == "FAIL_ANALYSIS"
    assert result["output_replaced"] is False
    assert "RENDER_OUTPUT_MISSING" in result["failure_detail"]
    assert digest(png) == stale_digest


def test_render_atomically_replaces_only_valid_new_png(
    tmp_path: Path, monkeypatch
) -> None:
    svg = tmp_path / "graph.svg"
    png = tmp_path / "graph.png"
    svg.write_text("<svg/>\n", encoding="utf-8")
    png.write_bytes(PNG + b"-stale")

    def fake_run(command: list[str], **kwargs: object) -> SimpleNamespace:
        output = next(value.split("=", 1)[1] for value in command if value.startswith("--export-filename="))
        Path(output).write_bytes(PNG + b"-new")
        return SimpleNamespace(returncode=0)

    monkeypatch.setattr(builder.subprocess, "run", fake_run)
    result = builder.render_one(
        svg=svg,
        png=png,
        log_dir=tmp_path / "logs",
        inkscape="/fixture/inkscape",
        root=tmp_path,
    )

    assert result["status"] == "PASS"
    assert result["failure_code"] is None
    assert result["output_replaced"] is True
    assert png.read_bytes() == PNG + b"-new"


def test_resume_json_and_fingerprint_failures_are_explicit(tmp_path: Path) -> None:
    record = tmp_path / "graph_record.json"
    svg = tmp_path / "graph.svg"
    png = tmp_path / "graph.png"
    svg.write_text("<svg/>\n", encoding="utf-8")
    png.write_bytes(PNG)

    record.write_text("{broken", encoding="utf-8")
    _, valid, error = builder.load_resume_record(record, "wanted", svg, png)
    assert valid is False
    assert "RESUME_RECORD_READ" in error

    record.write_text(
        json.dumps({"status": "PASS", "fingerprint": "old", "outputs": {}}),
        encoding="utf-8",
    )
    _, valid, error = builder.load_resume_record(record, "wanted", svg, png)
    assert valid is False
    assert "OUTPUT_SETTINGS_CONFLICT" in error


def test_supporting_evidence_rebases_only_success_artifacts(tmp_path: Path) -> None:
    artifact = tmp_path / "models" / "fixture" / "krnl.mlir"
    artifact.parent.mkdir(parents=True)
    artifact.write_text("module {}\n", encoding="utf-8")
    legacy = "/legacy/work/oldrepo/models/fixture/krnl.mlir"

    passed = builder.supporting_evidence(
        {
            "model_id": "M00",
            "krnl_status": "PASS",
            "krnl_artifact": legacy,
            "krnl_sha256": digest(artifact),
        },
        "fp32",
        "krnl",
        tmp_path,
    )
    failed = builder.supporting_evidence(
        {
            "model_id": "M00",
            "llvm_status": "FAIL",
            "llvm_artifact": "/legacy/oldrepo/models/missing/llvm.mlir",
            "llvm_sha256": "failure-evidence-sha",
        },
        "fp32",
        "llvm",
        tmp_path,
    )

    assert passed["validation"] == "CHECKSUM_VERIFIED"
    assert passed["artifact"] == "models/fixture/krnl.mlir"
    assert passed["failure"] == ""
    assert failed["validation"] == "FAILURE_EVIDENCE_ONLY"
    assert failed["artifact"] == "/legacy/oldrepo/models/missing/llvm.mlir"
    assert failed["sha256"] == "failure-evidence-sha"


def _write_fixture_repository(root: Path) -> tuple[Path, list[str]]:
    model_ids = sorted(common.AFFINE_PAIR_IDS) + [f"M{index:02d}" for index in range(14)]
    model_ids = sorted(model_ids)
    registry = root / "model_registry.csv"
    with registry.open("w", newline="", encoding="utf-8") as stream:
        writer = csv.DictWriter(
            stream, fieldnames=["model_id", "model_name", "eligibility"]
        )
        writer.writeheader()
        for model_id in model_ids:
            writer.writerow(
                {
                    "model_id": model_id,
                    "model_name": f"Fixture {model_id}",
                    "eligibility": "ELIGIBLE",
                }
            )

    mlir = """module {
  func.func @main(%arg0: tensor<1xf32>) -> tensor<1xf32> {
    %0 = \"onnx.Relu\"(%arg0) : (tensor<1xf32>) -> tensor<1xf32>
    return %0 : tensor<1xf32>
  }
  \"onnx.EntryPoint\"() {func = @main} : () -> ()
}
"""
    rows: list[dict[str, str]] = []
    for model_id in model_ids:
        for variant in common.VARIANTS:
            directory = root / "models" / model_id / variant
            directory.mkdir(parents=True)
            onnx = directory / "onnx.mlir"
            onnx.write_text(mlir, encoding="utf-8")
            affine = directory / "affine_scf_memref.mlir"
            if model_id in common.AFFINE_PAIR_IDS:
                affine.write_text(mlir, encoding="utf-8")
            rows.append(
                {
                    "model_id": model_id,
                    "task": "fixture",
                    "variant": variant,
                    "onnx_status": "PASS",
                    "onnx_artifact": str(onnx.relative_to(root)),
                    "onnx_sha256": digest(onnx),
                    "krnl_status": "FAIL",
                    "krnl_artifact": "/legacy/oldrepo/models/failure/krnl.mlir",
                    "krnl_sha256": "failure-only",
                    "affine_scf_memref_status": (
                        "PASS" if model_id in common.AFFINE_PAIR_IDS else "FAIL"
                    ),
                    "affine_scf_memref_artifact": (
                        str(affine.relative_to(root))
                        if model_id in common.AFFINE_PAIR_IDS
                        else "/legacy/oldrepo/models/failure/affine.mlir"
                    ),
                    "affine_scf_memref_sha256": (
                        digest(affine)
                        if model_id in common.AFFINE_PAIR_IDS
                        else "failure-only"
                    ),
                    "llvm_status": "FAIL",
                    "llvm_artifact": "/legacy/oldrepo/models/failure/llvm.mlir",
                    "llvm_sha256": "failure-only",
                    "last_fully_successful_ir": "AFFINE_SCF_MEMREF"
                    if model_id in common.AFFINE_PAIR_IDS
                    else "ONNX",
                }
            )
    matrix = root / "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)
    runtime_inventory = root / "reports" / "graphs" / "frontend" / "variant_inventory.csv"
    runtime_inventory.parent.mkdir(parents=True)
    runtime_inventory.write_text("fixture\n", encoding="utf-8")
    return matrix, model_ids


def test_main_isolates_parse_failure_and_finishes_other_variants(
    tmp_path: Path, monkeypatch
) -> None:
    matrix, model_ids = _write_fixture_repository(tmp_path)
    real_parse = builder.parse_mlir

    def parse_with_one_failure(path: Path, graph_id: str):
        if graph_id == f"{model_ids[0]}:fp32:ONNX":
            raise SyntaxError("fixture parser failure", (str(path), 7, 1, "bad"))
        return real_parse(path, graph_id)

    def fake_render(*, svg: Path, png: Path, log_dir: Path, **kwargs: object):
        png.write_bytes(PNG)
        log_dir.mkdir(parents=True, exist_ok=True)
        stdout = log_dir / "render.stdout.log"
        stderr = log_dir / "render.stderr.log"
        stdout.write_text("", encoding="utf-8")
        stderr.write_text("", encoding="utf-8")
        return {
            "status": "PASS",
            "failure_code": None,
            "failure_stage": None,
            "failure_detail": "",
            "command": "fixture-render",
            "command_argv": ["fixture-render"],
            "exit_code": 0,
            "output_replaced": True,
            "stdout_log": common.file_record(stdout, tmp_path),
            "stderr_log": common.file_record(stderr, tmp_path),
            "command_log": {"path": "", "sha256": ""},
            "exit_code_log": {"path": "", "sha256": ""},
        }

    monkeypatch.setattr(builder, "parse_mlir", parse_with_one_failure)
    monkeypatch.setattr(builder, "render_one", fake_render)
    monkeypatch.setattr(builder.shutil, "which", lambda name: "/fixture/inkscape")
    monkeypatch.setattr(
        builder.subprocess,
        "run",
        lambda *args, **kwargs: SimpleNamespace(
            returncode=0, stdout="Inkscape fixture", stderr=""
        ),
    )
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "build_mlir_ir_graphs.py",
            "--repo-root",
            str(tmp_path),
            "--coverage-matrix",
            str(matrix),
            "--output-dir",
            "reports/graphs/mlir",
            "--render-log-dir",
            "logs/graphs/mlir/fixture/renders",
            "--checkpoint",
            "logs/graphs/mlir/fixture/checkpoint.json",
            "--render-workers",
            "2",
        ],
    )

    assert builder.main() == 1
    report = tmp_path / "reports" / "graphs" / "mlir"
    inventory = list(
        csv.DictReader((report / "operation_inventory.csv").open(newline=""))
    )
    assert len(inventory) == 56
    assert sum(row["analysis_status"] == "FAIL" for row in inventory) == 1
    assert sum(row["analysis_status"] == "PASS" for row in inventory) == 55
    failure = next(row for row in inventory if row["analysis_status"] == "FAIL")
    failure_record = json.loads(
        (tmp_path / failure["graph_record_json"]).read_text(encoding="utf-8")
    )
    assert failure_record["failure_code"] == "FAIL_ANALYSIS"
    assert failure_record["failure_stage"] == "PARSE"
    assert failure_record["analysis_diagnostics"][0]["source_line"] == 7
    assert "fixture parser failure" in failure_record["failure_detail"]
    checkpoint = json.loads(
        (tmp_path / "logs/graphs/mlir/fixture/checkpoint.json").read_text(
            encoding="utf-8"
        )
    )
    assert checkpoint["completed_graph_count"] == 55
    assert checkpoint["failed_graph_count"] == 1
    assert all(row["graph_record_written"] for row in checkpoint["graphs"])