tests_python/test_service.py

from arena.codebase_archive import LocalCodebaseArchiveStore
import pytest

from arena.models import Criteria, ModelProviderConfig, ValidationCheck
from arena.run_models import RunRequest
from arena.run_store import InMemoryRunStore
from arena.service import ArenaService, ArenaServiceError
from arena.swarm_runtime import SwarmRuntime
from arena.validator_graph import DeterministicValidatorExecutor, ValidatorGraph

from conftest import make_local_session


def make_service(archive_store=None):
    store = archive_store or LocalCodebaseArchiveStore()
    return ArenaService(
        session_factory=make_local_session,
        run_store=InMemoryRunStore(),
        swarm_runtime=SwarmRuntime(
            session_factory=make_local_session,
            archive_store=store,
        ),
        validator_graph=ValidatorGraph(
            executor=DeterministicValidatorExecutor(),
            rubric_enabled=False,
        ),
        archive_store=store,
    )


def test_service_creates_run_and_starts_swarm() -> None:
    service = make_service()

    access = service.create_run(
        RunRequest(
            prompt="Build a small FastAPI app",
            criteria=[Criteria(text="pytest must pass")],
            user_tests=["echo ok"],
        )
    )

    assert access.token
    assert access.run.prompt == "Build a small FastAPI app"
    assert access.run.criteria[0].text == "pytest must pass"

    view = service.get_run(access.run.id)
    assert view.status == "completed"
    assert view.tasks[0].status == "completed"
    assert "Built the app" in view.summary
    assert view.codebase_archive is not None
    assert view.validation_report is not None


def test_service_accepts_model_provider_when_model_is_within_limit() -> None:
    service = make_service()

    access = service.create_run(
        RunRequest(
            prompt="Build a CLI",
            user_tests=[],
            model_provider=ModelProviderConfig(provider="custom", model="qwen2.5-7b-instruct"),
        )
    )

    assert access.run.model_provider is not None
    assert access.run.model_provider.model == "qwen2.5-7b-instruct"


def test_service_rejects_model_provider_when_size_unknown() -> None:
    service = make_service()

    with pytest.raises(ArenaServiceError, match="unknown"):
        service.create_run(
            RunRequest(
                prompt="Build a CLI",
                model_provider=ModelProviderConfig(provider="gemini", model="gemini-2.5-flash"),
            )
        )


def test_service_rejects_model_provider_over_32b() -> None:
    service = make_service()

    with pytest.raises(ArenaServiceError, match="<=32B"):
        service.create_run(
            RunRequest(
                prompt="Build a CLI",
                model_provider=ModelProviderConfig(provider="openrouter", model="llama-3.1-70b-instruct"),
            )
        )


def test_service_rejoins_run_by_token() -> None:
    service = make_service()
    access = service.create_run(RunRequest(prompt="Build a CLI", user_tests=[]))

    rejoined = service.rejoin_run(access.token)

    assert rejoined.run.id == access.run.id
    assert service.get_run(access.run.id).id == access.run.id


def test_service_add_run_message_records_user_activity() -> None:
    service = make_service()
    access = service.create_run(RunRequest(prompt="Build a CLI", user_tests=[]))

    view = service.add_run_message(access.run.id, "Please keep the sandbox warm")

    assert view.events[-1].message == "User: Please keep the sandbox warm"


def test_service_downloads_codebase_zip() -> None:
    service = make_service()
    access = service.create_run(RunRequest(prompt="Build a CLI", user_tests=[]))

    payload = service.get_run_codebase_zip(access.run.id)

    assert payload.filename.endswith(".zip")
    assert payload.content_type == "application/zip"
    assert payload.data.startswith(b"PK")


def test_service_exports_field_notes_and_trace() -> None:
    service = make_service()
    access = service.create_run(RunRequest(prompt="Build a CLI", user_tests=["echo ok"]))

    notes = service.get_run_field_notes(access.run.id)
    trace = service.get_run_trace_json(access.run.id)
    submission = service.get_run_submission_markdown(access.run.id)
    space_readme = service.get_run_space_readme(access.run.id)

    assert notes.filename.endswith("-field-notes.md")
    assert "Mom Test Questions" in notes.data
    assert "Scrap or scale" in notes.data
    assert trace.filename.endswith("-trace.json")
    assert '"prompt": "Build a CLI"' in trace.data
    assert submission.filename.endswith("-submission.md")
    assert "Build Small Hackathon Chapter 1" in submission.data
    assert "port 7860" in submission.data
    assert space_readme.filename.endswith("-space-README.md")
    assert "sdk: gradio" in space_readme.data
    assert "7860" in space_readme.data


def test_service_validates_and_marks_completed() -> None:
    service = make_service()
    access = service.create_run(
        RunRequest(prompt="Build a CLI", user_tests=["echo ok"])
    )

    report = service.validate_run(access.run.id)
    view = service.get_run(access.run.id)

    assert report.passed is True
    assert view.status == "completed"
    assert view.validation_report is not None
    assert view.validation_report.passed is True
    assert view.validation_checks == [
        ValidationCheck(id="user-test-1", name="User test 1", command="echo ok", source="user")
    ]


def test_arena_service_does_not_expose_thread_inspection_methods() -> None:
    service = make_service()

    forbidden = [
        "create_thread",
        "send_thread_message",
        "list_thread_files",
        "read_thread_file",
        "delete_thread",
        "sessions",
    ]
    leaked = [name for name in forbidden if hasattr(service, name)]

    assert leaked == [], f"ArenaService leaked Thread surface: {leaked}"