moonchess/tests/test_server.py

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