File size: 2,507 Bytes
4fbc241 9b64226 4fbc241 9b64226 4fbc241 1d6826f 9b64226 4fbc241 | 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 | from __future__ import annotations
import asyncio
import os
import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
from server.app import create_application, shared_env
def _route_map():
app = create_application(enable_web=False)
return {getattr(route, "path", None): route.endpoint for route in app.routes}
def _call(endpoint, *args, **kwargs):
result = endpoint(*args, **kwargs)
if asyncio.iscoroutine(result):
return asyncio.run(result)
return result
def test_required_routes_registered() -> None:
routes = _route_map()
for path in ["/health", "/metadata", "/schema", "/reset", "/step", "/state", "/tasks", "/grader", "/baseline", "/demo"]:
assert path in routes
def test_session_routes_are_not_duplicated() -> None:
app = create_application(enable_web=False)
paths = [getattr(route, "path", None) for route in app.routes]
assert paths.count("/reset") == 1
assert paths.count("/step") == 1
assert paths.count("/state") == 1
def test_reset_endpoint_accepts_empty_json_body() -> None:
os.environ.setdefault("MPLCONFIGDIR", "/tmp")
client = TestClient(create_application(enable_web=False))
response = client.post("/reset", json={})
assert response.status_code == 200
payload = response.json()
assert payload["observation"]["task_id"] == "static_workload"
assert payload["observation"]["step_index"] == 0
def test_health_endpoint_direct() -> None:
data = _call(_route_map()["/health"])
status = data["status"] if isinstance(data, dict) else data.status
assert status in {"ok", "healthy"}
def test_tasks_endpoint_direct() -> None:
data = _call(_route_map()["/tasks"])
assert len(data["tasks"]) == 3
assert "batch_cap" in data["action_schema"]
def test_grader_requires_episode_before_grading() -> None:
shared_env.actions.clear()
shared_env.observations.clear()
shared_env.rewards.clear()
with pytest.raises(HTTPException):
_call(_route_map()["/grader"])
def test_baseline_endpoint_direct() -> None:
data = _call(_route_map()["/baseline"], task_id="static_workload", use_openai=False, model="gpt-4.1-mini")
assert data["mode"] == "deterministic"
assert "static_workload" in data["baseline"]
assert data["baseline"]["static_workload"]["grader"]["score"] >= 0.25
def test_demo_redirects_to_web() -> None:
response = _call(_route_map()["/demo"])
assert response.headers["location"] == "/web"
|