woggioni/tavolo
2
0
Fork 0
Code Actions Packages Releases Activity
Files
68200bbe6217d754d061ece45e4597175835ed9d
tavolo/server/tests/test_briscola.py
T
woggioni 7bb6b74fdc
CI / Build and push docker image (push) Successful in 1m45s
Details
Add briscola: classic two-player and briscola a quattro 
New tavolo-briscola distribution implementing briscola on the platform:

- pure rules (state.py, engine.py): 3-card deals, a face-up trump that
  is the last card drawn, trump-first trick resolution (A > 3 > K > H >
  J > 7 > 6 > 5 > 4 > 2), 120 points per deal, 60-60 patta re-deals, a
  configurable target score (default 61), deal-end acknowledgements and
  turn auto-play;
- BriscolaEngine (briscola, 2 players) and BriscolaQuattroEngine
  (briscola_a_quattro, 4 players in fixed partnerships) adapters;
- registered in tavolo-app, packaged in the Docker image, documented.

Platform fixes surfaced by a two-player lobby:

- the websocket state and snapshot payloads now carry seats_open (they
  were only in the REST lobby payload), so lobbies with player counts
  other than four render correctly; seats_open is clamped at zero;
- the web client logs unparseable server messages instead of silently
  dropping them.

Web client: briscola table for both variants (stock, face-up trump,
trick, last-trick banner), deal summary modal and game-over overlay.

Tests: tavolo-briscola rules/adapter suites, app integration tests
(lobby, websocket play, turn timeout, deal-end acks, persistence), a
platform seats_open test and frontend model parsing tests.
2026-10-06 14:08:25 +08:00

369 lines
17 KiB
Python
Raw Blame History

