commit c52f30dc59691cdc0d54d146dd0c72199c4eeded Author: tukaunu Date: Mon Sep 28 03:36:46 2026 +0200 Moonchess: vanishing tic-tac-toe, online and practice modes Online play over WebSockets with any-string game codes, seats and spectators, and a 1-second heartbeat. Practice mode against Columbina, backed by a full solver, with an accuracy slider, spoilers, undo and piece freshness. Co-Authored-By: Claude Opus 5.5 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0adab77 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.venv/ +__pycache__/ +.DS_Store +.claude/settings.local.json diff --git a/README.md b/README.md new file mode 100644 index 0000000..c809474 --- /dev/null +++ b/README.md @@ -0,0 +1,109 @@ +# Moonchess + +Tic-tac-toe where each player can have at most three pieces on the board. Placing a +fourth isn't possible: once you have three down, your oldest vanishes as your turn +starts (it's shown faded during your opponent's turn), so you can place right back on +its square. This follows Genshin Impact's Moon Chess. + +## Running locally + +```sh +python3 -m venv .venv +.venv/bin/pip install -r requirements.txt +.venv/bin/python server.py # http://127.0.0.1:8080 +.venv/bin/python server.py --host 0.0.0.0 --port 8080 # reachable on your LAN +.venv/bin/python practice.py # vs the computer: http://127.0.0.1:8081 +``` + +Tests: `.venv/bin/python -m unittest discover -s tests -t .` + +## How it works + +- `game.py`: the rules, no networking. +- `server.py`: one aiohttp process. `/ws?game=CODE` is the live connection, and + everything else is static files. +- `practice.py`: a separate server for playing against the computer. It holds no state: + the page sends the position to `POST /api/play` and gets the next one back. +- `solver.py`: solves every reachable position (~67k, about a second at startup) so the + computer knows the perfect move everywhere. +- `static/`: the pages. Plain HTML/CSS/JS, no build step. `board.js` is the board + drawing shared by `app.js` (online) and `practice.js`. + +Practice mode: you start as X, which moves first. Click either side at any time to swap, +and the computer (Columbina, after the Genshin Impact minigame this game comes from) +plays whichever side you aren't. The accuracy slider is the chance each +computer move is perfect; otherwise it plays a random empty square. It can be changed +mid-game too. Undo (or Ctrl/Cmd+Z) goes back to your previous turn, taking back +Columbina's reply as well. Perfect play is a draw: you can only threaten one line +at a time, so you only lose by missing a block. + +Seating: each game has a list of slots. Slots 0 and 1 are X and O, and the rest are +spectators. A new connection takes the earliest empty slot, so if a player drops, +whoever connects next gets their seat. Refreshing fast enough to grab it is allowed. + +Heartbeat: the client pings every second. A seat whose player has been silent for 2 +seconds shows as "Disconnected", and after 4 seconds they're dropped and the seat +shows as "Empty seat". The client dims the board and reconnects if it hasn't heard +from the server in 4 seconds. That automatic reconnect (unlike a refresh) comes from +the same page load, so it reclaims its own slot even if the server hasn't dropped the +old connection yet. A player who refreshes during the "Disconnected" window lands in +a spectator slot and has to refresh again once their seat shows "Empty seat". + +Codes: any string of up to 32 characters works as a code, ignoring case and extra +whitespace. Joining a code that has no game creates one. "Generate code" just picks a +random 5-character code in the browser. Games live in memory: a game is deleted as soon as +nobody is connected to it (so if the last person in a game refreshes, it starts over), +and restarting the server ends all games. + +## Deploying behind Cloudflare (Cloudflare Tunnel) + +A tunnel lets Cloudflare reach the server without opening router ports or exposing +your home IP. WebSockets work through it without any extra configuration. + +1. Run the server on the box, bound to localhost: `.venv/bin/python server.py --port 8080` +2. Install `cloudflared`, then: + ```sh + cloudflared tunnel login + cloudflared tunnel create moonchess + cloudflared tunnel route dns moonchess moonchess.example.com + cloudflared tunnel route dns moonchess practice.moonchess.example.com # if running practice.py + ``` +3. `~/.cloudflared/config.yml`: + ```yaml + tunnel: moonchess + credentials-file: /home/YOU/.cloudflared/.json + ingress: + - hostname: moonchess.example.com + service: http://localhost:8080 + - hostname: practice.moonchess.example.com # optional, if you run practice.py + service: http://localhost:8081 + - service: http_status:404 + ``` +4. `cloudflared tunnel run moonchess` (or `sudo cloudflared service install` to run it at boot). + +To keep the game server running, an example systemd unit (`/etc/systemd/system/moonchess.service`): + +```ini +[Unit] +Description=Moonchess +After=network.target + +[Service] +WorkingDirectory=/opt/moonchess +ExecStart=/opt/moonchess/.venv/bin/python server.py --port 8080 +Restart=always +User=moonchess + +[Install] +WantedBy=multi-user.target +``` + +If you'd rather use the regular orange-cloud proxy with port forwarding, point the +DNS record at your IP and forward a port to the server. That also works, because +Cloudflare proxies WebSockets on all plans. Cloudflare only proxies certain ports +(e.g. 80, 443, 8080, 8443), so bind to one of those or put a reverse proxy in front. + +## Future work + +- Seat-open alert: give everyone (especially spectators) an obvious signal when a + player seat becomes empty, so they can race to refresh and grab it. diff --git a/game.py b/game.py new file mode 100644 index 0000000..5dd751b --- /dev/null +++ b/game.py @@ -0,0 +1,111 @@ +"""Rules for vanishing tic-tac-toe. + +Each seat may have at most MAX_PIECES on the board. With a full set, your oldest +piece is removed as your turn starts, so you have at most MAX_PIECES - 1 pieces down +while placing and may place on the square it left. +""" + +LINES = [ + (0, 1, 2), (3, 4, 5), (6, 7, 8), # rows + (0, 3, 6), (1, 4, 7), (2, 5, 8), # columns + (0, 4, 8), (2, 4, 6), # diagonals +] +MAX_PIECES = 3 + + +class Game: + def __init__(self): + self.first = 0 # seat that moves first this round; alternates on rematch + self.new_round() + + @classmethod + def from_history(cls, history, turn): + """Rebuild a round from each seat's cells (oldest first). Raises ValueError if invalid.""" + try: + hx, ho = (list(h) for h in history) + except (TypeError, ValueError): + raise ValueError("Invalid position") + cells = hx + ho + if (turn not in (0, 1) + or any(type(c) is not int or not 0 <= c < 9 for c in cells) + or len(set(cells)) != len(cells) + or len(hx) > MAX_PIECES or len(ho) > MAX_PIECES): + raise ValueError("Invalid position") + game = cls() + game.history = (hx, ho) + game.turn = turn + for seat, hist in enumerate(game.history): + for c in hist: + game.board[c] = seat + for seat in (0, 1): + line = game._winning_line(seat) + if line: + game.winner, game.win_line = seat, line + if game.winner is None and len(game.history[turn]) == MAX_PIECES: + raise ValueError("Invalid position") # the mover's oldest should already be gone + return game + + def new_round(self): + self.board = [None] * 9 # None, 0 (X) or 1 (O) + self.history = ([], []) # per seat, oldest cell first + self.turn = self.first + self.winner = None + self.win_line = None + self.rematch_votes = set() + + def doomed(self): + """The waiting player's oldest piece, which vanishes when their turn starts.""" + if self.winner is not None: + return None + hist = self.history[1 - self.turn] + return hist[0] if len(hist) == MAX_PIECES else None + + def play(self, seat, cell): + """Place a piece. Returns an error string, or None on success.""" + if self.winner is not None: + return "The round is over" + if seat != self.turn: + return "Not your turn" + if not isinstance(cell, int) or not 0 <= cell < 9: + return "Invalid square" + if self.board[cell] is not None: + return "That square is taken" + + self.board[cell] = seat + self.history[seat].append(cell) + + line = self._winning_line(seat) + if line: + self.winner = seat + self.win_line = line + else: + self.turn = 1 - seat + hist = self.history[self.turn] + if len(hist) == MAX_PIECES: + self.board[hist.pop(0)] = None + return None + + def _winning_line(self, seat): + for line in LINES: + if all(self.board[i] == seat for i in line): + return list(line) + return None + + def vote_rematch(self, seat): + if self.winner is None: + return + self.rematch_votes.add(seat) + if self.rematch_votes >= {0, 1}: + self.first = 1 - self.first + self.new_round() + + def to_dict(self): + return { + "board": self.board, + "history": self.history, + "turn": self.turn, + "winner": self.winner, + "winLine": self.win_line, + "doomed": self.doomed(), + "rematch": [0 in self.rematch_votes, 1 in self.rematch_votes], + } diff --git a/practice.py b/practice.py new file mode 100644 index 0000000..acfbf7d --- /dev/null +++ b/practice.py @@ -0,0 +1,60 @@ +"""Moonchess practice server: play against the computer. + +Stateless: the page sends the current position with every request and gets the new +position back, so there's no session to lose. The page decides whose turn belongs to +the computer; this server only applies moves and picks them. +""" + +import argparse +import logging + +from aiohttp import web + +from game import Game +from server import STATIC_DIR, no_cache +from solver import Solver + +solver = Solver() # solves every position up front; takes about a second + + +async def index(request): + return web.FileResponse(STATIC_DIR / "practice.html") + + +async def play(request): + """Body: {history, turn, cell} to play `cell`, or {history, turn, accuracy} for a + computer move, where accuracy (0-1) is the chance it plays a perfect move. A computer + move's reply also says how it was chosen, as computerMove: {cell, how}.""" + try: + body = await request.json() + game = Game.from_history(body["history"], body["turn"]) + cell = body.get("cell") + accuracy = float(body.get("accuracy", 1)) + except (ValueError, KeyError, TypeError, AttributeError): + return web.json_response({"error": "Invalid request"}, status=400) + + computer_move = None + if cell is None and game.winner is None: + cell, how = solver.choose(game, accuracy) + computer_move = {"cell": cell, "how": how} + error = game.play(game.turn, cell) + if error: + return web.json_response({"error": error}, status=400) + return web.json_response({**game.to_dict(), "computerMove": computer_move}) + + +def make_app(): + app = web.Application(middlewares=[no_cache]) + app.router.add_get("/", index) + app.router.add_post("/api/play", play) + app.router.add_static("/static", STATIC_DIR) + return app + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Moonchess practice server") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8081) + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") + web.run_app(make_app(), host=args.host, port=args.port) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6a37f7b --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +aiohttp>=3.9 diff --git a/server.py b/server.py new file mode 100644 index 0000000..25c84e7 --- /dev/null +++ b/server.py @@ -0,0 +1,243 @@ +"""Moonchess server: serves the page and runs live games over WebSockets. + +Any string works as a game code. Joining a code with no game creates one, and a game +is deleted as soon as nobody is connected to it. + +Seating: each room keeps a list of slots. Slots 0 and 1 are the players (X and O), +everything after is spectators. A new connection takes the earliest empty slot, +so when a player drops, whoever connects next -- including a spectator who +refreshes fast enough -- inherits their seat. That is intentional. +""" + +import argparse +import asyncio +import logging +import time +from pathlib import Path + +from aiohttp import WSMsgType, web + +from game import Game + +STATIC_DIR = Path(__file__).parent / "static" +STALE_AFTER = 2.0 # seconds without a message before a seat shows as disconnected +HEARTBEAT_TIMEOUT = 4.0 # seconds without a message before a client is dropped +MAX_ROOMS = 10_000 +MAX_CODE_LENGTH = 32 +CLOSE_INVALID_CODE = 4400 +CLOSE_SERVER_FULL = 4503 +MAX_MESSAGE_SIZE = 4096 # bytes; real messages are tiny, larger ones close the connection + +log = logging.getLogger("moonchess") + + +class Client: + def __init__(self, ws, client_id): + self.ws = ws + self.client_id = client_id + self.last_seen = time.monotonic() + + +class Room: + def __init__(self, token): + self.token = token + self.game = Game() + self.slots = [] # Client or None + self.last_seats = None # seat_status() as of the last broadcast + + def join(self, client): + # A reconnect from the same page load reclaims its own not-yet-reaped slot. + for i, c in enumerate(self.slots): + if c is not None and client.client_id and c.client_id == client.client_id: + self.slots[i] = client + asyncio.ensure_future(c.ws.close()) + return + for i, c in enumerate(self.slots): + if c is None: + self.slots[i] = client + return + self.slots.append(client) + + def leave(self, client): + if client not in self.slots: + return False + i = self.slots.index(client) + self.slots[i] = None + if i < 2: + self.game.rematch_votes.discard(i) + while len(self.slots) > 2 and self.slots[-1] is None: + self.slots.pop() + return True + + def connected(self): + return [c for c in self.slots if c is not None] + + def seat_of(self, client): + i = self.slots.index(client) + return i if i < 2 else None + + def players_present(self): + return [len(self.slots) > i and self.slots[i] is not None for i in (0, 1)] + + def seat_status(self): + """'connected', 'disconnected' (silent but not yet dropped) or 'empty' per seat.""" + now = time.monotonic() + status = [] + for i in (0, 1): + c = self.slots[i] if len(self.slots) > i else None + if c is None: + status.append("empty") + elif now - c.last_seen > STALE_AFTER: + status.append("disconnected") + else: + status.append("connected") + return status + + async def broadcast(self): + self.last_seats = self.seat_status() + base = { + "type": "state", + "token": self.token, + "first": self.game.first, + "players": self.last_seats, + "spectators": sum(1 for c in self.slots[2:] if c is not None), + **self.game.to_dict(), + } + for c in self.connected(): + try: + await c.ws.send_json({**base, "seat": self.seat_of(c)}) + except (ConnectionError, RuntimeError): + pass # the reaper or the handler will clean it up + + +rooms: dict[str, Room] = {} + + +def normalize_code(raw): + """Case-insensitive, with surrounding and repeated whitespace ignored.""" + return " ".join(raw.split()).upper() + + +async def index(request): + return web.FileResponse(STATIC_DIR / "index.html") + + +async def game_socket(request): + code = normalize_code(request.query.get("game", "")) + ws = web.WebSocketResponse(max_msg_size=MAX_MESSAGE_SIZE) + await ws.prepare(request) + + if not 0 < len(code) <= MAX_CODE_LENGTH: + await ws.close(code=CLOSE_INVALID_CODE, + message=f"Codes must be 1-{MAX_CODE_LENGTH} characters".encode()) + return ws + room = rooms.get(code) + if room is None: + if len(rooms) >= MAX_ROOMS: + await ws.close(code=CLOSE_SERVER_FULL, message=b"Server is full, try again later") + return ws + room = rooms[code] = Room(code) + log.info("created room %r (%d rooms)", code, len(rooms)) + + client = Client(ws, request.query.get("client", "")[:64]) + room.join(client) + await room.broadcast() + + try: + async for msg in ws: + if msg.type != WSMsgType.TEXT: + continue + client.last_seen = time.monotonic() + try: + data = msg.json() + except ValueError: + continue + if not isinstance(data, dict): + continue + kind = data.get("type") + + if kind == "ping": + await ws.send_json({"type": "pong"}) + continue + if client not in room.slots: + break # reaped or replaced while this message was in flight + seat = room.seat_of(client) + if seat is None: + continue # spectators can't act + + if kind == "move": + if not all(room.players_present()): + error = "Waiting for an opponent" + else: + error = room.game.play(seat, data.get("cell")) + if error: + await ws.send_json({"type": "error", "message": error}) + else: + await room.broadcast() + elif kind == "rematch": + room.game.vote_rematch(seat) + await room.broadcast() + finally: + if room.leave(client) and not delete_if_empty(room): + await room.broadcast() + return ws + + +def delete_if_empty(room): + if room.connected(): + return False + if rooms.get(room.token) is room: + del rooms[room.token] + log.info("deleted empty room %r (%d rooms)", room.token, len(rooms)) + return True + + +async def reaper(app): + """Drop silent clients, announce seats going quiet, and delete abandoned rooms.""" + while True: + await asyncio.sleep(1) + now = time.monotonic() + for room in list(rooms.values()): + changed = False + for c in room.connected(): + if now - c.last_seen > HEARTBEAT_TIMEOUT: + room.leave(c) + asyncio.ensure_future(c.ws.close()) + changed = True + if delete_if_empty(room): + continue + if changed or room.seat_status() != room.last_seats: + await room.broadcast() + + +async def start_reaper(app): + task = asyncio.create_task(reaper(app)) + yield + task.cancel() + + +@web.middleware +async def no_cache(request, handler): + # Keep Cloudflare and browsers from serving stale copies after an update. + response = await handler(request) + if not isinstance(response, web.WebSocketResponse): + response.headers["Cache-Control"] = "no-cache" + return response + + +def make_app(): + app = web.Application(middlewares=[no_cache]) + app.router.add_get("/", index) + app.router.add_get("/ws", game_socket) + app.router.add_static("/static", STATIC_DIR) + app.cleanup_ctx.append(start_reaper) + return app + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Moonchess server") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8080) + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") + web.run_app(make_app(), host=args.host, port=args.port) diff --git a/solver.py b/solver.py new file mode 100644 index 0000000..1e565c3 --- /dev/null +++ b/solver.py @@ -0,0 +1,110 @@ +"""Perfect play for vanishing tic-tac-toe. + +Every position reachable from the empty board (~17k) is solved by retrograde analysis: +positions where the player to move can win immediately are wins, a position is a loss +once every move leads to an opponent win, and whatever is never resolved is a draw +(the players can keep cycling forever). Positions are (mine, theirs): the cells of the +player to move and of their opponent, each oldest first. +""" + +import random +from collections import deque + +from game import LINES, MAX_PIECES + +WIN, LOSS, DRAW = "win", "loss", "draw" + + +def _wins(cells): + return any(all(i in cells for i in line) for line in LINES) + + +def _moves(mine, theirs): + """(cell, next position) for each legal move; next position is None if the move wins. + The opponent's oldest piece goes as their turn starts if they have a full set.""" + occupied = set(mine) | set(theirs) + remaining = theirs[1:] if len(theirs) == MAX_PIECES else theirs + for cell in range(9): + if cell in occupied: + continue + placed = mine + (cell,) + yield cell, (None if _wins(placed) else (remaining, placed)) + + +class Solver: + def __init__(self): + self.value = {} + self.distance = {} # plies until the game ends with best play from both sides + self._solve() + + def _solve(self): + start = ((), ()) + succ, preds = {}, {} + queue = deque([start]) + succ[start] = None + while queue: + pos = queue.popleft() + succ[pos] = list(_moves(*pos)) + for _, nxt in succ[pos]: + if nxt is None: + continue + preds.setdefault(nxt, []).append(pos) + if nxt not in succ: + succ[nxt] = None + queue.append(nxt) + + unresolved = {pos: sum(n is not None for _, n in moves) for pos, moves in succ.items()} + queue = deque() + for pos, moves in succ.items(): + if any(n is None for _, n in moves): + self.value[pos], self.distance[pos] = WIN, 1 + queue.append(pos) + while queue: + pos = queue.popleft() + for prev in preds.get(pos, ()): + if prev in self.value: + continue + if self.value[pos] == LOSS: + self.value[prev], self.distance[prev] = WIN, self.distance[pos] + 1 + queue.append(prev) + else: + unresolved[prev] -= 1 + if unresolved[prev] == 0: + self.value[prev] = LOSS + self.distance[prev] = 1 + max(self.distance[n] for _, n in succ[prev]) + queue.append(prev) + for pos in succ: + self.value.setdefault(pos, DRAW) + + def best_moves(self, mine, theirs): + """All optimal cells: win fastest, else draw, else lose as slowly as possible. + Returns None for positions not reachable from the empty board.""" + if (mine, theirs) not in self.value: + return None + scored = [] + for cell, nxt in _moves(mine, theirs): + if nxt is None: + score = (3, 0) + elif self.value[nxt] == LOSS: + score = (2, -self.distance[nxt]) + elif self.value[nxt] == DRAW: + score = (1, 0) + else: + score = (0, self.distance[nxt]) + scored.append((score, cell)) + best = max(score for score, _ in scored) + return [cell for score, cell in scored if score == best] + + def choose(self, game, accuracy, rng=random): + """Pick a move for the player to move: optimal with probability `accuracy`, else random. + + Returns (cell, how): how is "perfect" if the roll chose an optimal move, "lucky" if + a random move happened to be optimal anyway, and "random" otherwise. + """ + mine = tuple(game.history[game.turn]) + theirs = tuple(game.history[1 - game.turn]) + best = self.best_moves(mine, theirs) or [] + if best and rng.random() < accuracy: + return rng.choice(best), "perfect" + cell = rng.choice([c for c in range(9) if game.board[c] is None]) + return cell, "lucky" if cell in best else "random" diff --git a/static/app.js b/static/app.js new file mode 100644 index 0000000..6cfd9d2 --- /dev/null +++ b/static/app.js @@ -0,0 +1,207 @@ +"use strict"; + +const HEARTBEAT_MS = 1000; +const TIMEOUT_MS = 4000; // no message from the server for this long -> reconnect +const RECONNECT_MS = 1000; +const CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"; // no 0/O, 1/I lookalikes +const CODE_LENGTH = 5; + +// Identifies this page load, so a reconnect can reclaim its own seat before the +// server notices the old connection died. A refresh gets a new id on purpose. +const clientId = Array.from(crypto.getRandomValues(new Uint8Array(8)), + b => b.toString(16).padStart(2, "0")).join(""); + +const $ = id => document.getElementById(id); +const board = createBoard($("board"), cell => send({ type: "move", cell })); + +let token = null; +let ws = null; +let lastMessage = 0; +let reconnectTimer = null; +let state = null; +let flashTimer = null; + +// ---------- lobby ---------- + +// Any string is a valid code, and the server creates the game on first join. +// This just picks a random one that's easy to read out loud. +$("create").addEventListener("click", () => { + const code = Array.from(crypto.getRandomValues(new Uint8Array(CODE_LENGTH)), + b => CODE_ALPHABET[b % CODE_ALPHABET.length]).join(""); + enterGame(code); +}); + +$("join-form").addEventListener("submit", e => { + e.preventDefault(); + const code = $("join-code").value.trim(); + if (code) enterGame(code); +}); + +function showLobby(error = "") { + leaveGame(); + $("game").hidden = true; + $("lobby").hidden = false; + $("lobby-error").textContent = error; + history.replaceState(null, "", "/"); +} + +// ---------- game ---------- + +function enterGame(code) { + token = code; + state = null; + board.reset(); + $("lobby").hidden = true; + $("game").hidden = false; + $("code").textContent = code; + $("status").textContent = "Connecting…"; + history.replaceState(null, "", "/?game=" + encodeURIComponent(code)); + connect(); +} + +function leaveGame() { + token = null; + clearTimeout(reconnectTimer); + if (ws) { + const old = ws; + ws = null; + old.close(); + } +} + +$("leave").addEventListener("click", () => showLobby()); + +$("copy").addEventListener("click", async () => { + const link = location.origin + "/?game=" + token; + try { + await navigator.clipboard.writeText(link); + flash("Link copied"); + } catch { + prompt("Copy this link:", link); // clipboard API needs https or localhost + } +}); + +$("rematch").addEventListener("click", () => send({ type: "rematch" })); + +// ---------- connection ---------- + +function connect() { + const proto = location.protocol === "https:" ? "wss:" : "ws:"; + const url = `${proto}//${location.host}/ws?game=${encodeURIComponent(token)}&client=${clientId}`; + const sock = new WebSocket(url); + ws = sock; + lastMessage = Date.now(); + + sock.onmessage = e => { + if (sock !== ws) return; + lastMessage = Date.now(); + const msg = JSON.parse(e.data); + if (msg.type === "state") { + render(msg); + } else if (msg.type === "error") { + flash(msg.message); + } + }; + + sock.onclose = e => { + if (sock !== ws) return; + ws = null; + if (e.code >= 4000) { // the server refused this code; retrying won't help + showLobby(e.reason || "Could not join that game."); + return; + } + scheduleReconnect(); + }; +} + +function scheduleReconnect() { + if (!token) return; + $("game").classList.add("offline"); + $("status").textContent = "Disconnected. Reconnecting…"; + clearTimeout(reconnectTimer); + reconnectTimer = setTimeout(connect, RECONNECT_MS); +} + +function send(msg) { + if (ws && ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify(msg)); +} + +setInterval(() => { + if (!ws) return; + if (Date.now() - lastMessage > TIMEOUT_MS) { + // The socket is probably dead; don't wait for the browser to agree. + const old = ws; + ws = null; + old.close(); + scheduleReconnect(); + return; + } + send({ type: "ping" }); +}, HEARTBEAT_MS); + +// ---------- rendering ---------- + +function render(s) { + state = s; + const { seat, turn, winner, players } = s; + $("code").textContent = s.token; + const bothHere = players[0] !== "empty" && players[1] !== "empty"; + $("game").classList.remove("offline"); + const myTurn = seat === turn && winner === null && bothHere; + + board.render(s, myTurn); + + for (const i of [0, 1]) { + const el = $("player-" + i); + el.querySelector(".who").textContent = + players[i] === "empty" ? "Empty seat" + : players[i] === "disconnected" ? "Disconnected" + : seat === i ? "You" : seat === null ? "Player" : "Opponent"; + el.classList.toggle("empty", players[i] === "empty"); + el.classList.toggle("gone", players[i] === "disconnected"); + el.classList.toggle("active", winner === null ? turn === i : winner === i); + } + + $("status").textContent = statusText(s, bothHere); + + const rematch = $("rematch"); + rematch.hidden = winner === null || seat === null; + if (!rematch.hidden) { + const mine = s.rematch[seat]; + const theirs = s.rematch[1 - seat]; + rematch.disabled = mine; + rematch.textContent = mine ? "Waiting for opponent…" : theirs ? "Opponent wants a rematch. Play again?" : "Rematch"; + } + + if (!flashTimer) $("meta").textContent = metaText(s); +} + +function statusText({ seat, turn, winner, players }, bothHere) { + if (winner !== null) { + if (seat === null) return `${MARKS[winner]} wins`; + return seat === winner ? "You win!" : "You lose"; + } + if (!bothHere) return seat === null ? "Waiting for players…" : "Waiting for an opponent…"; + if (seat !== turn && players[turn] === "disconnected") return `${MARKS[turn]} disconnected…`; + if (seat === null) return `${MARKS[turn]} to move`; + return seat === turn ? "Your turn" : `Waiting for ${MARKS[turn]}…`; +} + +function metaText({ seat, spectators }) { + const role = seat === null ? "Spectating" : `You are ${MARKS[seat]}`; + return spectators ? `${role} · ${spectators} watching` : role; +} + +function flash(text) { + $("meta").textContent = text; + clearTimeout(flashTimer); + flashTimer = setTimeout(() => { + flashTimer = null; + if (state) $("meta").textContent = metaText(state); + }, 2000); +} + +// ---------- start ---------- + +const initial = new URLSearchParams(location.search).get("game"); +if (initial) enterGame(initial); diff --git a/static/board.js b/static/board.js new file mode 100644 index 0000000..0d2d34d --- /dev/null +++ b/static/board.js @@ -0,0 +1,58 @@ +"use strict"; + +// Board rendering shared by the online game (app.js) and practice mode (practice.js). + +const MARKS = ["X", "O"]; + +const PIECE_SVG = [ + '', + '', +]; + +// Fills `container` with nine cells. `onPlay(cell)` fires when a playable cell is clicked. +function createBoard(container, onPlay) { + const cells = []; + let shown = null; // board as currently drawn; null means draw the next state without animation + + for (let i = 0; i < 9; i++) { + const cell = document.createElement("button"); + cell.className = "cell"; + cell.addEventListener("click", () => { + if (cell.classList.contains("playable")) onPlay(i); + }); + container.appendChild(cell); + cells.push(cell); + } + + return { + // `s` is a game state from the server; `canPlay` makes empty cells clickable. + render(s, canPlay) { + // Age of each piece within its owner's pieces: 0 = newest. Styled only when the + // board has the "freshness" class. + const ages = Array(9).fill(null); + for (const hist of s.history || []) { + hist.forEach((cell, i) => { ages[cell] = hist.length - 1 - i; }); + } + s.board.forEach((v, i) => { + const cell = cells[i]; + if (!shown || v !== shown[i]) { + cell.innerHTML = v === null ? "" : PIECE_SVG[v]; + cell.classList.toggle("placed", v !== null && shown !== null); + } + cell.classList.toggle("playable", canPlay && v === null); + cell.classList.toggle("doomed", s.doomed === i); + if (ages[i] === null) delete cell.dataset.age; + else cell.dataset.age = ages[i]; + cell.classList.toggle("win", !!(s.winLine && s.winLine.includes(i))); + cell.setAttribute("aria-label", + `Square ${i + 1}: ${v === null ? "empty" : MARKS[v]}${s.doomed === i ? " (vanishes next)" : ""}`); + }); + shown = s.board.slice(); + }, + + // Next render draws without the placement animation (e.g. after joining a game). + reset() { + shown = null; + }, + }; +} diff --git a/static/index.html b/static/index.html new file mode 100644 index 0000000..2bc15a7 --- /dev/null +++ b/static/index.html @@ -0,0 +1,52 @@ + + + + + + Moonchess + + + +
+
+

