woggioni
6932a3272c
CI / Build and push docker image (push) Successful in 3m12s
The platform now hosts multiple card games, with scopone scientifico as the first one. Rename the brand wherever it is not a game rule: - move the Python package to server/src/tavolo and update imports - rename the Postgres database/user, OIDC issuer path, client id and Redis key prefixes to tavolo (clean break: existing pgdata volumes and live games are not migrated) - rename the Cargo package to tavolo-web and set the page title to Tavolo - update docs and the Docker image path to woggioni/tavolo The scopa game term (clearing the table) in the engine, state and web UI is intentionally left untouched.
288 lines
14 KiB
Python
288 lines
14 KiB
Python
"""WebSocket live-play tests via kaya's ASGI websocket transport."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
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 tavolo.app import app, game_store
|
|
from tavolo.game import engine
|
|
from tavolo.game.state import Card, GameState, PlayerState
|
|
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)
|
|
|
|
|
|
async def _seed_last_play_state(hand_ack_timeout: int = 30) -> str:
|
|
"""Seed a game where a single play ends the hand: p0 holds the only
|
|
card left and can capture the only table card."""
|
|
state = GameState(
|
|
id="hand-end-1",
|
|
join_code="HEND01",
|
|
creator_sub="alice",
|
|
target_score=11,
|
|
phase="playing",
|
|
turn=0,
|
|
table=[Card.parse("02C")],
|
|
)
|
|
state.players = [
|
|
PlayerState(sub="alice", name="Alice", seat=0, hand=[Card.parse("02D")]),
|
|
PlayerState(sub="bob", name="Bob", seat=1),
|
|
PlayerState(sub="carol", name="Carol", seat=2),
|
|
PlayerState(sub="dave", name="Dave", seat=3),
|
|
]
|
|
state.hand_ack_timeout = hand_ack_timeout
|
|
await game_store.save(state)
|
|
return state.id
|
|
|
|
|
|
class HandEndWebSocketTest(unittest.TestCase):
|
|
@async_test
|
|
async def test_hand_end_ack_flow(self) -> None:
|
|
import contextlib
|
|
|
|
game_id = await _seed_last_play_state()
|
|
ws_transport = ASGIWebSocketTransport(app=app)
|
|
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
|
|
async with contextlib.AsyncExitStack() as stack:
|
|
with ws_users([make_user(name) for name in PLAYERS]):
|
|
sockets = [
|
|
await stack.enter_async_context(
|
|
aconnect_ws(f"/ws/games/{game_id}", ws_client)
|
|
)
|
|
for _ in PLAYERS
|
|
]
|
|
for ws in sockets:
|
|
await ws.receive_json() # initial state
|
|
|
|
# Alice plays the last card: the hand ends and the game
|
|
# pauses for acknowledgements.
|
|
await sockets[0].send_json(
|
|
{"action": "play", "card": "02D", "capture": ["02C"]}
|
|
)
|
|
summaries = [await ws.receive_json() for ws in sockets]
|
|
for summary in summaries:
|
|
self.assertEqual("state", summary["type"])
|
|
self.assertEqual("hand_end", summary["game"]["phase"])
|
|
self.assertEqual([], summary["game"]["acknowledged"])
|
|
self.assertIsNotNone(summary["game"]["hand_end_deadline"])
|
|
award = summary["game"]["last_hand"]["award"]
|
|
self.assertEqual("A", award["carte"])
|
|
self.assertEqual("A", award["denara"])
|
|
|
|
# Everyone acknowledges; the fourth ack deals the next hand.
|
|
for i, ws in enumerate(sockets):
|
|
await ws.send_json({"action": "ack"})
|
|
updates = [await other.receive_json() for other in sockets]
|
|
for update in updates:
|
|
if i < 3:
|
|
self.assertEqual("hand_end", update["game"]["phase"])
|
|
self.assertEqual(
|
|
list(range(i + 1)),
|
|
update["game"]["acknowledged"],
|
|
)
|
|
else:
|
|
self.assertEqual("playing", update["game"]["phase"])
|
|
self.assertEqual(2, update["game"]["hand_number"])
|
|
self.assertEqual(
|
|
10, update["game"]["players"][i]["cards_left"]
|
|
)
|
|
|
|
@async_test
|
|
async def test_hand_end_timeout_deals_next_hand(self) -> None:
|
|
game_id = await _seed_last_play_state(hand_ack_timeout=1)
|
|
ws_transport = ASGIWebSocketTransport(app=app)
|
|
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
|
|
with ws_users([make_user("alice")]):
|
|
async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
|
|
await ws.receive_json() # initial state
|
|
await ws.send_json(
|
|
{"action": "play", "card": "02D", "capture": ["02C"]}
|
|
)
|
|
summary = await ws.receive_json()
|
|
self.assertEqual("hand_end", summary["game"]["phase"])
|
|
# Nobody acks: the timer must deal the next hand.
|
|
update = None
|
|
for _ in range(20):
|
|
try:
|
|
update = await asyncio.wait_for(
|
|
ws.receive_json(), timeout=2
|
|
)
|
|
except asyncio.TimeoutError:
|
|
break
|
|
if (
|
|
update.get("type") == "state"
|
|
and update["game"]["phase"] == "playing"
|
|
):
|
|
break
|
|
self.assertIsNotNone(update)
|
|
assert update is not None
|
|
self.assertEqual("playing", update["game"]["phase"])
|
|
self.assertEqual(2, update["game"]["hand_number"])
|
|
|
|
|
|
class TurnTimeoutWebSocketTest(unittest.TestCase):
|
|
@async_test
|
|
async def test_turn_timeout_auto_plays_a_card(self) -> None:
|
|
state = engine.create_game(
|
|
"turn-timeout-1", "TT0001", "alice", "Alice",
|
|
target_score=11, turn_timeout=1,
|
|
)
|
|
for name in PLAYERS[1:]:
|
|
engine.join_game(state, name, name.capitalize())
|
|
await game_store.save(state)
|
|
|
|
ws_transport = ASGIWebSocketTransport(app=app)
|
|
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
|
|
with ws_users([make_user("alice")]):
|
|
async with aconnect_ws(f"/ws/games/{state.id}", ws_client) as ws:
|
|
first = await ws.receive_json()
|
|
# Bob (seat 1) is first to act and never connects.
|
|
self.assertEqual(1, first["game"]["turn"])
|
|
deadline = first["game"]["turn_deadline"]
|
|
self.assertIsNotNone(deadline)
|
|
|
|
# Nobody plays: the timer must play a random card for Bob.
|
|
update = None
|
|
for _ in range(20):
|
|
try:
|
|
update = await asyncio.wait_for(
|
|
ws.receive_json(), timeout=2
|
|
)
|
|
except asyncio.TimeoutError:
|
|
break
|
|
if (
|
|
update.get("type") == "state"
|
|
and update["game"]["turn"] == 2
|
|
):
|
|
break
|
|
self.assertIsNotNone(update)
|
|
assert update is not None
|
|
self.assertEqual(2, update["game"]["turn"])
|
|
self.assertEqual(1, update["game"]["last_move"]["seat"])
|
|
self.assertEqual(
|
|
9, update["game"]["players"][1]["cards_left"]
|
|
)
|
|
self.assertNotEqual(deadline, update["game"]["turn_deadline"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|