File size: 1,961 Bytes
47428ef a1e0c1e 47428ef a1e0c1e 47428ef | 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 | from __future__ import annotations
import unittest
from app import app as deployed_app
from server.app import (
StepRequest,
app,
health,
metadata,
reset,
schema,
state,
step,
)
class ServerApiTest(unittest.TestCase):
def test_deployment_entrypoint_exposes_openenv_routes(self) -> None:
route_paths = {route.path for route in deployed_app.routes}
self.assertIn("/reset", route_paths)
self.assertIn("/step", route_paths)
self.assertIn("/state", route_paths)
self.assertIn("/health", route_paths)
self.assertIn("/metadata", route_paths)
self.assertIn("/schema", route_paths)
def test_required_routes_are_registered(self) -> None:
route_paths = {route.path for route in app.routes}
self.assertIn("/health", route_paths)
self.assertIn("/metadata", route_paths)
self.assertIn("/schema", route_paths)
self.assertIn("/reset", route_paths)
self.assertIn("/step", route_paths)
self.assertIn("/state", route_paths)
self.assertIn("/ws", route_paths)
def test_handlers_return_openenv_shaped_payloads(self) -> None:
self.assertEqual(health().status, "healthy")
self.assertEqual(metadata().name, "support-ops-env")
self.assertIn("action_type", schema().action["properties"])
reset_response = reset()
self.assertEqual(reset_response.observation["task_id"], "easy_account_takeover")
self.assertFalse(reset_response.done)
state_response = state()
self.assertEqual(state_response["task_id"], "easy_account_takeover")
self.assertIn("episode_id", state_response)
step_response = step(StepRequest(action={"action_type": "inspect_ticket", "target": "T1"}))
self.assertIn("observation", step_response.model_dump())
self.assertIsInstance(step_response.reward, float)
if __name__ == "__main__":
unittest.main()
|