tests_python/test_api.py

from dataclasses import dataclass
import asyncio

import arena.api as api
from arena.agent import _load_dotenv
from arena.api import (
    RejoinRequest,
    RunMessageRequest,
    add_run_message,
    create_run,
    create_thread,
    delete_thread,
    get_run,
    get_run_codebase_zip,
    get_run_field_notes,
    get_run_space_readme,
    get_run_submission_markdown,
    get_run_trace_json,
    health,
    list_provider_models,
    list_skills,
    list_files,
    read_file,
    refresh_provider_models,
    rejoin_run,
    sessions,
    validate_run,
)
from arena.codebase_archive import LocalCodebaseArchiveStore
from arena.model_catalog import ModelProviderConfig, ProviderModel
from arena.models import CodebaseArchive, Criteria, Run, RunEvent, RunRequest, RunTask, ValidationReport
from arena.run_store import InMemoryRunStore
from arena.swarm_runtime import SwarmRunResult
from arena.validator_graph import DeterministicValidatorExecutor, DeterministicRubricReviewer, ValidatorGraph


def run_async(coro):
    return asyncio.run(coro)


@dataclass
class FakeResult:
    error: str | None = None
    entries: list[dict] | None = None
    file_data: dict | None = None


class FakeBackend:
    def write(self, path: str, content: str) -> FakeResult:
        return FakeResult()

    def ls(self, path: str) -> FakeResult:
        assert path == "/"
        return FakeResult(entries=[{"path": "/hello.txt", "is_dir": False}])

    def read(self, path: str) -> FakeResult:
        assert path == "/hello.txt"
        return FakeResult(file_data={"content": "hi"})


class FakeAgent:
    def invoke(self, payload: dict, config: dict) -> dict:
        return {"messages": [{"role": "assistant", "content": "ok"}]}


class FakeSession:
    thread_id = "thread-1"
    provider = "local"
    backend = FakeBackend()
    agent = FakeAgent()
    messages: list[dict[str, str]] = []
    closed = False

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


class FakeSwarmRuntime:
    def __init__(self) -> None:
        self.touched_runs: list[str] = []

    def start_run(self, run: Run):
        return object()

    def execute_run(self, run_id: str, **kwargs) -> SwarmRunResult:
        return SwarmRunResult(
            summary="Built CLI.",
            tasks=[RunTask(id="task-1", title="Build codebase", status="completed")],
            events=[RunEvent(id="event-1", message="Built CLI.")],
            codebase_archive=CodebaseArchive(
                pointer="local-snapshot://run-1",
                file_count=1,
                size_bytes=10,
            ),
        )

    def close_run(self, run_id: str) -> None:
        pass

    def touch_run(self, run_id: str) -> bool:
        self.touched_runs.append(run_id)
        return True


def test_health() -> None:
    assert run_async(health()) == {"status": "ok"}


def test_skills_route_reports_repo_bundled_deepagents_skills() -> None:
    skills = run_async(list_skills())

    assert {skill.name for skill in skills} >= {"submission-checklist", "gradio-ui-builder"}


def test_load_dotenv_does_not_override_existing_env(tmp_path, monkeypatch) -> None:
    env_file = tmp_path / ".env"
    env_file.write_text("EXISTING=from-file\nNEW_VALUE='loaded'\n# skip\n")
    monkeypatch.setenv("EXISTING", "from-env")

    _load_dotenv(env_file)

    assert __import__("os").environ["EXISTING"] == "from-env"
    assert __import__("os").environ["NEW_VALUE"] == "loaded"


def test_file_listing_and_read_use_backend_result_shapes() -> None:
    sessions["fake"] = FakeSession()
    try:
        entries = run_async(list_files("fake", path="/"))
        content = run_async(read_file("fake", "/hello.txt"))
    finally:
        run_async(delete_thread("fake"))

    assert entries[0].path == "/hello.txt"
    assert entries[0].type == "file"
    assert content == {"path": "/hello.txt", "content": "hi"}


def test_run_routes_use_shared_service_layer(monkeypatch) -> None:
    from arena.service import ArenaService

    monkeypatch.setattr(
        api,
        "service",
        ArenaService(
            session_factory=lambda: FakeSession(),
            run_store=InMemoryRunStore(),
            swarm_runtime=FakeSwarmRuntime(),
            archive_store=LocalCodebaseArchiveStore(),
            validator_graph=ValidatorGraph(
                executor=DeterministicValidatorExecutor(),
                rubric_reviewer=DeterministicRubricReviewer(),
            ),
        ),
    )

    run = run_async(
        create_run(
            RunRequest(
                prompt="Build a CLI",
                criteria=[Criteria(text="pytest must pass")],
                user_tests=["uv run pytest"],
            )
        )
    )
    view = run_async(get_run(run.id))

    assert run.prompt == "Build a CLI"
    assert not hasattr(run, "token")
    assert view.criteria[0].text == "pytest must pass"
    assert view.events[0].message == "Built CLI."


def test_run_message_route_records_user_message(monkeypatch) -> None:
    from arena.service import ArenaService

    runtime = FakeSwarmRuntime()
    monkeypatch.setattr(
        api,
        "service",
        ArenaService(
            session_factory=lambda: FakeSession(),
            run_store=InMemoryRunStore(),
            swarm_runtime=runtime,
            archive_store=LocalCodebaseArchiveStore(),
            validator_graph=ValidatorGraph(executor=DeterministicValidatorExecutor()),
        ),
    )

    run = run_async(create_run(RunRequest(prompt="Build a CLI")))
    updated = run_async(add_run_message(run.id, RunMessageRequest(content="Keep going")))

    assert "token" not in updated.model_dump()
    assert updated.events[-1].message == "User: Keep going"
    assert runtime.touched_runs == [run.id]


