Spaces:
Running
Running
File size: 1,790 Bytes
877add7 | 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 | import json
from fastapi.testclient import TestClient
from app.env.fastapi_app import app
def test_health_endpoint_healthy() -> None:
client = TestClient(app)
response = client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
def test_metadata_endpoint_available() -> None:
client = TestClient(app)
response = client.get("/metadata")
assert response.status_code == 200
payload = response.json()
assert payload["name"] == "polyguard-openenv"
assert isinstance(payload["description"], str)
assert payload["reward_range"] == [0.001, 0.999]
def test_schema_and_mcp_endpoints_available() -> None:
client = TestClient(app)
schema = client.get("/schema")
assert schema.status_code == 200
schema_payload = schema.json()
assert "action" in schema_payload
assert "observation" in schema_payload
assert "state" in schema_payload
mcp = client.post("/mcp", json={})
assert mcp.status_code == 200
mcp_payload = mcp.json()
assert mcp_payload["jsonrpc"] == "2.0"
def test_websocket_reset_and_step_roundtrip() -> None:
client = TestClient(app)
with client.websocket_connect("/ws") as ws:
ws.send_text(json.dumps({"type": "reset", "data": {"seed": 7, "difficulty": "easy"}}))
reset_message = ws.receive_json()
assert reset_message["type"] == "result"
assert "observation" in reset_message["data"]
legal = client.get("/env/legal_actions").json()
ws.send_text(json.dumps({"type": "step", "data": legal[0]}))
step_message = ws.receive_json()
assert step_message["type"] == "result"
assert 0.001 <= step_message["data"]["reward"] <= 0.999
assert "done" in step_message["data"]
|