Spaces:
Sleeping
Sleeping
File size: 5,891 Bytes
6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 14170d7 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 8c359c3 14170d7 775befb | 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 | """Tests for OpenEnv server/client integration (dispatch domain)."""
from __future__ import annotations
import asyncio
import pytest
from fastapi.testclient import TestClient
import src.server.app as server_app
from src.models import Action, DispatchAction
from src.openenv_environment import OpenEnvEnvironment
@pytest.fixture(autouse=True)
def reset_env() -> None:
server_app._env = None
yield
server_app._env = None
class TestOpenEnvEnvironment:
def test_reset_and_state(self) -> None:
env = OpenEnvEnvironment(task_id="single_incident", seed=42)
obs = asyncio.run(env.reset())
assert obs.result == "dispatch center online"
assert obs.protocol_ok is True
st = env.state()
assert st.task_id == "single_incident"
assert st.step_count == 0
def test_step_returns_tuple(self) -> None:
env = OpenEnvEnvironment(task_id="single_incident", seed=42)
asyncio.run(env.reset())
action = Action(
action_type=DispatchAction.DISPATCH,
unit_id="MED-1",
incident_id="INC-001",
)
obs, reward, done = asyncio.run(env.step(action))
assert isinstance(obs.result, str)
assert isinstance(reward, float)
assert isinstance(done, bool)
env.close()
class TestResetEndpoint:
def test_reset_returns_observation(self) -> None:
c = TestClient(server_app.app)
response = c.post("/reset", json={"task_id": "single_incident", "seed": 42})
assert response.status_code == 200
data = response.json()
assert data["result"] == "dispatch center online"
assert data["protocol_ok"] is True
def test_reset_with_empty_body_returns_200(self) -> None:
"""Verify prevalidation.sh compatible: POST /reset with {} returns 200."""
c = TestClient(server_app.app)
response = c.post("/reset", json={})
assert response.status_code == 200
data = response.json()
assert data["result"] == "dispatch center online"
class TestStepEndpoint:
def test_step_requires_reset_first(self) -> None:
c = TestClient(server_app.app)
response = c.post(
"/step",
json={
"action": {
"action_type": "DISPATCH",
"unit_id": "MED-1",
"incident_id": "INC-001",
}
},
)
assert response.status_code == 500
assert "not initialized" in response.json()["detail"].lower()
def test_step_invalid_action_rejected(self) -> None:
c = TestClient(server_app.app)
c.post("/reset", json={"task_id": "single_incident", "seed": 42})
response = c.post("/step", json={"action": {"invalid": "field"}})
assert response.status_code == 500
assert "invalid action" in response.json()["detail"].lower()
def test_step_ok(self) -> None:
c = TestClient(server_app.app)
c.post("/reset", json={"task_id": "single_incident", "seed": 42})
response = c.post(
"/step",
json={
"action": {
"action_type": "DISPATCH",
"unit_id": "MED-1",
"incident_id": "INC-001",
}
},
)
assert response.status_code == 200
data = response.json()
assert set(data.keys()) == {"observation", "reward", "done"}
class TestStateEndpoint:
def test_state_requires_reset_first(self) -> None:
c = TestClient(server_app.app)
response = c.get("/state")
assert response.status_code == 500
def test_state_returns_current_state(self) -> None:
c = TestClient(server_app.app)
c.post("/reset", json={"task_id": "single_incident", "seed": 42})
response = c.get("/state")
assert response.status_code == 200
data = response.json()
assert data["task_id"] == "single_incident"
assert data["step_count"] == 0
class TestHealthEndpoint:
def test_health_ok(self) -> None:
c = TestClient(server_app.app)
response = c.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "healthy"}
class TestTasksEndpoint:
def test_tasks_endpoint_returns_four_tasks(self) -> None:
c = TestClient(server_app.app)
response = c.get("/tasks")
assert response.status_code == 200
tasks = response.json()
assert len(tasks) == 4
task_ids = {t["task_id"] for t in tasks}
assert task_ids == {
"single_incident",
"multi_incident",
"mass_casualty",
"shift_surge",
}
class TestDashboardEndpoint:
def test_dashboard_state_before_reset_returns_valid_shape(self) -> None:
c = TestClient(server_app.app)
response = c.get("/dashboard/state")
assert response.status_code == 200
data = response.json()
assert data["task_id"] == "none"
assert data["step_count"] == 0
assert isinstance(data["units"], dict)
assert isinstance(data["incidents"], dict)
assert isinstance(data["legal_actions"], list)
assert isinstance(data["issues"], list)
assert data["observation"] is None
def test_dashboard_state_after_reset_exposes_legal_actions(self) -> None:
c = TestClient(server_app.app)
reset_response = c.post("/reset", json={"task_id": "single_incident", "seed": 42})
assert reset_response.status_code == 200
response = c.get("/dashboard/state")
assert response.status_code == 200
data = response.json()
assert data["task_id"] == "single_incident"
assert isinstance(data["legal_actions"], list)
assert data["observation"] is not None
|