"""Briscola integration tests: lobby, websocket play, timeouts and summaries.
The per-package suites cover the rules and the engine contract; these
tests wire both briscola variants through the full platform stack (REST,
WebSocket, deadline scheduler).
"""
from __future__ import annotations
import asyncio
import contextlib
import unittest
from httpx import ASGITransport, AsyncClient
from httpx_ws import aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport
from tavolo.app import app, game_store, platform
from tavolo.briscola.state import (
PHASE_HAND_END,
PHASE_PLAYING,
BriscolaState,
Card,
PlayerState,
)
from tavolo.platform import GameSession, Seat
from tests.helpers import async_test, make_user, oidc_user, ws_users
class BriscolaRoutesTest(unittest.TestCase):
@async_test
async def test_create_and_start_two_player_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={"game_type": "briscola", "options": {"target_score": 101}},
)
self.assertEqual(201, created.status_code)
body = created.json()
self.assertEqual("briscola", body["game_type"])
self.assertEqual("lobby", body["phase"])
self.assertEqual(1, body["seats_open"])
self.assertEqual(101, body["target_score"])
self.assertEqual("A", body["players"][0]["team"])
code = body["join_code"]
with oidc_user("bob"):
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(2, len(state["players"]))
self.assertEqual(["A", "B"], [p["team"] for p in state["players"]])
for participant in state["players"]:
self.assertEqual(3, participant["cards_left"])
self.assertEqual(1, state["turn"])
self.assertEqual(33, state["stock_count"])
self.assertTrue(state["briscola_suit"])
self.assertIsNotNone(state["briscola_card"])
# Bob's view carries his own hand only.
self.assertIn("hand", state["players"][1])
self.assertNotIn("hand", state["players"][0])
# A non-player watches without a hand or the join code.
watched = await client.get(f"/api/games/{state['id']}")
self.assertEqual(200, watched.status_code)
spectator = watched.json()
self.assertTrue(spectator["spectator"])
self.assertNotIn("join_code", spectator)
for participant in spectator["players"]:
self.assertNotIn("hand", participant)
@async_test
async def test_create_and_start_four_player_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={"game_type": "briscola_a_quattro"}
)
self.assertEqual(201, created.status_code)
body = created.json()
self.assertEqual("briscola_a_quattro", body["game_type"])
self.assertEqual(3, body["seats_open"])
code = body["join_code"]
for player in ("bob", "carol"):
with oidc_user(player):
joined = await client.post("/api/games/join", json={"code": code})
self.assertEqual("lobby", joined.json()["phase"])
with oidc_user("dave"):
started = await client.post("/api/games/join", json={"code": code})
state = started.json()
self.assertEqual("playing", state["phase"])
self.assertEqual(["A", "B", "A", "B"], [p["team"] for p in state["players"]])
for participant in state["players"]:
self.assertEqual(3, participant["cards_left"])
self.assertEqual(40 - 12 - 1, state["stock_count"])
@async_test
async def test_preview_shows_briscola_options(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": "briscola", "options": {"target_score": 101}},
)
code = created.json()["join_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("briscola", body["game_type"])
self.assertEqual("Briscola", body["game_name"])
self.assertTrue(body["game_description"])
self.assertEqual(101, body["target_score"])
self.assertEqual(1, body["seats_open"])
@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:
for bad in (60, 121, "sixtyone", True):
with oidc_user("alice"):
response = await client.post(
"/api/games",
json={"game_type": "briscola", "options": {"target_score": bad}},
)
self.assertEqual(400, response.status_code, bad)
class BriscolaWebSocketTest(unittest.TestCase):
async def _started_two_player(self, client: AsyncClient) -> dict:
with oidc_user("alice"):
created = await client.post("/api/games", json={"game_type": "briscola"})
code = created.json()["join_code"]
with oidc_user("bob"):
started = await client.post("/api/games/join", json={"code": code})
return started.json()
@async_test
async def test_moves_resolve_the_trick(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_two_player(client)
game_id = state["id"]
bob_card = state["players"][1]["hand"][0]
with oidc_user("alice"):
alice_view = await client.get(f"/api/games/{game_id}")
alice_card = alice_view.json()["players"][0]["hand"][0]
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:
await alice_ws.receive_json()
await bob_ws.send_json(
{"action": "play", "card": bob_card}
)
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(0, update["game"]["turn"])
self.assertEqual(
[{"seat": 1, "card": bob_card}],
update["game"]["trick"],
)
await alice_ws.send_json(
{"action": "play", "card": alice_card}
)
bob_update = await bob_ws.receive_json()
alice_update = await alice_ws.receive_json()
for update in (bob_update, alice_update):
self.assertEqual([], update["game"]["trick"])
last = update["game"]["last_trick"]
self.assertIsNotNone(last)
self.assertIn(last["winner"], (0, 1))
self.assertEqual(2, len(last["cards"]))
self.assertEqual(last["winner"], update["game"]["turn"])
@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_two_player(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("alice")]):
async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as alice_ws:
await alice_ws.receive_json()
# Bob (seat 1) is on turn, not Alice.
await alice_ws.send_json({"action": "play", "card": "01D"})
error = await alice_ws.receive_json()
self.assertEqual("error", error["type"])
@async_test
async def test_spectator_cannot_act(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_two_player(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")]):
async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
first = await ws.receive_json()
self.assertTrue(first["game"]["spectator"])
self.assertNotIn("join_code", first["game"])
for player in first["game"]["players"]:
self.assertNotIn("hand", player)
await ws.send_json({"action": "play", "card": "01D"})
error = await ws.receive_json()
self.assertEqual("forbidden", error["code"])
class BriscolaTurnTimeoutTest(unittest.TestCase):
@async_test
async def test_turn_timeout_auto_plays_a_card(self) -> None:
engine = platform.registry.require("briscola")
session = GameSession(
id="briscola-turn-timeout-1",
game_type=engine.id,
join_code="BT0001",
creator_sub="alice",
players=[
Seat(user_sub="alice", display_name="Alice", team="A"),
],
)
engine.create(session, {})
session.state.turn_timeout = 1
engine.join(session, "bob", "Bob")
await game_store.save(session)
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/{session.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"])
self.assertIsNotNone(first["game"]["turn_deadline"])
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"]["last_trick"] is not None
):
break
self.assertIsNotNone(update)
assert update is not None
# Both absent players were auto-played; the trick
# resolved and the winner leads with a full hand.
self.assertIn(update["game"]["last_trick"]["winner"], (0, 1))
self.assertEqual([], update["game"]["trick"])
for participant in update["game"]["players"]:
self.assertEqual(3, participant["cards_left"])
async def _seed_hand_end_state() -> str:
"""Seed a game where a single trick ends the deal with target 120 so
the match continues to the summary."""
engine = platform.registry.require("briscola")
session = GameSession(
id="briscola-hand-end-1",
game_type=engine.id,
join_code="BH0001",
creator_sub="alice",
players=[
Seat(user_sub="alice", display_name="Alice", team="A"),
Seat(user_sub="bob", display_name="Bob", team="B"),
],
)
session.state = BriscolaState(
player_count=2,
target_score=120,
phase=PHASE_PLAYING,
dealer=0,
turn=0,
briscola_suit="S",
players=[
PlayerState(
sub="alice", name="Alice", seat=0,
hand=[Card.parse("02D")], captured=[Card.parse("01C")],
),
PlayerState(
sub="bob", name="Bob", seat=1,
hand=[Card.parse("05D")], captured=[Card.parse("10C")],
),
],
hand_ack_timeout=30,
)
await game_store.save(session)
return session.id
class BriscolaHandEndTest(unittest.TestCase):
@async_test
async def test_deal_end_ack_flow(self) -> None:
game_id = await _seed_hand_end_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("alice"), make_user("bob")]):
sockets = [
await stack.enter_async_context(
aconnect_ws(f"/ws/games/{game_id}", ws_client)
)
for _ in range(2)
]
for ws in sockets:
await ws.receive_json() # initial state
# Alice leads, Bob takes the trick: the deal ends.
await sockets[0].send_json(
{"action": "play", "card": "02D"}
)
for ws in sockets:
await ws.receive_json()
await sockets[1].send_json(
{"action": "play", "card": "05D"}
)
summaries = [await ws.receive_json() for ws in sockets]
for summary in summaries:
self.assertEqual("state", summary["type"])
self.assertEqual(PHASE_HAND_END, summary["game"]["phase"])
self.assertEqual([], summary["game"]["acknowledged"])
self.assertIsNotNone(summary["game"]["hand_end_deadline"])
deal = summary["game"]["last_deal"]
self.assertEqual({"A": 11, "B": 4}, deal["points"])
self.assertEqual({"A": 11, "B": 4}, deal["totals"])
self.assertFalse(deal["patta"])
# Both acknowledge; the second ack deals the next deal.
await sockets[0].send_json({"action": "ack"})
updates = [await ws.receive_json() for ws in sockets]
for update in updates:
self.assertEqual(PHASE_HAND_END, update["game"]["phase"])
self.assertEqual([0], update["game"]["acknowledged"])
await sockets[1].send_json({"action": "ack"})
updates = [await ws.receive_json() for ws in sockets]
for update in updates:
self.assertEqual(PHASE_PLAYING, update["game"]["phase"])
self.assertEqual(2, update["game"]["hand_number"])
for player in update["game"]["players"]:
self.assertEqual(3, player["cards_left"])
if __name__ == "__main__":
unittest.main()
View Git Blame Copy Permalink