Moonchess

+

Tic-tac-toe, but you only get three pieces. Place a fourth and your oldest one vanishes first.

+ +
or enter any code
+
+ + +
+ +
+ + +
+ + + + diff --git a/static/practice.html b/static/practice.html new file mode 100644 index 0000000..c4afbc6 --- /dev/null +++ b/static/practice.html @@ -0,0 +1,48 @@ + + + + + + Moonchess Practice + + + +
+
+

Moonchess practice

+
+ +
+ + +
+

Click a side to play as it, any time. X moves first.

+ +

+ +
+ +
+ + +

Chance that each of Columbina's moves is perfect. Otherwise she plays a random empty square.

+
+ +
+ Spoilers +

+ +
+ +
+
+ + +
+

+
+
+ + + + diff --git a/static/practice.js b/static/practice.js new file mode 100644 index 0000000..38156b4 --- /dev/null +++ b/static/practice.js @@ -0,0 +1,137 @@ +"use strict"; + +const OPPONENT = "Columbina"; // the computer's display name +const THINK_MS = 450; +const SQUARE_NAMES = ["top left", "top middle", "top right", "middle left", "center", + "middle right", "bottom left", "bottom middle", "bottom right"]; // pause before the computer's move shows, so it's easy to follow + +const $ = id => document.getElementById(id); +const board = createBoard($("board"), cell => request({ cell })); +const EMPTY = { board: Array(9).fill(null), history: [[], []], turn: 0, winner: null, winLine: null, doomed: null }; + +let state = EMPTY; +let human = 0; // the computer plays the other side +let requestId = 0; // bumped whenever the situation changes, so stale replies are dropped +let thinkTimer = null; +let lastComputerMove = null; // {cell, how} from the server, shown under "Spoilers" +let past = []; // earlier {state, lastComputerMove}, oldest first, for undo + +// Sends the position to the server: with `cell` to play it, without to get a computer move. +async function request(extra) { + const id = ++requestId; + clearTimeout(thinkTimer); + board.render(state, false); // no second click while this move is on its way + try { + const res = await fetch("/api/play", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ history: state.history, turn: state.turn, accuracy: accuracy(), ...extra }), + }); + const body = await res.json(); + if (id !== requestId) return; + if (!res.ok) throw new Error(body.error || "Something went wrong"); + past.push({ state, lastComputerMove }); + state = body; + if (body.computerMove) lastComputerMove = body.computerMove; + $("meta").textContent = ""; + } catch (e) { + if (id === requestId) $("meta").textContent = e.message; + } + if (id === requestId) update(); +} + +// Renders, then lets the computer move if it's its turn. +function update() { + const computerTurn = state.winner === null && state.turn !== human; + render(computerTurn); + clearTimeout(thinkTimer); + if (computerTurn) thinkTimer = setTimeout(() => request({}), THINK_MS); +} + +function render(computerTurn) { + board.render(state, !computerTurn && state.winner === null); + for (const i of [0, 1]) { + const el = $("player-" + i); + el.querySelector(".who").textContent = i === human ? "You" : OPPONENT; + el.classList.toggle("active", state.winner === null ? state.turn === i : state.winner === i); + el.setAttribute("aria-pressed", String(i === human)); + } + $("status").textContent = + state.winner !== null ? (state.winner === human ? "You win!" : `${OPPONENT} wins`) + : computerTurn ? `${OPPONENT} is thinking…` : "Your turn"; + $("new-game").textContent = state.winner === null ? "New game" : "Play again"; + $("spoiler-text").innerHTML = spoilerText(); + $("undo").disabled = undoTarget() < 0; +} + +// Index in `past` of the latest earlier position where it was your turn, or -1. +function undoTarget() { + let i = past.length - 1; + while (i >= 0 && past[i].state.turn !== human) i--; + return i; +} + +// Back to your previous turn, taking back Columbina's reply (or her pending move) too. +function undo() { + const i = undoTarget(); + if (i < 0) return; + ({ state, lastComputerMove } = past[i]); + past.length = i; + requestId++; + $("meta").textContent = ""; + update(); +} + +function spoilerText() { + if (!lastComputerMove) return `${OPPONENT} hasn't moved yet.`; + const where = `${OPPONENT}'s last move (${SQUARE_NAMES[lastComputerMove.cell]})`; + return { + perfect: `${where} was perfect.`, + lucky: `${where} was random, but happened to be a perfect move anyway.`, + random: `${where} was random.`, + }[lastComputerMove.how]; +} + +function accuracy() { + return Number($("accuracy").value) / 100; +} + +for (const i of [0, 1]) { + $("player-" + i).addEventListener("click", () => { + if (human === i) return; + human = i; + requestId++; // drop any computer move in flight for the side you just took over + update(); + }); +} + +$("accuracy").addEventListener("input", () => { + $("accuracy-value").textContent = $("accuracy").value + "%"; +}); + +$("new-game").addEventListener("click", () => { + requestId++; + state = EMPTY; + lastComputerMove = null; + past = []; + board.reset(); + $("meta").textContent = ""; + update(); +}); + +$("undo").addEventListener("click", undo); + +$("freshness").addEventListener("change", e => { + $("board").classList.toggle("freshness", e.target.checked); +}); +$("board").classList.toggle("freshness", $("freshness").checked); + +document.addEventListener("keydown", e => { + if ((e.ctrlKey || e.metaKey) && !e.shiftKey && e.key.toLowerCase() === "z") { + e.preventDefault(); + undo(); + } +}); + +$("accuracy-value").textContent = $("accuracy").value + "%"; +update(); diff --git a/static/style.css b/static/style.css new file mode 100644 index 0000000..997b6e7 --- /dev/null +++ b/static/style.css @@ -0,0 +1,181 @@ +:root { + --bg: #f6f5f1; + --surface: #ffffff; + --border: #dedbd2; + --text: #1d1c1a; + --muted: #6f6c64; + --x: #d0543a; + --o: #2f6fb5; + --win: #e9c46a; + --error: #b3261e; + --radius: 14px; + color-scheme: light; +} + +@media (prefers-color-scheme: dark) { + :root { + --bg: #15161a; + --surface: #1f2026; + --border: #33353d; + --text: #ecebe6; + --muted: #9a988f; + --x: #f07a5f; + --o: #6aa6ea; + --win: #c9a64a; + --error: #ff8a80; + color-scheme: dark; + } +} + +* { box-sizing: border-box; } + +body { + margin: 0; + min-height: 100vh; + background: var(--bg); + color: var(--text); + font: 16px/1.5 system-ui, -apple-system, "Segoe UI", sans-serif; + display: flex; + justify-content: center; +} + +main { + width: 100%; + max-width: 420px; + padding: 48px 16px; +} + +h1 { font-size: 2.2rem; margin: 0 0 8px; letter-spacing: -0.02em; } +.lede { color: var(--muted); margin: 0 0 32px; } + +button { + font: inherit; + font-weight: 600; + padding: 12px 18px; + border-radius: 10px; + border: 1px solid var(--border); + background: var(--surface); + color: var(--text); + cursor: pointer; +} +button:hover { border-color: var(--muted); } +button.primary { background: var(--text); color: var(--bg); border-color: var(--text); width: 100%; } +button.primary:hover { opacity: 0.9; } +button.small { padding: 6px 12px; font-size: 0.875rem; } +button.ghost { background: transparent; } + +.divider { + display: flex; align-items: center; gap: 12px; + color: var(--muted); font-size: 0.875rem; margin: 28px 0 16px; +} +.divider::before, .divider::after { content: ""; flex: 1; height: 1px; background: var(--border); } + +#join-form { display: flex; gap: 8px; } +#join-code { + flex: 1; min-width: 0; + font: 600 1.25rem/1 ui-monospace, "SF Mono", Menlo, monospace; + letter-spacing: 0.05em; text-transform: uppercase; + padding: 12px 14px; + border-radius: 10px; border: 1px solid var(--border); + background: var(--surface); color: var(--text); +} +#join-code::placeholder { text-transform: none; letter-spacing: 0; font-family: system-ui, sans-serif; font-weight: 400; font-size: 1rem; } +#join-code:focus { outline: 2px solid var(--o); outline-offset: -1px; } + +.error { color: var(--error); min-height: 1.5em; margin: 12px 0 0; } + +.bar { display: flex; justify-content: space-between; align-items: center; gap: 8px; } +.code-wrap { display: flex; align-items: center; gap: 10px; min-width: 0; } +.label { color: var(--muted); font-size: 0.875rem; } +.code { + font: 700 1.3rem ui-monospace, "SF Mono", Menlo, monospace; letter-spacing: 0.1em; + min-width: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; +} +.code-wrap button, .bar > button { flex-shrink: 0; } + +.players { display: grid; grid-template-columns: 1fr 1fr; gap: 8px; margin: 24px 0 8px; } +.player { + display: flex; align-items: center; gap: 10px; + padding: 10px 12px; border-radius: 10px; + border: 1px solid var(--border); background: var(--surface); + font-size: 0.9rem; color: var(--muted); + transition: border-color 0.15s; +} +.player.active { border-color: var(--text); color: var(--text); } +.player.empty { border-style: dashed; } +.player.gone { border-style: dashed; } +.player.gone .who { color: var(--error); } +.player .mark { font-weight: 800; font-size: 1.1rem; } +.mark.x { color: var(--x); } +.mark.o { color: var(--o); } + +button.player { + font: inherit; font-size: 0.9rem; font-weight: 400; + text-align: left; width: 100%; cursor: pointer; +} +button.player[aria-pressed="true"] .who { font-weight: 600; } +.page-title { font-size: 1.4rem; margin: 0; } +.hint { color: var(--muted); font-size: 0.85rem; margin: 6px 0 0; } + +.difficulty { margin-top: 24px; } +.difficulty label { display: flex; justify-content: space-between; font-weight: 600; } +.difficulty output { font-variant-numeric: tabular-nums; } +.difficulty input { width: 100%; margin: 10px 0 0; accent-color: var(--text); } + +.spoiler { + margin-top: 20px; padding: 10px 12px; + border: 1px solid var(--border); border-radius: 10px; background: var(--surface); +} +.spoiler summary { cursor: pointer; font-weight: 600; } +.spoiler p { margin: 8px 0 0; color: var(--muted); } +.spoiler strong { color: var(--text); } + +.toggle { display: flex; align-items: center; gap: 8px; margin-top: 12px; cursor: pointer; user-select: none; } +.toggle input { width: 18px; height: 18px; margin: 0; accent-color: var(--text); cursor: pointer; } + +.status { font-size: 1.15rem; font-weight: 600; text-align: center; margin: 20px 0 16px; min-height: 1.5em; } + +.board { + /* minmax(0, 1fr) and absolutely positioned marks keep cell sizes independent of content */ + display: grid; gap: 8px; + grid-template-columns: repeat(3, minmax(0, 1fr)); + grid-template-rows: repeat(3, minmax(0, 1fr)); + aspect-ratio: 1; width: min(100%, 60vh); margin: 0 auto; +} +.cell { + position: relative; + padding: 0; border-radius: var(--radius); + border: 1px solid var(--border); background: var(--surface); + display: grid; place-items: center; + cursor: default; +} +.cell.playable { cursor: pointer; } +.cell.playable:hover { border-color: var(--muted); } +.cell.win { background: color-mix(in srgb, var(--win) 30%, var(--surface)); border-color: var(--win); } +.cell svg { position: absolute; inset: 21%; width: 58%; height: 58%; overflow: visible; } +.cell svg * { fill: none; stroke-width: 12; stroke-linecap: round; } +.cell .x-mark * { stroke: var(--x); } +.cell .o-mark * { stroke: var(--o); } +/* Freshness view: newest piece solid, older ones fading toward the doomed look */ +.board.freshness .cell[data-age="1"] svg { opacity: 0.65; } +.board.freshness .cell[data-age="2"] svg { opacity: 0.3; } +.cell.doomed svg { opacity: 0.3; animation: fade 1.6s ease-in-out infinite; } +.cell.placed svg { animation: pop 0.18s ease-out; } + +@keyframes fade { 50% { opacity: 0.12; } } +@keyframes pop { from { transform: scale(0.6); opacity: 0; } } + +@media (prefers-reduced-motion: reduce) { + .cell.doomed svg, .cell.placed svg { animation: none; } +} + +#game.offline .board, +#game.offline .players { opacity: 0.4; pointer-events: none; } +#game.offline .status { color: var(--error); } + +.footer { margin-top: 20px; } +.actions { display: flex; gap: 8px; } +.actions > button { flex: 1; } +.actions > button.primary { flex: 2; } +.actions > button:disabled { opacity: 0.45; cursor: default; border-color: var(--border); } +.meta { color: var(--muted); font-size: 0.875rem; text-align: center; margin: 12px 0 0; min-height: 1.5em; } diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_game.py b/tests/test_game.py new file mode 100644 index 0000000..1245f89 --- /dev/null +++ b/tests/test_game.py @@ -0,0 +1,92 @@ +import unittest + +from game import Game + + +def play_all(game, cells): + for cell in cells: + err = game.play(game.turn, cell) + assert err is None, (cell, err) + + +class GameTest(unittest.TestCase): + def test_plain_win(self): + g = Game() + play_all(g, [0, 3, 1, 4, 2]) + self.assertEqual(g.winner, 0) + self.assertEqual(g.win_line, [0, 1, 2]) + + def test_oldest_vanishes_as_turn_starts(self): + g = Game() + play_all(g, [0, 8, 1, 7, 5]) # X: 0,1,5 O: 8,7 + self.assertEqual(g.doomed(), 0) # X's oldest, dimmed during O's turn + play_all(g, [3]) # O: 8,7,3; X's turn starts + self.assertIsNone(g.board[0]) + self.assertEqual(g.history[0], [1, 5]) + self.assertEqual(g.doomed(), 8) # O's oldest is next + + def test_can_place_where_own_piece_vanished(self): + g = Game() + play_all(g, [0, 8, 1, 7, 5, 3]) + self.assertIsNone(g.play(0, 0)) + self.assertEqual(g.history[0], [1, 5, 0]) + self.assertIsNone(g.board[8]) # and O's 8 went as O's turn started + + def test_opponents_dimmed_piece_is_taken(self): + g = Game() + play_all(g, [0, 8, 1, 7, 5, 3]) + self.assertEqual(g.play(0, 8), "That square is taken") + + def test_vanishing_piece_does_not_count_toward_win(self): + g = Game() + # X: 0, 4, 2 ... then 1 would make 0-1-2, but 0 vanishes when 1 is placed. + play_all(g, [0, 3, 4, 6, 2, 7]) + self.assertIsNone(g.board[0]) + play_all(g, [1]) + self.assertIsNone(g.winner) + self.assertEqual(g.history[0], [4, 2, 1]) + + def test_win_with_rotated_pieces(self): + g = Game() + play_all(g, [0, 3, 1, 4, 8, 6]) # X: 0,1,8 O: 3,4,6 + play_all(g, [2]) # X: 1,8,2 (0 vanished) -> no line + self.assertIsNone(g.winner) + play_all(g, [5]) # O: 4,6,5 (3 vanished) + play_all(g, [0]) # X: 8,2,0 (1 vanished) + play_all(g, [3]) # O: 6,5,3 (4 vanished) + play_all(g, [4]) # X: 2,0,4 (8 vanished) + play_all(g, [7]) # O: 5,3,7 (6 vanished) + play_all(g, [6]) # X: 0,4,6 (2 vanished) + self.assertIsNone(g.winner) + play_all(g, [8]) # O: 3,7,8 (5 vanished) + play_all(g, [2]) # X: 4,6,2 (0 vanished) -> diagonal 2-4-6 + self.assertEqual(g.winner, 0) + self.assertEqual(g.win_line, [2, 4, 6]) + + def test_rejects_bad_moves(self): + g = Game() + self.assertEqual(g.play(1, 0), "Not your turn") + self.assertEqual(g.play(0, 9), "Invalid square") + self.assertEqual(g.play(0, "3"), "Invalid square") + play_all(g, [4]) + self.assertEqual(g.play(1, 4), "That square is taken") + + def test_rematch_needs_both_and_alternates_first(self): + g = Game() + play_all(g, [0, 3, 1, 4, 2]) + g.vote_rematch(0) + self.assertEqual(g.winner, 0) + g.vote_rematch(1) + self.assertIsNone(g.winner) + self.assertEqual(g.board, [None] * 9) + self.assertEqual(g.turn, 1) + + def test_rematch_ignored_mid_round(self): + g = Game() + g.vote_rematch(0) + g.vote_rematch(1) + self.assertEqual(g.first, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_practice.py b/tests/test_practice.py new file mode 100644 index 0000000..bf0ebba --- /dev/null +++ b/tests/test_practice.py @@ -0,0 +1,54 @@ +from aiohttp.test_utils import AioHTTPTestCase + +import practice + +EMPTY = {"history": [[], []], "turn": 0} + + +class PracticeTest(AioHTTPTestCase): + async def get_application(self): + return practice.make_app() + + async def play(self, body): + res = await self.client.post("/api/play", json=body) + return res.status, await res.json() + + async def test_page_served(self): + res = await self.client.get("/") + self.assertEqual(res.status, 200) + self.assertIn("practice.js", await res.text()) + + async def test_human_move(self): + status, s = await self.play({**EMPTY, "cell": 4}) + self.assertEqual(status, 200) + self.assertEqual(s["board"][4], 0) + self.assertEqual(s["history"], [[4], []]) + self.assertEqual(s["turn"], 1) + + async def test_perfect_computer_blocks(self): + for _ in range(10): + status, s = await self.play({"history": [[0, 1], [4]], "turn": 1, "accuracy": 1}) + self.assertEqual(status, 200) + self.assertEqual(s["computerMove"], {"cell": 2, "how": "perfect"}) + + async def test_computer_takes_the_win(self): + status, s = await self.play({"history": [[0, 1], [4, 8]], "turn": 0, "accuracy": 1}) + self.assertEqual(s["winner"], 0) + self.assertEqual(s["winLine"], [0, 1, 2]) + self.assertEqual(s["computerMove"], {"cell": 2, "how": "perfect"}) + + async def test_human_move_has_no_computer_move(self): + status, s = await self.play({**EMPTY, "cell": 4}) + self.assertIsNone(s["computerMove"]) + + async def test_rejects_bad_requests(self): + for body in [{}, [], {"history": [[0], [0]], "turn": 0}, + {**EMPTY, "cell": 9}, {**EMPTY, "accuracy": "lots"}]: + status, s = await self.play(body) + self.assertEqual(status, 400, body) + self.assertIn("error", s) + + async def test_no_moves_after_a_win(self): + status, s = await self.play({"history": [[0, 1, 2], [4, 5]], "turn": 1}) + self.assertEqual(status, 400) + self.assertEqual(s["error"], "The round is over") diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..bd86f64 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,180 @@ +import asyncio +import unittest +from unittest import mock + +from aiohttp.test_utils import AioHTTPTestCase + +import server + + +class ServerTest(AioHTTPTestCase): + async def get_application(self): + server.rooms.clear() + return server.make_app() + + async def new_game(self): + return "TESTGAME" + + async def join(self, token, client_id=""): + return await self.client.ws_connect("/ws", params={"game": token, "client": client_id}) + + async def next_state(self, ws): + while True: + msg = await asyncio.wait_for(ws.receive_json(), 2) + if msg["type"] == "state": + return msg + + async def settle(self, ws): + """Return the most recent state after pending broadcasts arrive.""" + state = await self.next_state(ws) + while True: + try: + msg = await asyncio.wait_for(ws.receive_json(), 0.2) + except asyncio.TimeoutError: + return state + if msg["type"] == "state": + state = msg + + async def test_any_code_creates_a_game(self): + a = await self.join(" friday night ") + s = await self.settle(a) + self.assertEqual(s["token"], "FRIDAY NIGHT") + self.assertEqual(s["seat"], 0) + b = await self.join("Friday Night") + self.assertEqual((await self.settle(b))["seat"], 1) + self.assertEqual(list(server.rooms), ["FRIDAY NIGHT"]) + + async def test_invalid_codes_are_rejected(self): + for code in ["", " ", "x" * (server.MAX_CODE_LENGTH + 1)]: + ws = await self.join(code) + await ws.receive() + self.assertEqual(ws.close_code, server.CLOSE_INVALID_CODE) + self.assertEqual(server.rooms, {}) + + async def test_oversized_and_odd_messages(self): + a = await self.join("junk") + await self.settle(a) + await a.send_str("[]") # valid JSON, but not an object: ignored + await a.send_json({"type": "ping"}) + self.assertEqual((await asyncio.wait_for(a.receive_json(), 2))["type"], "pong") + await a.send_str("x" * (server.MAX_MESSAGE_SIZE + 1)) + while not a.closed: + await asyncio.wait_for(a.receive(), 2) + self.assertEqual(a.close_code, 1009) # message too big + + async def test_empty_game_is_deleted_immediately(self): + a = await self.join("gone") + b = await self.join("gone") + await self.settle(a) + await a.send_json({"type": "move", "cell": 4}) + await self.settle(b) + await a.close() + await b.close() + await asyncio.sleep(0.1) + self.assertNotIn("GONE", server.rooms) + c = await self.join("gone") + self.assertEqual((await self.settle(c))["board"][4], None) + + async def test_game_survives_while_anyone_stays(self): + a = await self.join("keep") + b = await self.join("keep") + await self.settle(a) + await a.send_json({"type": "move", "cell": 4}) + await self.settle(b) + await a.close() + a = await self.join("keep") + self.assertEqual((await self.settle(a))["board"][4], 0) + + async def test_index_served(self): + res = await self.client.get("/") + self.assertEqual(res.status, 200) + self.assertIn("Moonchess", await res.text()) + + async def test_seats_and_spectators(self): + token = await self.new_game() + a = await self.join(token) + self.assertEqual((await self.settle(a))["seat"], 0) + b = await self.join(token) + self.assertEqual((await self.settle(b))["seat"], 1) + c = await self.join(token) + s = await self.settle(c) + self.assertIsNone(s["seat"]) + self.assertEqual(s["spectators"], 1) + + # spectators can't move + await c.send_json({"type": "move", "cell": 0}) + await b.send_json({"type": "ping"}) + await self.settle(a) + self.assertEqual(server.rooms[token].game.board, [None] * 9) + + # a player leaves; the spectator refreshes and takes the seat + await a.close() + s = await self.settle(b) + self.assertEqual(s["players"], ["empty", "connected"]) + await c.close() + d = await self.join(token) + self.assertEqual((await self.settle(d))["seat"], 0) + + async def test_moves_need_both_players(self): + token = await self.new_game() + a = await self.join(token) + await self.settle(a) + await a.send_json({"type": "move", "cell": 4}) + msg = await asyncio.wait_for(a.receive_json(), 2) + self.assertEqual(msg, {"type": "error", "message": "Waiting for an opponent"}) + + b = await self.join(token) + await self.settle(a) + await a.send_json({"type": "move", "cell": 4}) + s = await self.settle(b) + self.assertEqual(s["board"][4], 0) + self.assertEqual(s["turn"], 1) + + async def test_reconnect_reclaims_own_slot(self): + token = await self.new_game() + a = await self.join(token, "page-a") + await self.settle(a) + b = await self.join(token, "page-b") + await self.settle(b) + # a's old socket is still "alive" from the server's view; same page reconnects + a2 = await self.join(token, "page-a") + s = await self.settle(a2) + self.assertEqual(s["seat"], 0) + self.assertEqual(s["spectators"], 0) + + async def test_silent_client_is_dropped(self): + token = await self.new_game() + with mock.patch.object(server, "HEARTBEAT_TIMEOUT", 0.5): + a = await self.join(token) + await self.settle(a) + b = await self.join(token) + await self.settle(b) + # b keeps pinging, a goes silent + for _ in range(8): + await b.send_json({"type": "ping"}) + await asyncio.sleep(0.25) + self.assertEqual(server.rooms[token].players_present(), [False, True]) + + async def test_silent_seat_shows_disconnected_then_recovers(self): + token = await self.new_game() + with mock.patch.object(server, "STALE_AFTER", 0.5): + a = await self.join(token) + await self.settle(a) + b = await self.join(token) + await self.settle(b) + for _ in range(8): # b keeps pinging, a goes quiet + await b.send_json({"type": "ping"}) + await asyncio.sleep(0.25) + s = await self.settle(b) + self.assertEqual(s["players"], ["disconnected", "connected"]) + + for _ in range(5): # a comes back + await a.send_json({"type": "ping"}) + await b.send_json({"type": "ping"}) + await asyncio.sleep(0.25) + s = await self.settle(b) + self.assertEqual(s["players"], ["connected", "connected"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_solver.py b/tests/test_solver.py new file mode 100644 index 0000000..e4b92e9 --- /dev/null +++ b/tests/test_solver.py @@ -0,0 +1,96 @@ +import random +import unittest + +from game import Game +from solver import DRAW, LOSS, WIN, Solver, _moves + +solver = Solver() + + +def game_at(mine, theirs): + """The Game for a solver position, with `mine` as the player to move (seat 0).""" + return Game.from_history([list(mine), list(theirs)], 0) + + +class SolverTest(unittest.TestCase): + def test_perfect_play_is_a_draw(self): + self.assertEqual(solver.value[((), ())], DRAW) + self.assertEqual(sorted(solver.best_moves((), ())), list(range(9))) + + def test_nobody_can_be_forced_to_lose(self): + # With only two of your pieces down while placing, you can make one threat at a time. + self.assertNotIn(LOSS, solver.value.values()) + + def test_blocks_a_threat(self): + # X: 0, 1 threatens 2 + self.assertEqual(solver.best_moves((4,), (0, 1)), [2]) + + def test_blocks_where_own_piece_just_vanished(self): + # X's newest two (0, 1) threaten 2, where O's oldest piece was until O's turn started + self.assertEqual(solver.best_moves((5, 6), (7, 0, 1)), [2]) + + def test_takes_an_immediate_win(self): + # X: 0, 1 with 2 open -> 2 wins now + self.assertEqual(solver.best_moves((0, 1), (4, 8)), [2]) + + def test_moves_match_game_rules(self): + rng = random.Random(1) + positions = rng.sample(sorted(solver.value), 3000) + for mine, theirs in positions: + for cell, nxt in _moves(mine, theirs): + game = game_at(mine, theirs) + self.assertIsNone(game.play(0, cell)) + if nxt is None: + self.assertEqual(game.winner, 0) + else: + self.assertIsNone(game.winner) + self.assertEqual(nxt, (tuple(game.history[1]), tuple(game.history[0]))) + + def test_values_are_consistent(self): + rng = random.Random(2) + for pos in rng.sample(sorted(solver.value), 3000): + children = [solver.value[n] if n else "immediate" for _, n in _moves(*pos)] + value = solver.value[pos] + if value == WIN: + self.assertTrue("immediate" in children or LOSS in children) + elif value == LOSS: + self.assertTrue(all(c == WIN for c in children)) + else: + self.assertNotIn(LOSS, children) + self.assertIn(DRAW, children) + + def test_accuracy_extremes(self): + game = game_at((0, 1), (4, 8)) + rng = random.Random(3) + self.assertTrue(all(solver.choose(game, 1.0, rng) == (2, "perfect") for _ in range(50))) + picks = {solver.choose(game, 0.0, rng) for _ in range(200)} + self.assertEqual(picks, {(2, "lucky"), (3, "random"), (5, "random"), + (6, "random"), (7, "random")}) + + +class FromHistoryTest(unittest.TestCase): + def test_rebuilds_board_and_winner(self): + g = Game.from_history([[0, 1, 2], [4, 5]], 1) + self.assertEqual(g.board[:3], [0, 0, 0]) + self.assertEqual(g.winner, 0) + self.assertEqual(g.win_line, [0, 1, 2]) + + def test_rejects_invalid(self): + for history, turn in [ + ([[0, 0], []], 0), # duplicate cell + ([[0], [0]], 0), # overlap + ([[9], []], 0), # out of range + ([[0, 1, 2, 3], []], 0), # too many pieces + ([[0, 1, 5], [3, 4, 8]], 0), # mover's oldest should have vanished + ([["a"], []], 0), + ([[True], []], 0), + ([[0]], 0), # one seat only + ([[], []], 2), # bad turn + (None, 0), + ]: + with self.assertRaises(ValueError, msg=(history, turn)): + Game.from_history(history, turn) + + +if __name__ == "__main__": + unittest.main()