File size: 5,887 Bytes
31c8075 | 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 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | 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}"
|