tests/test_evaluation.py

from __future__ import annotations

import asyncio
import json

from autocad_bench.evaluation.run import evaluate_rollout
from autocad_bench.tasks.manifest import load_manifest
from autocad_bench.evaluation.scoring import GoldCacheStore
from autocad_bench.evaluation.scoring.models import EvaluationFailure
from tests.scoring_helpers import invalid_drawing, valid_drawing


def test_completed_rollout_is_scored_and_publishes_status(tmp_path) -> None:
    entry = next(item for item in load_manifest() if item.task_id == "task-001")
    candidate_dwg = entry.resolve_gold_path().read_bytes()
    drawing = valid_drawing(
        candidate_dwg,
        evaluator_version="static-test-v1",
        drawing_type="2d",
    )
    rollout_dir = tmp_path / "rollout"
    evaluation_dir = rollout_dir / "evaluation"
    evaluation_dir.mkdir(parents=True)
    (rollout_dir / "attempt.dwg").write_bytes(candidate_dwg)
    (evaluation_dir / "metadata.json").write_text(
        drawing.metadata.model_dump_json(indent=2),
        encoding="utf-8",
    )
    (evaluation_dir / "render.png").write_bytes(drawing.render_png)
    cache = GoldCacheStore(evaluator_version="static-test-v1", root=tmp_path / "gold")
    cache.write("task-001", drawing)

    outcome = asyncio.run(
        evaluate_rollout(
            rollout_dir=rollout_dir,
            task_id="task-001",
            evaluator_version="static-test-v1",
            gold_cache_root=tmp_path / "gold",
            vision_api_key=None,
            vision_enabled=False,
        )
    )

    status = json.loads((evaluation_dir / "status.json").read_text())
    score = json.loads((evaluation_dir / "score.json").read_text())
    assert outcome["deterministic_score"] == 1.0
    assert outcome["vision_skipped_reason"] == "disabled"
    assert score["score"] == 1.0
    assert (evaluation_dir / "diff.png").is_file()
    assert status["stage"] == "completed"
    assert status["deterministic_score"] == 1.0


def test_invalid_dwg_gets_a_terminal_zero_without_calling_vision(tmp_path) -> None:
    entry = next(item for item in load_manifest() if item.task_id == "task-001")
    candidate_dwg = b"AC1032-invalid-candidate"
    invalid = invalid_drawing(
        candidate_dwg,
        EvaluationFailure.EMPTY_MODEL_SPACE,
        evaluator_version="static-test-v1",
    )
    invalid = invalid.model_copy(
        update={
            "metadata": invalid.metadata.model_copy(
                update={"drawing_type": "2d"}
            )
        }
    )
    rollout_dir = tmp_path / "rollout"
    evaluation_dir = rollout_dir / "evaluation"
    evaluation_dir.mkdir(parents=True)
    (rollout_dir / "attempt.dwg").write_bytes(candidate_dwg)
    (evaluation_dir / "metadata.json").write_text(
        invalid.metadata.model_dump_json(indent=2),
        encoding="utf-8",
    )
    (evaluation_dir / "render.png").write_bytes(b"")
    gold_dwg = entry.resolve_gold_path().read_bytes()
    cache = GoldCacheStore(evaluator_version="static-test-v1", root=tmp_path / "gold")
    cache.write(
        "task-001",
        valid_drawing(
            gold_dwg,
            evaluator_version="static-test-v1",
            drawing_type="2d",
        ),
    )

    outcome = asyncio.run(
        evaluate_rollout(
            rollout_dir=rollout_dir,
            task_id="task-001",
            evaluator_version="static-test-v1",
            gold_cache_root=tmp_path / "gold",
            vision_api_key=None,
            vision_enabled=True,
        )
    )

    score = json.loads((evaluation_dir / "score.json").read_text())
    status = json.loads((evaluation_dir / "status.json").read_text())
    assert outcome["deterministic_score"] == 0.0
    assert outcome["vision_skipped_reason"] == "invalid_candidate"
    assert score["failure_reason"] == "empty_model_space"
    assert status["stage"] == "completed"