tests_python/test_validator_graph.py

from arena.models import CodebaseArchive, ValidationCheck
from arena.validator_plan import ValidatorContext, ValidatorPlan
from arena.validator_graph import DeterministicValidatorExecutor, DeterministicRubricReviewer, RubricReview, ValidatorGraph


def test_validator_passes_when_all_checks_succeed() -> None:
    graph = ValidatorGraph(executor=DeterministicValidatorExecutor())

    report = graph.validate(
        run_id="run-1",
        codebase_archive=CodebaseArchive(pointer="local-snapshot://run-1", file_count=2, size_bytes=128),
        plan=ValidatorPlan.from_checks([
            ValidationCheck(id="check-1", name="tests", command="uv run pytest", source="user"),
            ValidationCheck(id="check-2", name="syntax", command="python -m compileall ."),
        ]),
    )

    assert report.run_id == "run-1"
    assert report.passed is True
    assert len(report.results) == 2
    assert all(r.passed for r in report.results)
    assert "Validation checks passed: 2/2" in report.summary
    assert report.risks == []


def test_validator_fails_when_check_fails() -> None:
    graph = ValidatorGraph(executor=DeterministicValidatorExecutor())

    report = graph.validate(
        run_id="run-1",
        codebase_archive=CodebaseArchive(pointer="local-snapshot://run-1", file_count=2, size_bytes=128),
        plan=ValidatorPlan.from_checks([
            ValidationCheck(id="check-1", name="tests", command="uv run pytest", source="user"),
            ValidationCheck(id="check-2", name="bad", command="fail the build"),
        ]),
    )

    assert report.passed is False
    assert [r.passed for r in report.results] == [True, False]
    assert "Validation checks passed: 1/2" in report.summary
    assert len(report.risks) >= 1


def test_validator_with_no_checks_passes() -> None:
    graph = ValidatorGraph(executor=DeterministicValidatorExecutor())

    report = graph.validate(
        run_id="run-1",
        codebase_archive=CodebaseArchive(pointer="local-snapshot://run-1", file_count=1, size_bytes=10),
        plan=ValidatorPlan(),
    )

    assert report.passed is True
    assert "No validation checks configured" in report.summary


def test_validator_includes_rubric_review_by_default() -> None:
    graph = ValidatorGraph(executor=DeterministicValidatorExecutor())

    report = graph.validate(
        run_id="run-1",
        prompt="Build a CLI",
        criteria=[],
        codebase_archive=CodebaseArchive(pointer="local-snapshot://run-1", file_count=1, size_bytes=10),
        plan=ValidatorPlan.from_checks([
            ValidationCheck(id="check-1", name="tests", command="uv run pytest", source="user"),
        ]),
    )

    assert report.rubric_score is not None
    assert 0.0 <= report.rubric_score <= 1.0
    assert "rubric score" in report.summary.lower()


def test_validator_with_custom_rubric_reviewer() -> None:
    class CustomReviewer:
        def review(self, *, prompt, criteria, check_results):
            return RubricReview(score=0.75, feedback="custom review", risks=["risk-1"])

    graph = ValidatorGraph(executor=DeterministicValidatorExecutor(), rubric_reviewer=CustomReviewer())

    report = graph.validate(
        run_id="run-1",
        prompt="Build a thing",
        codebase_archive=CodebaseArchive(pointer="local-snapshot://run-1", file_count=1, size_bytes=10),
        plan=ValidatorPlan.from_checks([
            ValidationCheck(id="check-1", name="tests", command="uv run pytest", source="user"),
        ]),
    )

    assert report.rubric_score == 0.75
    assert report.rubric_feedback == "custom review"
    assert "risk-1" in report.risks


def test_validator_uses_prompt_and_criteria_from_plan_context() -> None:
    from arena.models import Criteria

    captured: dict = {}

    class CaptureReviewer:
        def review(self, *, prompt, criteria, check_results):
            captured["prompt"] = prompt
            captured["criteria"] = list(criteria)
            return RubricReview(score=0.9, feedback="ok", risks=[])

    context = ValidatorContext(prompt="Build a CLI", criteria=(Criteria(text="tests must pass"),))

    graph = ValidatorGraph(
        executor=DeterministicValidatorExecutor(),
        rubric_reviewer=CaptureReviewer(),
    )

    graph.validate(
        run_id="run-1",
        codebase_archive=CodebaseArchive(pointer="local-snapshot://run-1", file_count=1, size_bytes=10),
        plan=ValidatorPlan.from_user_tests(["echo ok"], context=context),
    )

    assert captured["prompt"] == "Build a CLI"
    assert captured["criteria"][0].text == "tests must pass"


def test_validator_runs_generated_gradio_smoke_check_from_plan_context() -> None:
    from arena.models import Criteria

    commands: list[str] = []

    class CaptureExecutor:
        def run(self, archive_pointer: str, command: str):
            commands.append(command)
            return True, "ok"

    context = ValidatorContext(
        prompt="Build a Gradio demo app",
        criteria=(Criteria(text="Works as a Gradio app"),),
    )
    graph = ValidatorGraph(executor=CaptureExecutor(), rubric_enabled=False)

    report = graph.validate(
        run_id="run-1",
        codebase_archive=CodebaseArchive(pointer="local-snapshot://run-1", file_count=1, size_bytes=10),
        plan=ValidatorPlan.from_user_tests(["python -m py_compile app.py"], context=context, rubric_enabled=False),
    )

    assert report.passed is True
    assert len(commands) == 2
    assert "127.0.0.1:{port}" in commands[1]


def test_validator_skips_rubric_when_plan_has_no_rubric_step() -> None:
    calls: list[bool] = []

    def counting_reviewer(*, prompt, criteria, check_results):
        calls.append(True)
        return RubricReview(score=1.0, feedback="ok", risks=[])

    graph = ValidatorGraph(
        executor=DeterministicValidatorExecutor(),
        rubric_reviewer=counting_reviewer,
        rubric_enabled=True,
    )

    plan = ValidatorPlan.from_user_tests(["echo ok"], rubric_enabled=False)
    report = graph.validate(
        run_id="run-1",
        codebase_archive=CodebaseArchive(pointer="local-snapshot://run-1", file_count=1, size_bytes=10),
        plan=plan,
    )

    assert calls == []
    assert report.rubric_score is None
    assert "rubric score" not in report.summary.lower()


def test_validator_closes_executor_after_validation() -> None:
    class CloseableExecutor:
        def __init__(self) -> None:
            self.closed = 0

        def run(self, archive_pointer: str, command: str):
            return True, "ok"

        def close(self) -> None:
            self.closed += 1

    executor = CloseableExecutor()
    graph = ValidatorGraph(executor=executor, rubric_enabled=False)

    report = graph.validate(
        run_id="run-1",
        codebase_archive=CodebaseArchive(pointer="local-snapshot://run-1", file_count=1, size_bytes=10),
        plan=ValidatorPlan.from_user_tests(["echo ok"]),
    )

    assert report.passed is True
    assert executor.closed == 1