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