""" test_server.py — headless tests for the turn server (session + render_state + app routes). Run: /Users/geekwrestler/railtycoon/prodcreation/.venv/bin/python frontend/server/test_server.py Covers: 1. FakeLLM full 20-round game, every RenderState jsonschema-validated 2. determinism: two sessions, same seed, model off -> identical RenderState dicts 3. play_card rejections: energy / locked card / occupied station / per-round cap 4. visual flags flip: is_night @ round 7, rain while a flood is active OR through peak, is_peak @ 16 (+ peak/win notifications) 5. HTTP endpoints via TestClient: schema 400s, /turn, /model, session isolation 6. optional real-model smoke turn if llama.cpp answers on :8092 """ from __future__ import annotations import copy import json import os import sys import time _SERVER_DIR = os.path.dirname(os.path.abspath(__file__)) _BACKEND_DIR = os.path.abspath(os.path.join(_SERVER_DIR, "..", "..", "backend")) for _p in (_BACKEND_DIR, _SERVER_DIR): if _p not in sys.path: sys.path.insert(0, _p) import jsonschema from llm import ping from session import GameSession _ROOT = os.path.abspath(os.path.join(_SERVER_DIR, "..", "..")) with open(os.path.join(_ROOT, "claude-handoff", "schemas", "render_state.schema.json"), encoding="utf-8") as f: RENDER_STATE_SCHEMA = json.load(f) _validator = jsonschema.Draft7Validator(RENDER_STATE_SCHEMA) VALIDATED = 0 FAILURES: list[str] = [] def validate(state: dict, where: str) -> None: global VALIDATED errs = sorted(_validator.iter_errors(state), key=str) if errs: FAILURES.append(f"{where}: schema violation: {errs[0].message}") VALIDATED += 1 def check(cond: bool, msg: str) -> None: if not cond: FAILURES.append(msg) class FakeLLM: """TEST-ONLY stub for llm.LLM — returns one fixed JSON reply instantly so the dispatcher pipeline (repair ladder, legality, budget) can be driven offline. This is test plumbing, NOT game logic and NOT a fallback policy.""" name = "fake" def complete(self, system, user): return ('{"priority":"test","selected_actions":["a1"],"announcement":"",' '"confidence":0.5}', 0.0) def fake_session(seed: int) -> GameSession: s = GameSession(seed=seed, model_on=True) s.disp.llm = FakeLLM() # swap the transport only; dispatcher logic untouched return s # --------------------------------------------------------------------------- 1. full game def test_full_game_fake_llm(): s = fake_session(seed=42) validate(s.last_state, "fakegame initial") rounds = 0 while not s.g.over and rounds < 30: st = s.next_turn() rounds += 1 validate(st, f"fakegame round {st['round']}") check(rounds == 20, f"fakegame: expected 20 rounds, got {rounds}") check(st["over"] is True and st["won"] is True, f"fakegame: over/won wrong: {st['reason']}") check(any(n["kind"] == "win" for n in st["notifications"]), "fakegame: no win notification") check(st["ai"]["priority"] == "test" and st["ai"]["confidence"] == 0.5, f"fakegame: ai telemetry wrong: {st['ai']}") check(st["ai"]["actions"] and st["ai"]["noop"] is False, "fakegame: ai actions missing") print(f" [1] FakeLLM full game: {rounds} rounds, won={st['won']} ({st['reason']!r})") # --------------------------------------------------------------------------- 2. determinism def test_determinism(): a = GameSession(seed=123, model_on=False) b = GameSession(seed=123, model_on=False) check(a.last_state == b.last_state, "determinism: initial states differ") validate(a.last_state, "determinism initial") for i in range(3): sa, sb = a.next_turn(), b.next_turn() validate(sa, f"determinism turn {i+1}") check(sa == sb, f"determinism: states differ after turn {i+1}") check(sa["ai"]["noop"] is True and sa["ai"]["priority"] == "(dispatcher offline)", f"determinism: model-off ai wrong: {sa['ai']}") print(" [2] determinism: 2 sessions x 3 model-off turns -> identical RenderStates") # --------------------------------------------------------------------------- 3. rejections def test_play_card_rejections(): s = GameSession(seed=7, model_on=False) ok, why = s.play_card("vip_special", "Dadar") # costs 12, start energy 10 check((ok, why) == (False, "no energy"), f"reject energy: got {(ok, why)}") ok, why = s.play_card("monsoon_flood", "Dadar") # locked until R8 check((ok, why) == (False, "unavailable"), f"reject locked: got {(ok, why)}") lc = {c["card"]: c for c in s.last_state["legal_cards"]} check(lc["monsoon_flood"]["reason"] == "locked until R8", f"reject locked reason: {lc['monsoon_flood']}") ok, why = s.play_card("track_cow", "Dadar") check((ok, why) == (True, "ok"), f"stage cow: got {(ok, why)}") ok, why = s.play_card("signal_failure", "Andheri") # cap 1/round in normal phase check((ok, why) == (False, "card limit this round"), f"reject cap: got {(ok, why)}") st = s.next_turn() validate(st, "rejections turn 1") check(st["chaos_last"] == [{"card": "track_cow", "location": "Dadar"}], f"chaos_last wrong: {st['chaos_last']}") check(any(n["kind"] == "chaos" and n["station"] == "Dadar" for n in st["notifications"]), "chaos notification missing") check(any(i["type"] == "track_cow" and i["location"] == "Dadar" for i in st["incidents"]), "cow incident missing after commit") ok, why = s.play_card("signal_failure", "Dadar") # cow still active @ Dadar check((ok, why) == (False, "station busy"), f"reject occupied: got {(ok, why)}") st = s.next_turn() # cow (duration 2) expires validate(st, "rejections turn 2") check(any(n["kind"] == "resolve" and n["station"] == "Dadar" for n in st["notifications"]), "resolve notification missing after cow expiry") print(" [3] play_card rejections: energy / locked / per-round cap / occupied station") # --------------------------------------------------------------------------- 4. visual flags def test_visual_flags(): s = GameSession(seed=5, model_on=False) by_round = {1: s.last_state} while not s.g.over: st = s.next_turn() by_round[st["round"]] = st validate(st, f"flags round {st['round']}") check(by_round[6]["visual"]["is_night"] is False, "is_night true too early (round 6)") check(by_round[7]["visual"]["is_night"] is True, "is_night not set at round 7") check(all(st["visual"]["rain"] is False for r, st in by_round.items() if r < 16), "rain set before peak without a monsoon_flood incident") check(all(st["visual"]["rain"] is True for r, st in by_round.items() if r >= 16), "rain not persistent through peak (rounds 16+)") check(by_round[15]["is_peak"] is False, "is_peak true too early (round 15)") check(by_round[16]["is_peak"] is True, "is_peak not set at round 16") check(by_round[8]["legal_cards"][3]["card"] == "monsoon_flood" and by_round[8]["legal_cards"][3]["available"] is True, "monsoon_flood not unlocked at round 8") peaks = [n for st in by_round.values() for n in st["notifications"] if n["kind"] == "peak"] check(len(peaks) == 1, f"expected exactly one peak notification, got {len(peaks)}") banner = next(st for st in by_round.values() if st["phase"] == "peak") check(banner["visual"]["peak_banner"] != "", "peak_banner empty during peak phase") # rain follows the flood card: on while a monsoon_flood incident is alive, off after it expires s2 = GameSession(seed=9, model_on=False) for _ in range(7): # advance to round 8 (flood unlock) s2.next_turn() ok, why = s2.play_card("monsoon_flood", "Bandra") check((ok, why) == (True, "ok"), f"flood stage failed at R8: {(ok, why)}") st = s2.next_turn() validate(st, "rain flood turn") check(st["visual"]["rain"] is True, "rain not on while flood incident active") rains = [st["visual"]["rain"]] while any(i["type"] == "monsoon_flood" for i in st["incidents"]) and not s2.g.over: st = s2.next_turn() rains.append(st["visual"]["rain"]) check(rains[-1] is False, "rain still on after flood incident expired") print(" [4] visual flags: is_night@7, rain=flood-or-peak, is_peak@16, banner on") # --------------------------------------------------------------------------- 5. HTTP def test_http(): from fastapi.testclient import TestClient import app as app_module client = TestClient(app_module.app) h1, h2 = {"X-Session-Id": "t1"}, {"X-Session-Id": "t2"} r = client.post("/turn", json={"action": "bogus"}) check(r.status_code == 400, f"/turn bad action: expected 400, got {r.status_code}") r = client.post("/turn", json={"action": "play_card", "card": "track_cow"}) check(r.status_code == 400, f"/turn missing location: expected 400, got {r.status_code}") r = client.post("/model", json={"model": "not-a-model"}, headers=h1) check(r.status_code == 400, f"/model unknown model: expected 400, got {r.status_code}") r = client.post("/turn", json={"action": "new_game", "seed": 11}, headers=h1) check(r.status_code == 200, f"/turn new_game: {r.status_code}") validate(r.json(), "http new_game") check(r.json()["round"] == 1, "http new_game: round != 1") r = client.post("/model", json={"on": False}, headers=h1) check(r.status_code == 200 and r.json()["on"] is False, f"/model off failed: {r.text}") r = client.post("/turn", json={"action": "play_card", "card": "vip_special", "location": "Dadar"}, headers=h1) check(r.status_code == 409 and r.json()["error"] == "no energy", f"http reject energy: {r.status_code} {r.text}") r = client.post("/turn", json={"action": "play_card", "card": "track_cow", "location": "Dadar"}, headers=h1) check(r.status_code == 200, f"http play cow: {r.status_code} {r.text}") r = client.post("/turn", json={"action": "play_card", "card": "track_cow", "location": "Bandra"}, headers=h1) check(r.status_code == 409 and r.json()["error"] == "card limit this round", f"http reject cap: {r.status_code} {r.text}") r = client.post("/turn", json={"action": "next_turn"}, headers=h1) check(r.status_code == 200, f"http next_turn: {r.status_code}") st = r.json() validate(st, "http next_turn") check(st["round"] == 2 and st["chaos_last"] == [{"card": "track_cow", "location": "Dadar"}], f"http next_turn state wrong: round={st['round']} chaos_last={st['chaos_last']}") r = client.post("/turn", json={"action": "new_game", "seed": 11}, headers=h2) check(r.status_code == 200 and r.json()["round"] == 1, "http session t2 not fresh") r = client.post("/turn", json={"action": "new_game", "seed": None}, headers=h2) check(r.status_code == 200 and r.json()["round"] == 1, "http new_game null seed failed") r = client.get("/") check(r.status_code == 200 and "mumbai local" in r.text.lower(), "static / not served") print(" [5] HTTP: 400 on schema mismatch, 409 rejections, /model, isolation, static /") # --------------------------------------------------------------------------- 6. real model def test_real_model_smoke(): url = "http://localhost:8092" if not ping(url): print(" [6] real-model smoke: SKIPPED (no server on :8092 — see /tmp/llama_qwen.log)") return s = GameSession(seed=99, model_name="qwen3-8b", model_on=True) s.play_card("track_cow", "Dadar") # give the dispatcher something to react to t0 = time.perf_counter() st = s.next_turn() dt = (time.perf_counter() - t0) * 1000.0 validate(st, "real-model turn") ai = st["ai"] print(f" [6] real-model smoke (qwen3-8b): {dt:.0f} ms — noop={ai['noop']} " f"confidence={ai['confidence']} priority={ai['priority']!r} " f"actions={ai['actions']} announcement={ai['announcement']!r}") # --------------------------------------------------------------------------- runner def main() -> int: test_full_game_fake_llm() test_determinism() test_play_card_rejections() test_visual_flags() test_http() test_real_model_smoke() print(f"\n RenderStates schema-validated: {VALIDATED}") if FAILURES: print(f" FAILURES ({len(FAILURES)}):") for msg in FAILURES: print(f" - {msg}") return 1 print(" ALL TESTS PASSED") return 0 if __name__ == "__main__": sys.exit(main())