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.
124 lines
5.4 KiB
Python
124 lines
5.4 KiB
Python
"""Game lobby route tests via kaya's ASGI transport."""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from httpx import ASGITransport, AsyncClient
|
|
from pwo import async_test
|
|
|
|
from tavolo.app import app
|
|
from tests.helpers import oidc_user
|
|
|
|
|
|
class GamesRouteTest(unittest.TestCase):
|
|
@async_test
|
|
async def test_create_requires_auth(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
response = await client.post("/api/games", json={})
|
|
self.assertEqual(401, response.status_code)
|
|
self.assertEqual({"error": "unauthenticated"}, response.json())
|
|
|
|
@async_test
|
|
async def test_create_and_read_lobby(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
with oidc_user("alice"):
|
|
created = await client.post("/api/games", json={"target_score": 16})
|
|
self.assertEqual(201, created.status_code)
|
|
body = created.json()
|
|
self.assertEqual("lobby", body["phase"])
|
|
self.assertEqual(3, body["seats_open"])
|
|
self.assertEqual(16, body["target_score"])
|
|
self.assertEqual(6, len(body["join_code"]))
|
|
game_id = body["id"]
|
|
|
|
with oidc_user("alice"):
|
|
snapshot = await client.get(f"/api/games/{game_id}")
|
|
self.assertEqual(200, snapshot.status_code)
|
|
self.assertEqual("alice", snapshot.json()["players"][0]["sub"])
|
|
|
|
with oidc_user("mallory"):
|
|
forbidden = await client.get(f"/api/games/{game_id}")
|
|
self.assertEqual(403, forbidden.status_code)
|
|
|
|
@async_test
|
|
async def test_join_fills_seats_and_starts_game(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
with oidc_user("alice"):
|
|
created = await client.post("/api/games", json={})
|
|
code = created.json()["join_code"]
|
|
|
|
for player in ("bob", "carol"):
|
|
with oidc_user(player):
|
|
joined = await client.post("/api/games/join", json={"code": code})
|
|
self.assertEqual(200, joined.status_code)
|
|
self.assertEqual("lobby", joined.json()["phase"])
|
|
|
|
with oidc_user("dave"):
|
|
started = await client.post("/api/games/join", json={"code": code})
|
|
self.assertEqual(200, started.status_code)
|
|
state = started.json()
|
|
self.assertEqual("playing", state["phase"])
|
|
self.assertEqual(4, len(state["players"]))
|
|
self.assertEqual([], state["table"])
|
|
for participant in state["players"]:
|
|
self.assertEqual(10, participant["cards_left"])
|
|
# The view is personalized to Dave: he sees his own hand in
|
|
# seat 3 but not Alice's in seat 0.
|
|
self.assertIn("hand", state["players"][3])
|
|
self.assertNotIn("hand", state["players"][0])
|
|
self.assertEqual(1, state["turn"])
|
|
|
|
@async_test
|
|
async def test_join_errors(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
with oidc_user("alice"):
|
|
created = await client.post("/api/games", json={})
|
|
code = created.json()["join_code"]
|
|
|
|
with oidc_user("bob"):
|
|
unknown = await client.post("/api/games/join", json={"code": "ZZZZZZ"})
|
|
self.assertEqual(404, unknown.status_code)
|
|
|
|
with oidc_user("alice"):
|
|
duplicate = await client.post("/api/games/join", json={"code": code})
|
|
self.assertEqual(409, duplicate.status_code)
|
|
|
|
with oidc_user("bob"):
|
|
missing = await client.post("/api/games/join", json={})
|
|
self.assertEqual(400, missing.status_code)
|
|
|
|
for player in ("bob", "carol", "dave"):
|
|
with oidc_user(player):
|
|
await client.post("/api/games/join", json={"code": code})
|
|
with oidc_user("erin"):
|
|
late = await client.post("/api/games/join", json={"code": code})
|
|
self.assertEqual(409, late.status_code)
|
|
|
|
@async_test
|
|
async def test_create_rejects_bad_target_score(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
with oidc_user("alice"):
|
|
zero = await client.post("/api/games", json={"target_score": 0})
|
|
text = await client.post("/api/games", json={"target_score": "eleven"})
|
|
huge = await client.post("/api/games", json={"target_score": 1000})
|
|
self.assertEqual(400, zero.status_code)
|
|
self.assertEqual(400, text.status_code)
|
|
self.assertEqual(400, huge.status_code)
|
|
|
|
@async_test
|
|
async def test_get_unknown_game(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
with oidc_user("alice"):
|
|
response = await client.get("/api/games/does-not-exist")
|
|
self.assertEqual(404, response.status_code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|