File size: 5,935 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 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | """Integration tests: real HTTP layer with real sandbox, stores, and validation."""
import asyncio
import httpx
import pytest
from arena.api import app, service
from arena.codebase_archive import LocalCodebaseArchiveStore
from arena.run_models import RunRequest
from arena.run_store import InMemoryRunStore
from arena.swarm_runtime import SwarmRuntime
from arena.validator_graph import DeterministicValidatorExecutor, ValidatorGraph
from conftest import make_local_session
def request(method: str, path: str, **kwargs) -> httpx.Response:
async def _request() -> httpx.Response:
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
return await client.request(method, path, **kwargs)
return asyncio.run(_request())
@pytest.fixture(autouse=True)
def _reset_service() -> None:
from arena.event_bus import EventBus
store = LocalCodebaseArchiveStore()
event_bus = EventBus()
service._run_store = InMemoryRunStore()
service._archive_store = store
service._swarm_runtime = SwarmRuntime(
session_factory=make_local_session,
archive_store=store,
event_bus=event_bus,
)
service._validator_graph = ValidatorGraph(
executor=DeterministicValidatorExecutor(),
rubric_enabled=False,
)
from arena.run_flow import RunFlow
service._run_flow = RunFlow(
run_store=service._run_store,
swarm_runtime=service._swarm_runtime,
validator_graph=service._validator_graph,
archive_store=service._archive_store,
event_bus=event_bus,
)
def test_health_endpoint_returns_ok() -> None:
response = request("GET", "/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_skills_endpoint_returns_list() -> None:
response = request("GET", "/skills")
assert response.status_code == 200
skills = response.json()
assert isinstance(skills, list)
def test_create_run_and_get_status() -> None:
response = request(
"POST",
"/runs",
json={"prompt": "Build a CLI", "user_tests": ["echo ok"]},
)
assert response.status_code == 200
data = response.json()
assert "token" not in data
run_id = data["id"]
status_resp = request("GET", f"/runs/{run_id}")
assert status_resp.status_code == 200
run = status_resp.json()
assert run["id"] == run_id
assert run["prompt"] == "Build a CLI"
assert run["status"] == "completed"
assert len(run["tasks"]) >= 1
assert run["codebase_archive"] is not None
assert run["validation_report"] is not None
def test_rejoin_run() -> None:
response = request(
"POST",
"/runs",
json={"prompt": "Build a CLI", "user_tests": []},
)
run_id = response.json()["id"]
access = service._run_store.create_run(RunRequest(prompt="Private rejoin"))
rejoin_resp = request("POST", "/runs/rejoin", json={"token": access.token})
assert rejoin_resp.status_code == 200
assert "token" not in rejoin_resp.json()
assert rejoin_resp.json()["id"] == access.run.id
assert rejoin_resp.json()["id"] != run_id
def test_validate_run() -> None:
response = request(
"POST",
"/runs",
json={"prompt": "Build a CLI", "user_tests": ["echo ok"]},
)
run_id = response.json()["id"]
validate_resp = request("POST", f"/runs/{run_id}/validate")
assert validate_resp.status_code == 200
report = validate_resp.json()
assert report["passed"] is True
assert report["run_id"] == run_id
final = request("GET", f"/runs/{run_id}")
assert final.json()["status"] == "completed"
assert final.json()["validation_report"] is not None
def test_download_codebase_zip() -> None:
response = request(
"POST",
"/runs",
json={"prompt": "Build a CLI", "user_tests": []},
)
run_id = response.json()["id"]
zip_resp = request("GET", f"/runs/{run_id}/codebase.zip")
assert zip_resp.status_code == 200
assert zip_resp.headers["content-type"] == "application/zip"
assert len(zip_resp.content) > 0
assert zip_resp.content[:2] == b"PK"
def test_start_run_nonblocking() -> None:
import time
response = request(
"POST",
"/runs/start",
json={"prompt": "Build a CLI", "user_tests": ["echo ok"]},
)
assert response.status_code == 200
run_id = response.json()["id"]
assert response.json()["status"] in ("queued", "planning", "working", "validating", "completed")
time.sleep(0.2)
final = request("GET", f"/runs/{run_id}")
assert final.json()["status"] in ("working", "validating", "completed")
def test_cancel_run() -> None:
response = request(
"POST",
"/runs",
json={"prompt": "Build a CLI", "user_tests": []},
)
run_id = response.json()["id"]
cancel_resp = request("POST", f"/runs/{run_id}/cancel")
assert cancel_resp.status_code == 200
assert cancel_resp.json()["status"] == "completed"
def test_add_run_message() -> None:
response = request(
"POST",
"/runs",
json={"prompt": "Build a CLI", "user_tests": []},
)
run_id = response.json()["id"]
message_resp = request(
"POST",
f"/runs/{run_id}/messages",
json={"content": "Please keep this active"},
)
assert message_resp.status_code == 200
assert "token" not in message_resp.json()
assert message_resp.json()["events"][-1]["message"] == "User: Please keep this active"
def test_run_not_found_returns_404() -> None:
response = request("GET", "/runs/nonexistent")
assert response.status_code == 404
def test_rejoin_bad_token_returns_404() -> None:
response = request("POST", "/runs/rejoin", json={"token": "not-a-real-token"})
assert response.status_code == 404
|