def test_run_zip_route_returns_download_response(monkeypatch) -> None:
    from arena.service import ArenaService, ZipPayload

    class FakeService(ArenaService):
        def __init__(self) -> None:
            pass

        def get_run_codebase_zip(self, run_id: str) -> ZipPayload:
            return ZipPayload(filename=f"{run_id}.zip", data=b"PKzip")

    monkeypatch.setattr(api, "service", FakeService())

    response = run_async(get_run_codebase_zip("run-1"))

    assert response.media_type == "application/zip"
    assert response.headers["content-disposition"] == 'attachment; filename="run-1.zip"'


def test_run_field_notes_and_trace_routes_return_downloads(monkeypatch) -> None:
    from arena.service import ArenaService, TextPayload

    class FakeService(ArenaService):
        def __init__(self) -> None:
            pass

        def get_run_field_notes(self, run_id: str) -> TextPayload:
            return TextPayload(filename=f"{run_id}-field-notes.md", content_type="text/markdown", data="# notes")

        def get_run_trace_json(self, run_id: str) -> TextPayload:
            return TextPayload(filename=f"{run_id}-trace.json", content_type="application/json", data="{}")

        def get_run_submission_markdown(self, run_id: str) -> TextPayload:
            return TextPayload(filename=f"{run_id}-submission.md", content_type="text/markdown", data="# submit")

        def get_run_space_readme(self, run_id: str) -> TextPayload:
            return TextPayload(filename=f"{run_id}-space-README.md", content_type="text/markdown", data="# readme")

    monkeypatch.setattr(api, "service", FakeService())

    notes = run_async(get_run_field_notes("run-1"))
    trace = run_async(get_run_trace_json("run-1"))
    submission = run_async(get_run_submission_markdown("run-1"))
    space_readme = run_async(get_run_space_readme("run-1"))

    assert notes.media_type == "text/markdown"
    assert notes.headers["content-disposition"] == 'attachment; filename="run-1-field-notes.md"'
    assert trace.media_type == "application/json"
    assert trace.headers["content-disposition"] == 'attachment; filename="run-1-trace.json"'
    assert submission.media_type == "text/markdown"
    assert submission.headers["content-disposition"] == 'attachment; filename="run-1-submission.md"'
    assert space_readme.media_type == "text/markdown"
    assert space_readme.headers["content-disposition"] == 'attachment; filename="run-1-space-README.md"'


def test_validate_run_route_returns_validation_report(monkeypatch) -> None:
    from arena.service import ArenaService

    class FakeService(ArenaService):
        def __init__(self) -> None:
            pass

        def validate_run(self, run_id: str) -> ValidationReport:
            return ValidationReport(run_id=run_id, passed=True, summary="ok")

    monkeypatch.setattr(api, "service", FakeService())

    report = run_async(validate_run("run-1"))

    assert report.run_id == "run-1"
    assert report.passed is True


def test_model_catalog_route_uses_service(monkeypatch) -> None:
    from arena.service import ArenaService

    class FakeService(ArenaService):
        def __init__(self) -> None:
            pass

        def list_provider_models(self, config: ModelProviderConfig | None = None) -> list[ProviderModel]:
            assert config is not None
            assert config.provider == "openrouter"
            return [ProviderModel(id="qwen/model", label="Qwen", provider="openrouter")]

    monkeypatch.setattr(api, "service", FakeService())

    models = run_async(list_provider_models(ModelProviderConfig(provider="openrouter")))

    assert models[0].id == "qwen/model"


def test_model_catalog_refresh_route_uses_service(monkeypatch) -> None:
    from arena.service import ArenaService

    class FakeService(ArenaService):
        def __init__(self) -> None:
            pass

        def refresh_provider_models(self, config: ModelProviderConfig | None = None) -> list[ProviderModel]:
            assert config is not None
            assert config.provider == "openrouter"
            return [ProviderModel(id="qwen/qwen2.5-7b", provider="openrouter")]

    monkeypatch.setattr(api, "service", FakeService())

    models = run_async(refresh_provider_models(ModelProviderConfig(provider="openrouter")))

    assert models[0].id == "qwen/qwen2.5-7b"


def test_thread_routes_use_standalone_inspector_not_service(monkeypatch) -> None:
    """Thread routes must call the module-level thread_inspector, not service."""
    from arena.service import ArenaService
    from arena.thread_inspector import ThreadResponse

    class CallCounter(ArenaService):
        def __init__(self) -> None:
            self.create_thread_calls = 0

        def create_thread(self):
            self.create_thread_calls += 1
            return ThreadResponse(thread_id="from-service", sandbox_provider="local")

    counter = CallCounter()
    monkeypatch.setattr(api, "service", counter)
    monkeypatch.setattr(api.thread_inspector, "create_thread", lambda: ThreadResponse(thread_id="from-inspector", sandbox_provider="local"))

    response = run_async(create_thread())

    assert response.thread_id == "from-inspector"
    assert counter.create_thread_calls == 0