tavolo/server/tests/test_websocket.py

"""WebSocket live-play tests via kaya's ASGI websocket transport."""
from __future__ import annotations

import unittest

from httpx import ASGITransport, AsyncClient
from httpx_ws import WebSocketDisconnect, aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport
from pwo import async_test

from scopa.app import app
from tests.helpers import make_user, oidc_user, ws_users

PLAYERS = ("alice", "bob", "carol", "dave")


class WebSocketTest(unittest.TestCase):
    async def _started_game(self, client: AsyncClient) -> dict:
        """Create a game and seat four players; return the playing state."""
        with oidc_user("alice"):
            created = await client.post("/api/games", json={})
        code = created.json()["join_code"]
        response = created
        for player in PLAYERS[1:]:
            with oidc_user(player):
                response = await client.post("/api/games/join", json={"code": code})
        return response.json()

    async def _bob_view(self, client: AsyncClient, game_id: str) -> dict:
        with oidc_user("bob"):
            return (await client.get(f"/api/games/{game_id}")).json()

    @async_test
    async def test_move_updates_all_connections(self) -> None:
        api_transport = ASGITransport(app=app)
        async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
            state = await self._started_game(client)
            game_id = state["id"]
            bob_hand = (await self._bob_view(client, game_id))["players"][1]["hand"]

            ws_transport = ASGIWebSocketTransport(app=app)
            async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
                with ws_users([make_user("bob"), make_user("alice")]):
                    async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as bob_ws:
                        first = await bob_ws.receive_json()
                        self.assertEqual("state", first["type"])
                        self.assertEqual(1, first["game"]["turn"])
                        self.assertTrue(first["game"].get("your_turn"))

                        async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as alice_ws:
                            alice_first = await alice_ws.receive_json()
                            self.assertEqual("state", alice_first["type"])
                            self.assertEqual(
                                "alice", alice_first["game"]["players"][0]["sub"]
                            )
                            self.assertNotIn("hand", alice_first["game"]["players"][1])

                            await bob_ws.send_json(
                                {"action": "play", "card": bob_hand[0]}
                            )
                            bob_update = await bob_ws.receive_json()
                            alice_update = await alice_ws.receive_json()

                        for update in (bob_update, alice_update):
                            self.assertEqual("state", update["type"])
                            self.assertEqual(2, update["game"]["turn"])
                            self.assertEqual(1, len(update["game"]["table"]))

    @async_test
    async def test_illegal_move_returns_error(self) -> None:
        api_transport = ASGITransport(app=app)
        async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
            state = await self._started_game(client)
            game_id = state["id"]
            bob_hand = (await self._bob_view(client, game_id))["players"][1]["hand"]

            ws_transport = ASGIWebSocketTransport(app=app)
            async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
                with ws_users([make_user("bob")]):
                    async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as bob_ws:
                        await bob_ws.receive_json()
                        await bob_ws.send_json(
                            {"action": "play", "card": bob_hand[0]}
                        )
                        await bob_ws.receive_json()  # the resulting state
                        # Bob cannot play twice in a row.
                        await bob_ws.send_json(
                            {"action": "play", "card": bob_hand[1]}
                        )
                        error = await bob_ws.receive_json()
                        self.assertEqual("error", error["type"])
                        self.assertEqual("illegal_move", error["code"])

    @async_test
    async def test_unknown_game_is_closed(self) -> None:
        ws_transport = ASGIWebSocketTransport(app=app)
        async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
            with ws_users([make_user("alice")]):
                with self.assertRaises(WebSocketDisconnect) as caught:
                    async with aconnect_ws("/ws/games/no-such-game", ws_client):
                        pass
                self.assertEqual(4404, caught.exception.code)

    @async_test
    async def test_non_player_is_closed(self) -> None:
        api_transport = ASGITransport(app=app)
        async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
            state = await self._started_game(client)
            game_id = state["id"]

            ws_transport = ASGIWebSocketTransport(app=app)
            async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
                with ws_users([make_user("mallory")]):
                    with self.assertRaises(WebSocketDisconnect) as caught:
                        async with aconnect_ws(f"/ws/games/{game_id}", ws_client):
                            pass
                    self.assertEqual(4403, caught.exception.code)

    @async_test
    async def test_unauthenticated_is_closed(self) -> None:
        ws_transport = ASGIWebSocketTransport(app=app)
        async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
            with ws_users([]):
                with self.assertRaises(WebSocketDisconnect) as caught:
                    async with aconnect_ws("/ws/games/whatever", ws_client):
                        pass
                self.assertEqual(4401, caught.exception.code)


if __name__ == "__main__":
    unittest.main()