Spaces:
Sleeping
Sleeping
File size: 2,835 Bytes
7111c77 09f204f 7111c77 09f204f 7111c77 09f204f 7111c77 09f204f 7111c77 09f204f | 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 | from fastapi.testclient import TestClient
from server.app import app
def test_openenv_wrapper_exposes_schema_and_state():
client = TestClient(app)
assert client.get("/schema").status_code == 200
assert client.get("/state").status_code == 200
web_response = client.get("/web")
assert web_response.status_code == 200
assert "GPU Scheduler Console" in web_response.text
assert "Operate the benchmark scenario like a compact placement console" in web_response.text
def test_openenv_wrapper_accepts_wrapped_action_payload():
client = TestClient(app)
reset_response = client.post("/reset", json={})
assert reset_response.status_code == 200
step_response = client.post(
"/step",
json={
"action": {
"action": "defer",
"rationale": "Defer placement for a better window.",
"metadata": {},
}
},
)
assert step_response.status_code == 200
body = step_response.json()
assert body["observation"]["current_tick"] == 1
assert body["done"] is False
assert "summary" in body["observation"]
assert "feasible_assignments" in body["observation"]
assert "action_hint" in body["observation"]
assert body["observation"]["feasible_assignments"] == []
assert body["observation"]["suggested_assignments"] == []
def test_web_reset_exposes_readable_summary_fields():
client = TestClient(app)
body = client.post("/web/reset").json()
observation = body["observation"]
assert observation["summary"].startswith("Tick 0/")
assert observation["priority_alerts"] == []
assert observation["suggested_assignments"] == []
assert observation["action_hint"] == "Inspect pending jobs and eligible GPUs, then choose a placement or defer."
def test_web_state_contains_dashboard_fields():
client = TestClient(app)
client.post("/web/reset", json={})
body = client.get("/web/state").json()
assert "action_hint" in body
assert "suggested_assignments" in body
assert "pending_jobs" in body
assert body["suggested_assignments"] == []
def test_web_state_matches_web_reset_and_step_session():
client = TestClient(app)
client.post("/web/reset", json={"scenario_id": "deadline_crunch"})
for _ in range(2):
response = client.post(
"/web/step",
json={
"action": {
"action": "defer",
"rationale": "Inspect the current pending queue first.",
"metadata": {},
}
},
)
assert response.status_code == 200
body = client.get("/web/state").json()
assert body["scenario_id"] == "deadline_crunch"
assert body["current_tick"] == 2
assert body["summary"].startswith("Tick 2/")
|