frontend/server/test_server.py

"""
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())