woggioni
b5219062f2
CI / Build and push docker image (push) Successful in 3m0s
Anonymous users and logged-in non-players can now open /game/<id> and
follow the match: they see the table, the score and each player's
captured cards and scope, but never a hand and never the join code.
Spectator actions are rejected with a forbidden error.
Backend (tavolo-platform / tavolo-scopone):
- /ws/games/{id} accepts anonymous and non-seated connections as
spectators; only unknown games are refused. The state payload carries
spectator: true and withholds join_code unless the viewer is seated.
- GET /api/games/{id} is public and returns the same personalized
(player) or read-only (spectator) view.
- GameEngine.view_for/game_over_view take Optional[str] (None = spectator)
and are documented to hide privileged information for non-seated
viewers.
Frontend (Sycamore):
- Parse the spectator flag and render a spectator table: card backs plus
captured/scope for every seat, no clickable hand, "Spectating" badge.
- A pre-start spectator sees a "game not started yet" waiting room with
the watch link and no join code.
- "Copy watch link" buttons in the waiting room and the game HUD.
- The hand-end summary carries no acknowledge button for spectators and
the game-over overlay hides "My matches".
Tests and docs updated.
305 lines
14 KiB
Python
305 lines
14 KiB
Python
"""Game lobby route tests via kaya's ASGI transport (scopone game)."""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from tavolo.app import app
|
|
from tests.helpers import async_test, 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={"options": {"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"])
|
|
|
|
# A logged-in non-player is a read-only spectator: no join
|
|
# code, and the view is marked as a spectator view.
|
|
with oidc_user("mallory"):
|
|
watched = await client.get(f"/api/games/{game_id}")
|
|
self.assertEqual(200, watched.status_code)
|
|
self.assertTrue(watched.json()["spectator"])
|
|
self.assertNotIn("join_code", watched.json())
|
|
|
|
# Anonymous users may watch too.
|
|
anonymous = await client.get(f"/api/games/{game_id}")
|
|
self.assertEqual(200, anonymous.status_code)
|
|
self.assertTrue(anonymous.json()["spectator"])
|
|
self.assertNotIn("join_code", anonymous.json())
|
|
|
|
@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()
|
|
game_id = state["id"]
|
|
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"])
|
|
|
|
# A spectator sees the table, the score and each player's
|
|
# captured count and scope — but no hand at all.
|
|
watched = await client.get(f"/api/games/{game_id}")
|
|
self.assertEqual(200, watched.status_code)
|
|
spect = watched.json()
|
|
self.assertTrue(spect["spectator"])
|
|
self.assertNotIn("join_code", spect)
|
|
self.assertEqual([], spect["table"])
|
|
self.assertEqual({"A": 0, "B": 0}, spect["scores"])
|
|
for participant in spect["players"]:
|
|
self.assertNotIn("hand", participant)
|
|
self.assertEqual(10, participant["cards_left"])
|
|
self.assertEqual(0, participant["captured_count"])
|
|
self.assertEqual(0, participant["scope"])
|
|
|
|
@async_test
|
|
async def test_create_napola_option(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
with oidc_user("alice"):
|
|
default = await client.post("/api/games", json={})
|
|
self.assertEqual(201, default.status_code)
|
|
self.assertTrue(default.json()["napola"])
|
|
|
|
with oidc_user("alice"):
|
|
disabled = await client.post(
|
|
"/api/games", json={"options": {"napola": False}}
|
|
)
|
|
self.assertEqual(201, disabled.status_code)
|
|
self.assertFalse(disabled.json()["napola"])
|
|
|
|
with oidc_user("alice"):
|
|
invalid = await client.post(
|
|
"/api/games", json={"options": {"napola": "yes"}}
|
|
)
|
|
self.assertEqual(400, invalid.status_code)
|
|
|
|
@async_test
|
|
async def test_create_asso_pigliatutto_option(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
with oidc_user("alice"):
|
|
default = await client.post("/api/games", json={})
|
|
self.assertEqual(201, default.status_code)
|
|
self.assertFalse(default.json()["asso_pigliatutto"])
|
|
|
|
with oidc_user("alice"):
|
|
enabled = await client.post(
|
|
"/api/games", json={"options": {"asso_pigliatutto": True}}
|
|
)
|
|
self.assertEqual(201, enabled.status_code)
|
|
self.assertTrue(enabled.json()["asso_pigliatutto"])
|
|
|
|
with oidc_user("alice"):
|
|
invalid = await client.post(
|
|
"/api/games", json={"options": {"asso_pigliatutto": "yes"}}
|
|
)
|
|
self.assertEqual(400, invalid.status_code)
|
|
|
|
@async_test
|
|
async def test_preview_join_code(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={
|
|
"options": {
|
|
"target_score": 16,
|
|
"napola": False,
|
|
"asso_pigliatutto": True,
|
|
}
|
|
},
|
|
)
|
|
code = created.json()["join_code"]
|
|
|
|
response = await client.get(f"/api/games/preview?code={code}")
|
|
self.assertEqual(401, response.status_code)
|
|
|
|
with oidc_user("bob"):
|
|
missing = await client.get("/api/games/preview")
|
|
unknown = await client.get("/api/games/preview?code=ZZZZZZ")
|
|
self.assertEqual(400, missing.status_code)
|
|
self.assertEqual(404, unknown.status_code)
|
|
|
|
with oidc_user("bob"):
|
|
preview = await client.get(f"/api/games/preview?code={code}")
|
|
self.assertEqual(200, preview.status_code)
|
|
body = preview.json()
|
|
self.assertEqual("scopone_scientifico", body["game_type"])
|
|
self.assertEqual("Scopone scientifico", body["game_name"])
|
|
self.assertTrue(body["game_description"])
|
|
self.assertEqual("lobby", body["phase"])
|
|
self.assertEqual(16, body["target_score"])
|
|
self.assertFalse(body["napola"])
|
|
self.assertTrue(body["asso_pigliatutto"])
|
|
self.assertEqual(3, body["seats_open"])
|
|
self.assertEqual(["alice"], [p["sub"] for p in body["players"]])
|
|
|
|
for player in ("bob", "carol", "dave"):
|
|
with oidc_user(player):
|
|
await client.post("/api/games/join", json={"code": code})
|
|
with oidc_user("bob"):
|
|
started = await client.get(f"/api/games/preview?code={code}")
|
|
self.assertEqual(200, started.status_code)
|
|
self.assertEqual("playing", started.json()["phase"])
|
|
self.assertEqual(0, started.json()["seats_open"])
|
|
|
|
@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={"options": {"target_score": 0}}
|
|
)
|
|
text = await client.post(
|
|
"/api/games", json={"options": {"target_score": "eleven"}}
|
|
)
|
|
huge = await client.post(
|
|
"/api/games", json={"options": {"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)
|
|
|
|
|
|
class GameTypesRouteTest(unittest.TestCase):
|
|
@async_test
|
|
async def test_lists_available_game_types(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
response = await client.get("/api/game-types")
|
|
self.assertEqual(200, response.status_code)
|
|
results = response.json()["results"]
|
|
self.assertEqual(["scopone_scientifico"], [g["id"] for g in results])
|
|
self.assertEqual("Scopone scientifico", results[0]["name"])
|
|
self.assertTrue(results[0]["description"])
|
|
self.assertEqual(4, results[0]["min_players"])
|
|
self.assertEqual(4, results[0]["max_players"])
|
|
self.assertIn("target_score", results[0]["options_schema"]["properties"])
|
|
self.assertIn("napola", results[0]["options_schema"]["properties"])
|
|
self.assertIn(
|
|
"asso_pigliatutto", results[0]["options_schema"]["properties"]
|
|
)
|
|
|
|
@async_test
|
|
async def test_create_defaults_game_type(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={})
|
|
self.assertEqual(201, created.status_code)
|
|
self.assertEqual("scopone_scientifico", created.json()["game_type"])
|
|
|
|
@async_test
|
|
async def test_create_with_explicit_game_type(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={"game_type": "scopone_scientifico"}
|
|
)
|
|
self.assertEqual(201, created.status_code)
|
|
body = created.json()
|
|
self.assertEqual("scopone_scientifico", body["game_type"])
|
|
|
|
with oidc_user("alice"):
|
|
snapshot = await client.get(f"/api/games/{body['id']}")
|
|
self.assertEqual("scopone_scientifico", snapshot.json()["game_type"])
|
|
|
|
@async_test
|
|
async def test_create_rejects_unknown_game_type(self) -> None:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
|
|
with oidc_user("alice"):
|
|
unknown = await client.post("/api/games", json={"game_type": "briscola"})
|
|
non_string = await client.post("/api/games", json={"game_type": 42})
|
|
self.assertEqual(400, unknown.status_code)
|
|
self.assertEqual(400, non_string.status_code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|