Spaces:
Sleeping
Sleeping
File size: 5,353 Bytes
f8f1ce8 426093b f8f1ce8 93cd78f f8f1ce8 93cd78f f8f1ce8 93cd78f f8f1ce8 93cd78f f8f1ce8 93cd78f f8f1ce8 e6ab3ac 93cd78f e6ab3ac f8f1ce8 93cd78f f8f1ce8 642444a | 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 | """Route-level tests: call handle_request directly (no socket, no thread)."""
from __future__ import annotations
import json
import proteus.game.scenarios # noqa: F401
from proteus.web.local import server
def _registry():
return {}
def test_config_lists_scenarios_difficulties_and_color_map():
reg = _registry()
status, payload, ctype = server.handle_request("GET", "/config", None, reg)
assert status == 200 and ctype == "application/json"
assert "template" in payload["scenarios"]
assert payload["difficulties"] == ["easy", "medium", "hard", "expert"]
# color_map keys are stringified ints -> hex.
assert payload["color_map"]["0"].startswith("#")
assert payload["default_seed"] == 42
def test_root_serves_html_bytes():
status, payload, ctype = server.handle_request("GET", "/", None, _registry())
assert status == 200 and ctype == "text/html; charset=utf-8"
assert isinstance(payload, bytes) and b"<" in payload
def test_create_session_returns_id_and_fair_state():
reg = _registry()
body = {"scenario": "template", "difficulty": "easy", "seed": 42,
"play_turns": 5, "probe": False}
status, payload, _ = server.handle_request("POST", "/session", body, reg)
assert status == 200
sid = payload["session_id"]
assert sid in reg
st = payload["state"]
assert st["phase"] == "cut_intro"
assert isinstance(st["grid"][0][0], int)
# fairness: no reward/optimal in the live state.
assert "reward" not in json.dumps(st)
assert "motive_action" not in json.dumps(st)
def test_act_advances_and_unknown_action_is_400():
reg = _registry()
_, created, _ = server.handle_request(
"POST", "/session",
{"scenario": "template", "difficulty": "easy", "seed": 42,
"play_turns": 5, "probe": False}, reg)
sid = created["session_id"]
status, payload, _ = server.handle_request(
"POST", f"/session/{sid}/act", {"action": "up"}, reg)
assert status == 200 and payload["state"]["turn_idx"] == 1
status, payload, _ = server.handle_request(
"POST", f"/session/{sid}/act", {"action": "fly"}, reg)
assert status == 400 and "error" in payload
def test_unknown_session_is_404():
status, payload, _ = server.handle_request(
"GET", "/session/does-not-exist", None, _registry())
assert status == 404 and "error" in payload
def test_unknown_scenario_is_400():
status, payload, _ = server.handle_request(
"POST", "/session",
{"scenario": "nope", "difficulty": "easy", "seed": 1,
"play_turns": 5, "probe": False}, _registry())
assert status == 400 and "error" in payload
def test_non_numeric_seed_is_400_not_500():
# A malformed seed must be a structured 400, never an unhandled 500.
status, payload, _ = server.handle_request(
"POST", "/session",
{"scenario": "template", "difficulty": "easy", "seed": "abc",
"play_turns": 5, "probe": False}, _registry())
assert status == 400 and "error" in payload
def test_query_string_is_ignored_in_routing():
# /config with a query string must still route to config, not fall to 404.
status, payload, ctype = server.handle_request(
"GET", "/config?cachebust=1", None, _registry())
assert status == 200 and "template" in payload["scenarios"]
def test_get_poll_state_is_fair_mid_game():
# The GET poll path must be as fair as create/act: no answer keys mid-game.
reg = _registry()
_, created, _ = server.handle_request(
"POST", "/session",
{"scenario": "template", "difficulty": "easy", "seed": 42,
"play_turns": 5, "probe": False}, reg)
sid = created["session_id"]
server.handle_request("POST", f"/session/{sid}/act", {"action": "up"}, reg)
status, payload, _ = server.handle_request("GET", f"/session/{sid}", None, reg)
assert status == 200
st = payload["state"]
assert st["outcome"] is None and st["review"] is None
blob = json.dumps(st)
assert "reward" not in blob and "motive_action" not in blob and "habit" not in blob
def test_finish_appends_trace_and_returns_metrics(tmp_path):
reg = _registry()
_, created, _ = server.handle_request(
"POST", "/session",
{"scenario": "template", "difficulty": "easy", "seed": 42,
"play_turns": 3, "probe": False}, reg)
sid = created["session_id"]
# Exhaust the budget.
for _ in range(3):
status, payload, _ = server.handle_request(
"POST", f"/session/{sid}/act", {"action": "up"}, reg)
if payload["state"]["outcome"] is not None:
break
out = tmp_path / "web.jsonl"
status, payload, _ = server.handle_request(
"POST", f"/session/{sid}/finish", {"out": str(out)}, reg)
assert status == 200
assert payload["trace_path"] == str(out)
assert "survival_fraction" in payload["metrics"]
assert out.exists() and out.read_text().strip() # one JSONL line written
def test_make_server_binds_and_has_router(tmp_path):
srv = server.make_server("127.0.0.1", 0) # port 0 = ephemeral, no serve_forever
try:
assert srv.server_address[0] == "127.0.0.1"
# the bound handler carries its own fresh registry
assert isinstance(srv.RequestHandlerClass.registry, dict)
finally:
srv.server_close()
|