diff --git a/requirements-web.txt b/requirements-web.txt new file mode 100644 index 0000000..b49d18f --- /dev/null +++ b/requirements-web.txt @@ -0,0 +1,11 @@ +# Commander AI Lab — Web server dependencies (Phase 5) +fastapi>=0.110.0 +uvicorn[standard]>=0.29.0 +httpx>=0.27.0 # async Scryfall art proxy +websockets>=12.0 # used by uvicorn WebSocket transport +pydantic>=2.6.0 + +# Dev / test +pytest>=8.0 +pytest-asyncio>=0.23 +httpx>=0.27.0 # TestClient uses httpx transport diff --git a/src/commander_ai_lab/web/__init__.py b/src/commander_ai_lab/web/__init__.py new file mode 100644 index 0000000..ba556ba --- /dev/null +++ b/src/commander_ai_lab/web/__init__.py @@ -0,0 +1,10 @@ +""" +Commander AI Lab — Web Server Package (Phase 5) + +Exposes: + create_app() — FastAPI application factory + run() — convenience entry point (uvicorn) +""" +from commander_ai_lab.web.app import create_app, run + +__all__ = ["create_app", "run"] diff --git a/src/commander_ai_lab/web/app.py b/src/commander_ai_lab/web/app.py new file mode 100644 index 0000000..0805e75 --- /dev/null +++ b/src/commander_ai_lab/web/app.py @@ -0,0 +1,102 @@ +""" +Commander AI Lab — FastAPI Application Factory (Phase 5) +========================================================= +Creates and wires the FastAPI application: + - REST API router (/api/v1/...) + - WebSocket router (/ws/game/{game_id}) + - Scryfall art proxy (/api/v1/art/{card_name}) + - Static file mount (/static -> web/static/) + - CORS middleware for browser dev server + +Usage:: + + uvicorn commander_ai_lab.web.app:app --reload --port 8000 + # or programmatically: + from commander_ai_lab.web.app import create_app + app = create_app() +""" + +from __future__ import annotations + +import os +from pathlib import Path + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from fastapi.staticfiles import StaticFiles + +from commander_ai_lab.web.routers.api import router as api_router +from commander_ai_lab.web.routers.ws import router as ws_router +from commander_ai_lab.web.routers.art import router as art_router + +_STATIC_DIR = Path(__file__).parent / "static" + + +def create_app( + allow_origins: list[str] | None = None, + static_dir: Path | None = None, +) -> FastAPI: + """ + FastAPI application factory. + + Args: + allow_origins: CORS origins. Defaults to ["*"] (dev mode). + static_dir: Path to static files. Defaults to web/static/. + """ + app = FastAPI( + title="Commander AI Lab", + version="0.5.0", + description="4-player Commander AI game server with WebSocket game channel.", + ) + + # CORS + origins = allow_origins or ["*"] + app.add_middleware( + CORSMiddleware, + allow_origins=origins, + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], + ) + + # Routers + app.include_router(api_router, prefix="/api/v1") + app.include_router(ws_router) # /ws/game/{game_id} + app.include_router(art_router, prefix="/api/v1") + + # Static files (web client) + sd = static_dir or _STATIC_DIR + if sd.exists(): + app.mount("/static", StaticFiles(directory=str(sd), html=True), name="static") + + @app.get("/health", tags=["meta"]) + async def health() -> dict: + return {"status": "ok", "version": "0.5.0"} + + return app + + +# Module-level app instance (for uvicorn) +app = create_app() + + +def run( + host: str = "0.0.0.0", + port: int = 8000, + reload: bool = False, +) -> None: + """Convenience entry point.""" + try: + import uvicorn + except ImportError: + raise RuntimeError("uvicorn is required: pip install uvicorn[standard]") + uvicorn.run( + "commander_ai_lab.web.app:app", + host=host, + port=port, + reload=reload, + ) + + +if __name__ == "__main__": + run(reload=True) diff --git a/src/commander_ai_lab/web/connection_manager.py b/src/commander_ai_lab/web/connection_manager.py new file mode 100644 index 0000000..96ab915 --- /dev/null +++ b/src/commander_ai_lab/web/connection_manager.py @@ -0,0 +1,98 @@ +""" +Commander AI Lab — WebSocket Connection Manager (Phase 5) +========================================================== +Manages active WebSocket connections grouped by game_id. +Handles: + - connect / disconnect lifecycle + - broadcast to all connections in a game room + - per-seat send (used for private hand data) +""" + +from __future__ import annotations + +import asyncio +import json +from collections import defaultdict +from typing import Optional + +from fastapi import WebSocket + + +class ConnectionManager: + """ + Thread-safe (asyncio) WebSocket connection registry. + + Connections are keyed by game_id. Each connection also records its seat + so the server can send private data (e.g. hand cards) to a specific player. + """ + + def __init__(self) -> None: + # game_id -> list of (WebSocket, seat) + self._rooms: dict[str, list[tuple[WebSocket, int]]] = defaultdict(list) + + async def connect( + self, + ws: WebSocket, + game_id: str, + seat: int, + ) -> None: + """Accept and register a new WebSocket connection.""" + await ws.accept() + self._rooms[game_id].append((ws, seat)) + + def disconnect( + self, + ws: WebSocket, + game_id: str, + ) -> None: + """Remove a disconnected WebSocket from its room.""" + self._rooms[game_id] = [ + (w, s) for w, s in self._rooms[game_id] if w is not ws + ] + if not self._rooms[game_id]: + del self._rooms[game_id] + + async def broadcast( + self, + game_id: str, + message: dict, + ) -> None: + """ + Send a JSON message to all connections in a game room. + Dead connections are silently removed. + """ + payload = json.dumps(message) + dead: list[WebSocket] = [] + + for ws, seat in list(self._rooms.get(game_id, [])): + try: + await ws.send_text(payload) + except Exception: + dead.append(ws) + + for ws in dead: + self.disconnect(ws, game_id) + + async def send_to_seat( + self, + game_id: str, + seat: int, + message: dict, + ) -> None: + """ + Send a JSON message only to connections with a matching seat. + Used for private hand updates. + """ + payload = json.dumps(message) + for ws, s in list(self._rooms.get(game_id, [])): + if s == seat: + try: + await ws.send_text(payload) + except Exception: + self.disconnect(ws, game_id) + + def connection_count(self, game_id: str) -> int: + return len(self._rooms.get(game_id, [])) + + def seats_connected(self, game_id: str) -> list[int]: + return [s for _, s in self._rooms.get(game_id, [])] diff --git a/src/commander_ai_lab/web/routers/__init__.py b/src/commander_ai_lab/web/routers/__init__.py new file mode 100644 index 0000000..2dc14e1 --- /dev/null +++ b/src/commander_ai_lab/web/routers/__init__.py @@ -0,0 +1 @@ +# Router sub-package diff --git a/src/commander_ai_lab/web/routers/api.py b/src/commander_ai_lab/web/routers/api.py new file mode 100644 index 0000000..9a6acc6 --- /dev/null +++ b/src/commander_ai_lab/web/routers/api.py @@ -0,0 +1,121 @@ +""" +Commander AI Lab — REST API Router (Phase 5) +============================================= +Endpoints: + POST /api/v1/games Create a new game session + GET /api/v1/games/{game_id} Get game state snapshot + POST /api/v1/games/{game_id}/move Submit a human move + GET /api/v1/games/{game_id}/moves List legal moves for human seat + POST /api/v1/games/{game_id}/concede Concede for a seat + DELETE /api/v1/games/{game_id} Tear down a game session + GET /api/v1/games List active sessions +""" + +from __future__ import annotations + +from fastapi import APIRouter, HTTPException, Body +from pydantic import BaseModel, Field + +from commander_ai_lab.web.session_store import SessionStore, GameSession + +router = APIRouter(tags=["game"]) +_store = SessionStore() + + +# --------------------------------------------------------------------------- +# Request / Response models +# --------------------------------------------------------------------------- + +class CreateGameRequest(BaseModel): + player_names: list[str] = Field( + default=["Human", "AI-Timmy", "AI-Spike", "AI-Johnny"], + description="Exactly 4 player names. Index 0 is always the human seat.", + ) + human_seat: int = Field(default=0, ge=0, le=3) + ai_personality: list[str] = Field( + default=["aggressive", "control", "combo"], + description="Personality for seats 1, 2, 3 respectively.", + ) + + +class MoveRequest(BaseModel): + seat: int = Field(..., ge=0, le=3) + move_id: int = Field(..., description="Move ID from /moves endpoint.") + + +class ConcedeRequest(BaseModel): + seat: int = Field(..., ge=0, le=3) + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + +@router.post("/games", status_code=201) +async def create_game(req: CreateGameRequest) -> dict: + """Create and start a new 4-player Commander game session.""" + if len(req.player_names) != 4: + raise HTTPException(400, "Exactly 4 player names are required.") + session = await _store.create_session( + player_names=req.player_names, + human_seat=req.human_seat, + ai_personalities=req.ai_personality, + ) + return {"game_id": session.game_id, "status": "started"} + + +@router.get("/games") +async def list_games() -> dict: + """List all active game session IDs.""" + return {"games": _store.list_session_ids()} + + +@router.get("/games/{game_id}") +async def get_game(game_id: str) -> dict: + """Return a JSON snapshot of the current game state.""" + session = _get_session(game_id) + return session.state_snapshot() + + +@router.get("/games/{game_id}/moves") +async def get_legal_moves(game_id: str, seat: int = 0) -> dict: + """Return legal moves for the given seat.""" + session = _get_session(game_id) + moves = session.get_legal_moves(seat) + return {"seat": seat, "moves": moves} + + +@router.post("/games/{game_id}/move") +async def submit_move(game_id: str, req: MoveRequest) -> dict: + """Submit a human move. Returns updated game state snapshot.""" + session = _get_session(game_id) + ok = await session.apply_human_move(seat=req.seat, move_id=req.move_id) + if not ok: + raise HTTPException(400, "Move is not legal or it is not your turn.") + return {"accepted": True, "state": session.state_snapshot()} + + +@router.post("/games/{game_id}/concede") +async def concede(game_id: str, req: ConcedeRequest) -> dict: + """Concede the game for the given seat.""" + session = _get_session(game_id) + await session.concede(req.seat) + return {"conceded": True, "seat": req.seat} + + +@router.delete("/games/{game_id}", status_code=204) +async def delete_game(game_id: str) -> None: + """Tear down a game session.""" + if not _store.remove_session(game_id): + raise HTTPException(404, f"Game {game_id!r} not found.") + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + +def _get_session(game_id: str) -> GameSession: + session = _store.get_session(game_id) + if session is None: + raise HTTPException(404, f"Game {game_id!r} not found.") + return session diff --git a/src/commander_ai_lab/web/routers/art.py b/src/commander_ai_lab/web/routers/art.py new file mode 100644 index 0000000..ec7f885 --- /dev/null +++ b/src/commander_ai_lab/web/routers/art.py @@ -0,0 +1,130 @@ +""" +Commander AI Lab — Scryfall Art Proxy (Phase 5) +================================================ +Endpoint: GET /api/v1/art/{card_name} + +Proxy for Scryfall card art images. + +Behaviour: + 1. Look up card in local LRU disk cache (~/.cache/commander_ai_lab/art/). + 2. If cached, stream the JPEG directly (cache-control: 7 days). + 3. If not cached, fetch from Scryfall API, store to disk, then stream. + 4. On Scryfall failure, return 404 with JSON detail. + +Rate limiting: Scryfall asks for ≤10 req/s. The proxy uses an asyncio +semaphore (max 8 concurrent outbound fetches) and a 50 ms delay between +requests to stay comfortably under the limit. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +from pathlib import Path +from urllib.parse import quote + +from fastapi import APIRouter, HTTPException +from fastapi.responses import FileResponse, JSONResponse + +router = APIRouter(tags=["art"]) + +_CACHE_DIR = Path(os.path.expanduser("~/.cache/commander_ai_lab/art")) +_CACHE_DIR.mkdir(parents=True, exist_ok=True) +_SCRYFALL_NAMED = "https://api.scryfall.com/cards/named?fuzzy={name}" +_FETCH_SEM = asyncio.Semaphore(8) # max concurrent Scryfall fetches +_FETCH_DELAY = 0.05 # seconds between fetches + + +def _cache_path(card_name: str) -> Path: + key = hashlib.sha1(card_name.lower().encode()).hexdigest() + return _CACHE_DIR / f"{key}.jpg" + + +@router.get("/art/{card_name}", response_class=FileResponse) +async def get_card_art(card_name: str) -> FileResponse: + """ + Return card art JPEG for the given card name. + Resolves fuzzy name via Scryfall, caches locally. + """ + cached = _cache_path(card_name) + + if cached.exists(): + return FileResponse( + path=str(cached), + media_type="image/jpeg", + headers={"Cache-Control": "max-age=604800"}, + ) + + # Fetch from Scryfall + img_bytes = await _fetch_scryfall_art(card_name) + if img_bytes is None: + raise HTTPException(404, f"Art not found for card: {card_name!r}") + + cached.write_bytes(img_bytes) + return FileResponse( + path=str(cached), + media_type="image/jpeg", + headers={"Cache-Control": "max-age=604800"}, + ) + + +async def _fetch_scryfall_art(card_name: str) -> bytes | None: + """Fetch card art bytes from Scryfall (rate-limited).""" + async with _FETCH_SEM: + await asyncio.sleep(_FETCH_DELAY) + try: + import httpx + except ImportError: + # Fall back to stdlib if httpx not installed + return await _fetch_stdlib(card_name) + + async with httpx.AsyncClient(timeout=10.0) as client: + # Step 1: resolve name + url = _SCRYFALL_NAMED.format(name=quote(card_name)) + r = await client.get(url) + if r.status_code != 200: + return None + data = r.json() + + # Step 2: pick image URI + image_uris = data.get("image_uris") or ( + data.get("card_faces", [{}])[0].get("image_uris", {}) + ) + img_url = image_uris.get("small") or image_uris.get("normal") + if not img_url: + return None + + # Step 3: download image + r2 = await client.get(img_url) + if r2.status_code != 200: + return None + return r2.content + + +async def _fetch_stdlib(card_name: str) -> bytes | None: + """Stdlib fallback for environments without httpx.""" + import asyncio + from urllib.request import urlopen + from urllib.parse import quote + import json as _json + + def _sync_fetch() -> bytes | None: + try: + url = _SCRYFALL_NAMED.format(name=quote(card_name)) + with urlopen(url, timeout=8) as r: + data = _json.loads(r.read()) + image_uris = data.get("image_uris") or ( + data.get("card_faces", [{}])[0].get("image_uris", {}) + ) + img_url = image_uris.get("small") or image_uris.get("normal") + if not img_url: + return None + with urlopen(img_url, timeout=10) as r: + return r.read() + except Exception: + return None + + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, _sync_fetch) diff --git a/src/commander_ai_lab/web/routers/ws.py b/src/commander_ai_lab/web/routers/ws.py new file mode 100644 index 0000000..55835ec --- /dev/null +++ b/src/commander_ai_lab/web/routers/ws.py @@ -0,0 +1,115 @@ +""" +Commander AI Lab — WebSocket Game Channel (Phase 5) +==================================================== +Endpoint: ws://host/ws/game/{game_id}?seat={0-3} + +Message protocol (JSON): + + Client → Server: + { "type": "move", "seat": 0, "move_id": 7 } + { "type": "concede", "seat": 0 } + { "type": "ping" } + + Server → Client (broadcast to all connections on the same game): + { "type": "state", "data": } + { "type": "event", "seat": N, "event_type": "...", "narration": "...", "turn": N } + { "type": "phase", "phase": "main1", "active_seat": N, "turn": N } + { "type": "thinking", "seat": N, "is_thinking": bool } + { "type": "elimination", "seat": N, "player_name": "..." } + { "type": "game_over", "winner": N, "reason": "..." } + { "type": "pong" } + { "type": "error", "detail": "..." } +""" + +from __future__ import annotations + +import json +from typing import Optional + +from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Query + +from commander_ai_lab.web.session_store import SessionStore +from commander_ai_lab.web.connection_manager import ConnectionManager + +router = APIRouter(tags=["websocket"]) +_store = SessionStore() # shares singleton with api.py via module-level instance +_manager = ConnectionManager() + + +@router.websocket("/ws/game/{game_id}") +async def game_websocket( + websocket: WebSocket, + game_id: str, + seat: int = Query(default=0, ge=0, le=3), +) -> None: + """ + WebSocket endpoint for a live game channel. + + Each connecting client identifies its seat via the `seat` query param. + Multiple clients may observe the same game (spectators pass seat=-1 in practice + but the validator clamps to 0-3; full spectator support is Phase 8). + """ + session = _store.get_session(game_id) + if session is None: + await websocket.close(code=4004, reason=f"Game {game_id!r} not found") + return + + await _manager.connect(websocket, game_id, seat) + + # Register broadcast callback on session so game events reach all clients + session.set_broadcast_callback( + lambda msg: _manager.broadcast(game_id, msg) + ) + + # Send initial full state snapshot + await websocket.send_text(json.dumps({ + "type": "state", + "data": session.state_snapshot(), + })) + + try: + while True: + raw = await websocket.receive_text() + try: + msg = json.loads(raw) + except json.JSONDecodeError: + await websocket.send_text(json.dumps( + {"type": "error", "detail": "Invalid JSON"} + )) + continue + + msg_type = msg.get("type", "") + + if msg_type == "ping": + await websocket.send_text(json.dumps({"type": "pong"})) + + elif msg_type == "move": + move_seat = msg.get("seat", seat) + move_id = msg.get("move_id") + if move_id is None: + await websocket.send_text(json.dumps( + {"type": "error", "detail": "move_id is required"} + )) + continue + ok = await session.apply_human_move(seat=move_seat, move_id=move_id) + if not ok: + await websocket.send_text(json.dumps( + {"type": "error", "detail": "Illegal move or not your turn"} + )) + else: + # Broadcast updated state to all watchers + await _manager.broadcast(game_id, { + "type": "state", + "data": session.state_snapshot(), + }) + + elif msg_type == "concede": + await session.concede(seat) + + else: + await websocket.send_text(json.dumps( + {"type": "error", "detail": f"Unknown message type: {msg_type!r}"} + )) + + except WebSocketDisconnect: + _manager.disconnect(websocket, game_id) diff --git a/src/commander_ai_lab/web/session_store.py b/src/commander_ai_lab/web/session_store.py new file mode 100644 index 0000000..31c2819 --- /dev/null +++ b/src/commander_ai_lab/web/session_store.py @@ -0,0 +1,214 @@ +""" +Commander AI Lab — Session Store (Phase 5) +========================================== +In-memory registry of live GameSession objects. + +Each session wraps: + - A CommanderGameState + - A CommanderTurnManager (running in a background asyncio task) + - A broadcast callback (set by the WebSocket router) + +SessionStore is intentionally not thread-safe beyond asyncio coroutines — +all mutations happen in the async event loop. +""" + +from __future__ import annotations + +import asyncio +import uuid +from dataclasses import dataclass, field +from typing import Callable, Optional, Awaitable + + +# --------------------------------------------------------------------------- +# Stub game state / turn manager (replaced by real imports once available) +# --------------------------------------------------------------------------- + +class _StubPlayer: + def __init__(self, name: str, seat: int) -> None: + self.name = name + self.seat = seat + self.life = 40 + self.eliminated = False + self.hand = [] + self.battlefield = [] + self.graveyard = [] + self.exile = [] + self.command_zone = [] + + def to_dict(self, private: bool = False) -> dict: + return { + "name": self.name, + "seat": self.seat, + "life": self.life, + "eliminated": self.eliminated, + "hand_count": len(self.hand), + "hand": self.hand if private else [], + "battlefield": self.battlefield, + "graveyard": self.graveyard, + "exile": self.exile, + "command_zone": self.command_zone, + } + + +class _StubGameState: + def __init__(self, player_names: list[str]) -> None: + self.players = [_StubPlayer(n, i) for i, n in enumerate(player_names)] + self.turn = 1 + self.current_phase = "main1" + self.active_player_seat = 0 + self.stack = [] + self.winner = None + self.game_over = False + + def get_legal_moves(self, seat: int) -> list[dict]: + return [ + {"id": 1, "category": "pass_priority", "description": "Pass priority"}, + {"id": 2, "category": "play_land", "description": "Play a land"}, + ] + + def apply_move(self, seat: int, move_id: int) -> None: + pass # real logic wired in by CommanderTurnManager + + +# --------------------------------------------------------------------------- +# GameSession +# --------------------------------------------------------------------------- + +class GameSession: + """ + Wraps a single live Commander game. + + Lifecycle: + 1. Created by SessionStore.create_session() + 2. Turn manager runs as an asyncio background task + 3. WebSocket router attaches a broadcast callback + 4. Destroyed by SessionStore.remove_session() + """ + + def __init__( + self, + game_id: str, + player_names: list[str], + human_seat: int, + ai_personalities: list[str], + ) -> None: + self.game_id = game_id + self.human_seat = human_seat + self._gs = _StubGameState(player_names) + self._pending_moves: dict[int, asyncio.Future] = {} # seat -> Future + self._broadcast: Optional[Callable] = None + self._task: Optional[asyncio.Task] = None + + # ------------------------------------------------------------------ + # Public API used by routers + # ------------------------------------------------------------------ + + def state_snapshot(self, private_seat: Optional[int] = None) -> dict: + """Return serialisable game state dict.""" + return { + "game_id": self.game_id, + "turn": self._gs.turn, + "current_phase": self._gs.current_phase, + "active_seat": self._gs.active_player_seat, + "game_over": self._gs.game_over, + "winner": self._gs.winner, + "stack_size": len(self._gs.stack), + "players": [ + p.to_dict(private=(p.seat == private_seat)) + for p in self._gs.players + ], + } + + def get_legal_moves(self, seat: int) -> list[dict]: + return self._gs.get_legal_moves(seat) + + async def apply_human_move( + self, + seat: int, + move_id: int, + ) -> bool: + """ + Inject a human move. + Resolves the pending Future if the turn manager is waiting. + """ + fut = self._pending_moves.get(seat) + if fut and not fut.done(): + fut.set_result(move_id) + return True + # Fallback: apply directly (e.g. during testing) + self._gs.apply_move(seat, move_id) + return True + + async def concede(self, seat: int) -> None: + """Mark a seat as eliminated.""" + if 0 <= seat < len(self._gs.players): + self._gs.players[seat].eliminated = True + self._gs.players[seat].life = 0 + if self._broadcast: + await self._broadcast({ + "type": "elimination", + "seat": seat, + "player_name": self._gs.players[seat].name, + }) + + def set_broadcast_callback( + self, + callback: Callable[[dict], Awaitable[None]], + ) -> None: + self._broadcast = callback + + def register_human_future( + self, + seat: int, + future: asyncio.Future, + ) -> None: + """Called by turn manager when waiting for human input.""" + self._pending_moves[seat] = future + + +# --------------------------------------------------------------------------- +# SessionStore +# --------------------------------------------------------------------------- + +class SessionStore: + """ + Singleton-ish in-memory store for GameSession objects. + One instance is shared between the API and WS routers via module globals. + """ + _instance: Optional[SessionStore] = None + + def __new__(cls) -> SessionStore: + # Simple module-level singleton + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance._sessions: dict[str, GameSession] = {} + return cls._instance + + async def create_session( + self, + player_names: list[str], + human_seat: int = 0, + ai_personalities: list[str] | None = None, + ) -> GameSession: + game_id = str(uuid.uuid4())[:8] + session = GameSession( + game_id=game_id, + player_names=player_names, + human_seat=human_seat, + ai_personalities=ai_personalities or ["aggressive", "control", "combo"], + ) + self._sessions[game_id] = session + return session + + def get_session(self, game_id: str) -> Optional[GameSession]: + return self._sessions.get(game_id) + + def list_session_ids(self) -> list[str]: + return list(self._sessions.keys()) + + def remove_session(self, game_id: str) -> bool: + if game_id in self._sessions: + del self._sessions[game_id] + return True + return False diff --git a/src/commander_ai_lab/web/static/index.html b/src/commander_ai_lab/web/static/index.html new file mode 100644 index 0000000..76e6cfd --- /dev/null +++ b/src/commander_ai_lab/web/static/index.html @@ -0,0 +1,489 @@ + + + + + + Commander AI Lab + + + + + +
+ — + Untap + Upkeep + Draw + Main 1 + Combat + Attackers + Blockers + Damage + Main 2 + End +
+ + +
+
+ +
+
+
+ + +
+ + + Connecting… +
+ + +
+
+
+
+
+
+
+ + + + diff --git a/tests/test_phase5.py b/tests/test_phase5.py new file mode 100644 index 0000000..472f19f --- /dev/null +++ b/tests/test_phase5.py @@ -0,0 +1,250 @@ +""" +Phase 5 Unit Tests — FastAPI Server & WebSocket Channel +======================================================= +Run with: pytest tests/test_phase5.py -v + +Requires: pip install fastapi httpx pytest-asyncio +""" + +from __future__ import annotations + +import json +import pytest + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def app(): + """Create a fresh FastAPI app for each test.""" + # Reset singleton so tests don't share session state + from commander_ai_lab.web.session_store import SessionStore + SessionStore._instance = None + from commander_ai_lab.web.app import create_app + return create_app() + + +@pytest.fixture +def client(app): + from fastapi.testclient import TestClient + return TestClient(app) + + +# --------------------------------------------------------------------------- +# Health +# --------------------------------------------------------------------------- + +class TestHealth: + def test_health_ok(self, client): + r = client.get("/health") + assert r.status_code == 200 + assert r.json()["status"] == "ok" + + def test_health_has_version(self, client): + r = client.get("/health") + assert "version" in r.json() + + +# --------------------------------------------------------------------------- +# Game lifecycle +# --------------------------------------------------------------------------- + +class TestGameLifecycle: + def _create(self, client, names=None): + names = names or ["A", "B", "C", "D"] + return client.post("/api/v1/games", json={ + "player_names": names, + "human_seat": 0, + "ai_personality": ["aggressive", "control", "combo"], + }) + + def test_create_game_201(self, client): + r = self._create(client) + assert r.status_code == 201 + assert "game_id" in r.json() + + def test_create_game_wrong_player_count(self, client): + r = client.post("/api/v1/games", json={ + "player_names": ["A", "B"], + "human_seat": 0, + "ai_personality": [], + }) + assert r.status_code == 400 + + def test_list_games_empty(self, client): + r = client.get("/api/v1/games") + assert r.status_code == 200 + assert r.json()["games"] == [] + + def test_list_games_after_create(self, client): + self._create(client) + r = client.get("/api/v1/games") + assert len(r.json()["games"]) == 1 + + def test_get_game_state(self, client): + gid = self._create(client).json()["game_id"] + r = client.get(f"/api/v1/games/{gid}") + assert r.status_code == 200 + data = r.json() + assert data["game_id"] == gid + assert len(data["players"]) == 4 + + def test_get_game_404(self, client): + r = client.get("/api/v1/games/doesnotexist") + assert r.status_code == 404 + + def test_get_legal_moves(self, client): + gid = self._create(client).json()["game_id"] + r = client.get(f"/api/v1/games/{gid}/moves?seat=0") + assert r.status_code == 200 + assert isinstance(r.json()["moves"], list) + assert len(r.json()["moves"]) > 0 + + def test_submit_move(self, client): + gid = self._create(client).json()["game_id"] + moves = client.get(f"/api/v1/games/{gid}/moves?seat=0").json()["moves"] + move_id = moves[0]["id"] + r = client.post(f"/api/v1/games/{gid}/move", json={"seat": 0, "move_id": move_id}) + assert r.status_code == 200 + assert r.json()["accepted"] is True + + def test_submit_move_404(self, client): + r = client.post("/api/v1/games/bad/move", json={"seat": 0, "move_id": 1}) + assert r.status_code == 404 + + def test_concede(self, client): + gid = self._create(client).json()["game_id"] + r = client.post(f"/api/v1/games/{gid}/concede", json={"seat": 1}) + assert r.status_code == 200 + assert r.json()["conceded"] is True + + def test_delete_game(self, client): + gid = self._create(client).json()["game_id"] + r = client.delete(f"/api/v1/games/{gid}") + assert r.status_code == 204 + r2 = client.get(f"/api/v1/games/{gid}") + assert r2.status_code == 404 + + def test_delete_game_404(self, client): + r = client.delete("/api/v1/games/ghost") + assert r.status_code == 404 + + +# --------------------------------------------------------------------------- +# State snapshot shape +# --------------------------------------------------------------------------- + +class TestStateSnapshot: + def test_snapshot_fields(self, client): + gid = client.post("/api/v1/games", json={ + "player_names": ["P0","P1","P2","P3"], + "human_seat": 0, + "ai_personality": ["aggressive","control","combo"], + }).json()["game_id"] + snap = client.get(f"/api/v1/games/{gid}").json() + for field in ["game_id","turn","current_phase","active_seat","game_over","winner","players"]: + assert field in snap, f"Missing field: {field}" + + def test_snapshot_player_fields(self, client): + gid = client.post("/api/v1/games", json={ + "player_names": ["P0","P1","P2","P3"], + "human_seat": 0, + "ai_personality": ["aggressive","control","combo"], + }).json()["game_id"] + snap = client.get(f"/api/v1/games/{gid}").json() + p = snap["players"][0] + for f in ["name","seat","life","eliminated","hand_count","battlefield","graveyard"]: + assert f in p, f"Missing player field: {f}" + + def test_starting_life_total(self, client): + gid = client.post("/api/v1/games", json={ + "player_names": ["P0","P1","P2","P3"], + "human_seat": 0, + "ai_personality": ["aggressive","control","combo"], + }).json()["game_id"] + snap = client.get(f"/api/v1/games/{gid}").json() + for p in snap["players"]: + assert p["life"] == 40 + + +# --------------------------------------------------------------------------- +# ConnectionManager +# --------------------------------------------------------------------------- + +class TestConnectionManager: + def test_connection_count_zero(self): + from commander_ai_lab.web.connection_manager import ConnectionManager + cm = ConnectionManager() + assert cm.connection_count("abc") == 0 + + def test_seats_connected_empty(self): + from commander_ai_lab.web.connection_manager import ConnectionManager + cm = ConnectionManager() + assert cm.seats_connected("abc") == [] + + +# --------------------------------------------------------------------------- +# SessionStore +# --------------------------------------------------------------------------- + +class TestSessionStore: + @pytest.fixture(autouse=True) + def reset_store(self): + from commander_ai_lab.web.session_store import SessionStore + SessionStore._instance = None + yield + SessionStore._instance = None + + @pytest.mark.asyncio + async def test_create_and_get(self): + from commander_ai_lab.web.session_store import SessionStore + store = SessionStore() + session = await store.create_session(["A","B","C","D"]) + assert store.get_session(session.game_id) is session + + @pytest.mark.asyncio + async def test_list_ids(self): + from commander_ai_lab.web.session_store import SessionStore + store = SessionStore() + s1 = await store.create_session(["A","B","C","D"]) + s2 = await store.create_session(["W","X","Y","Z"]) + ids = store.list_session_ids() + assert s1.game_id in ids + assert s2.game_id in ids + + @pytest.mark.asyncio + async def test_remove(self): + from commander_ai_lab.web.session_store import SessionStore + store = SessionStore() + session = await store.create_session(["A","B","C","D"]) + ok = store.remove_session(session.game_id) + assert ok is True + assert store.get_session(session.game_id) is None + + @pytest.mark.asyncio + async def test_state_snapshot_shape(self): + from commander_ai_lab.web.session_store import SessionStore + store = SessionStore() + session = await store.create_session(["A","B","C","D"]) + snap = session.state_snapshot() + assert snap["game_id"] == session.game_id + assert len(snap["players"]) == 4 + + @pytest.mark.asyncio + async def test_concede_marks_eliminated(self): + from commander_ai_lab.web.session_store import SessionStore + store = SessionStore() + session = await store.create_session(["A","B","C","D"]) + await session.concede(seat=2) + assert session._gs.players[2].eliminated is True + assert session._gs.players[2].life == 0 + + @pytest.mark.asyncio + async def test_legal_moves_returned(self): + from commander_ai_lab.web.session_store import SessionStore + store = SessionStore() + session = await store.create_session(["A","B","C","D"]) + moves = session.get_legal_moves(0) + assert isinstance(moves, list) + assert len(moves) > 0