Spaces:
Sleeping
Sleeping
| """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() | |