File size: 3,848 Bytes
1866d85 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | 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"
|