From c52f30dc59691cdc0d54d146dd0c72199c4eeded Mon Sep 17 00:00:00 2001 From: tukaunu Date: Mon, 28 Sep 2026 03:36:46 +0200 Subject: [PATCH] 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 --- .gitignore | 4 + README.md | 109 ++++++++++++++++++ game.py | 111 +++++++++++++++++++ practice.py | 60 ++++++++++ requirements.txt | 1 + server.py | 243 +++++++++++++++++++++++++++++++++++++++++ solver.py | 110 +++++++++++++++++++ static/app.js | 207 +++++++++++++++++++++++++++++++++++ static/board.js | 58 ++++++++++ static/index.html | 52 +++++++++ static/practice.html | 48 ++++++++ static/practice.js | 137 +++++++++++++++++++++++ static/style.css | 181 ++++++++++++++++++++++++++++++ tests/__init__.py | 0 tests/test_game.py | 92 ++++++++++++++++ tests/test_practice.py | 54 +++++++++ tests/test_server.py | 180 ++++++++++++++++++++++++++++++ tests/test_solver.py | 96 ++++++++++++++++ 18 files changed, 1743 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 game.py create mode 100644 practice.py create mode 100644 requirements.txt create mode 100644 server.py create mode 100644 solver.py create mode 100644 static/app.js create mode 100644 static/board.js create mode 100644 static/index.html create mode 100644 static/practice.html create mode 100644 static/practice.js create mode 100644 static/style.css create mode 100644 tests/__init__.py create mode 100644 tests/test_game.py create mode 100644 tests/test_practice.py create mode 100644 tests/test_server.py create mode 100644 tests/test_solver.py 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()