woggioni/tavolo
2
0
Fork 0
Code Actions Packages Releases Activity
Files
43d5a6e2f9d31f5a421c6653a80a4ff635e32d5d
tavolo/server/tests/test_briscola_chiamata.py
T
woggioni 43d5a6e2f9 Add briscola chiamata 
Five-player briscola with a descending calling auction and a secret
partner. Starting with the player after the dealer, each player bids a
lower rank (A > 3 > K > H > J > 7 > 6 > 5 > 4 > 2) or passes; once the
standing bid is the two the auction continues on points (62, 64, ...).
The last player standing must bid, and the winner names any card of the
won rank: its holder is the secret partner, or the caller plays alone
when it is one of their own cards, and its suit becomes the trump. The
calling side must reach 61 (or the raised target); a 60-60 deal with the
plain target is a patta and is replayed. A match is a single deal, so
the caller and partner (or the lone caller) against the other players
map onto the platform's two-team result, Elo and match history.

The rules live in tavolo.briscola.chiamata with the
BriscolaChiamataEngine adapter registered alongside the other briscola
variants; the partner's role and the side totals stay hidden until the
called card is played. The web client gets a five-seat table, the
auction panel with legal bids and point raises, the call picker, the
reveal bands and a chiamata-aware final summary.
2026-10-08 04:27:15 +00:00

344 lines
16 KiB
Python
Raw Blame History

"""Briscola chiamata integration tests: lobby, auction, websocket play, timeouts.
The per-package suite covers the rules and the engine contract; these
tests wire the game through the full platform stack (REST, WebSocket,
deadline scheduler).
"""
from __future__ import annotations
import asyncio
import unittest
from datetime import datetime, timedelta, timezone
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.chiamata import (
PHASE_AUCTION,
PHASE_CALLING,
PHASE_PLAYING,
Auction,
Bid,
)
from tavolo.platform import GameSession, Seat
from tests.helpers import async_test, make_user, oidc_user, ws_users
PLAYERS = ("alice", "bob", "carol", "dave", "erin")
async def _create_and_join(client: AsyncClient) -> dict:
"""Create a chiamata game and seat all five players; return the view."""
with oidc_user(PLAYERS[0]):
created = await client.post("/api/games", json={"game_type": "briscola_chiamata"})
assert created.status_code == 201, created.text
code = created.json()["join_code"]
body = created.json()
for player in PLAYERS[1:]:
with oidc_user(player):
joined = await client.post("/api/games/join", json={"code": code})
assert joined.status_code == 200, joined.text
body = joined.json()
return body
class ChiamataRoutesTest(unittest.TestCase):
@async_test
async def test_create_and_seat_five_players(self) -> None:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
with oidc_user(PLAYERS[0]):
created = await client.post(
"/api/games", json={"game_type": "briscola_chiamata"}
)
self.assertEqual(201, created.status_code)
body = created.json()
self.assertEqual("briscola_chiamata", body["game_type"])
self.assertEqual("lobby", body["phase"])
self.assertEqual(4, body["seats_open"])
self.assertIsNone(body["players"][0]["team"])
code = body["join_code"]
for player in PLAYERS[1:4]:
with oidc_user(player):
joined = await client.post("/api/games/join", json={"code": code})
self.assertEqual("lobby", joined.json()["phase"])
self.assertEqual(
4 - (PLAYERS.index(player)), joined.json()["seats_open"]
)
with oidc_user(PLAYERS[4]):
started = await client.post("/api/games/join", json={"code": code})
state = started.json()
self.assertEqual(PHASE_AUCTION, state["phase"])
self.assertEqual(0, state["seats_open"])
self.assertEqual(5, len(state["players"]))
for seat in state["players"]:
# Roles are hidden until the called card is played.
self.assertEqual("", seat["team"])
self.assertEqual(8, seat["cards_left"])
# The auction opens with the player after the dealer (seat 0).
self.assertEqual(1, state["auction"]["turn"])
self.assertEqual([0, 1, 2, 3], state["auction"]["active"][0:4])
self.assertIsNone(state["auction"]["current"])
self.assertIsNotNone(state["auction_deadline"])
# The last joiner's personalized view carries their own hand.
self.assertIn("hand", state["players"][4])
self.assertEqual(8, len(state["players"][4]["hand"]))
self.assertNotIn("hand", state["players"][0])
# There is no stock in this variant.
self.assertNotIn("stock_count", state)
# A non-player watches without hands or the join code.
with oidc_user("mallory"):
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 seat in spectator["players"]:
self.assertNotIn("hand", seat)
@async_test
async def test_preview_lists_chiamata(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_chiamata"}
)
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_chiamata", body["game_type"])
self.assertEqual("Briscola chiamata", body["game_name"])
self.assertTrue(body["game_description"])
self.assertEqual("lobby", body["phase"])
self.assertEqual(4, body["seats_open"])
class ChiamataWebSocketTest(unittest.TestCase):
@async_test
async def test_auction_call_and_play_over_the_socket(self) -> None:
api_transport = ASGITransport(app=app)
async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
state = await _create_and_join(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("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(PHASE_AUCTION, first["game"]["phase"])
self.assertTrue(first["game"].get("your_turn"))
self.assertIn("legal_bids", first["game"])
async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as alice_ws:
await alice_ws.receive_json()
# Bob (seat 1) opens the bidding with the two.
await bob_ws.send_json({"action": "bid", "rank": 2})
update = await bob_ws.receive_json()
self.assertEqual("state", update["type"])
current = update["game"]["auction"]["current"]
self.assertEqual(
{"seat": 1, "rank": 2, "points": 61}, current
)
self.assertEqual(2, update["game"]["auction"]["turn"])
# Alice receives the broadcast of Bob's bid.
broadcast = await alice_ws.receive_json()
self.assertEqual("state", broadcast["type"])
# Alice (seat 0) is not on turn: her bid is an error.
await alice_ws.send_json({"action": "bid", "rank": 3})
error = await alice_ws.receive_json()
self.assertEqual("error", error["type"])
# The remaining players pass; Bob wins the auction and is
# asked to name the card.
engine = platform.registry.require("briscola_chiamata")
session = await game_store.load(game_id)
assert session is not None
for sub in ("carol", "dave", "erin", "alice"):
engine.handle_action(session, sub, "pass", {})
await game_store.save(session)
self.assertEqual(PHASE_CALLING, session.state.phase)
self.assertEqual(1, session.state.caller)
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 ws:
first = await ws.receive_json()
self.assertEqual(PHASE_CALLING, first["game"]["phase"])
self.assertTrue(first["game"].get("your_turn"))
self.assertEqual(
["02D", "02C", "02S", "02B"],
first["game"]["callable_cards"],
)
await ws.send_json(
{"action": "call", "card": "02D"}
)
update = await ws.receive_json()
self.assertEqual(PHASE_PLAYING, update["game"]["phase"])
self.assertEqual("D", update["game"]["briscola_suit"])
self.assertEqual("02D", update["game"]["called_card"])
self.assertEqual(1, update["game"]["turn"])
# Bob is first to play (player after the dealer).
bob = next(
p for p in update["game"]["players"] if p["seat"] == 1
)
card = bob["hand"][0]
await ws.send_json({"action": "play", "card": card})
update = await ws.receive_json()
self.assertEqual(
[{"seat": 1, "card": card}], update["game"]["trick"]
)
@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 _create_and_join(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 seat in first["game"]["players"]:
self.assertNotIn("hand", seat)
await ws.send_json({"action": "bid", "rank": 1})
error = await ws.receive_json()
self.assertEqual("forbidden", error["code"])
class ChiamataTimeoutTest(unittest.TestCase):
async def _seed(self, game_id: str, join_code: str, state_mutator) -> str:
from tavolo.briscola import chiamata as rules
session = GameSession(
id=game_id,
game_type="briscola_chiamata",
join_code=join_code,
creator_sub=PLAYERS[0],
players=[
Seat(user_sub=name, display_name=name.capitalize())
for name in PLAYERS
],
)
session.state = rules.create_game(PLAYERS[0], "Alice", turn_timeout=1)
for name in PLAYERS[1:]:
rules.join_game(session.state, name, name.capitalize())
state_mutator(session.state)
await game_store.save(session)
return session.id
@async_test
async def test_auction_timeout_auto_bids(self) -> None:
game_id = await self._seed(
"chiamata-auction-timeout-1", "CH0001", lambda state: None
)
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 ws:
first = await ws.receive_json()
self.assertEqual(PHASE_AUCTION, first["game"]["phase"])
self.assertIsNotNone(first["game"]["auction_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"]["auction"]["current"] is not None
):
break
self.assertIsNotNone(update)
assert update is not None
# The absent seat 1 was auto-bid for.
current = update["game"]["auction"]["current"]
self.assertEqual(1, current["seat"])
self.assertIn(current["rank"], (1, 3, 10, 9, 8, 7, 6, 5, 4, 2))
@async_test
async def test_call_and_turn_timeouts(self) -> None:
def seed(state) -> None:
state.phase = PHASE_CALLING
state.auction = Auction(
turn=1,
active=[1],
passed=[0, 2, 3, 4],
current=Bid(seat=1, rank=2, points=61),
)
state.caller = 1
state.turn = 1
state.auction_deadline = None
state.call_deadline = (
datetime.now(timezone.utc) + timedelta(seconds=1)
).isoformat()
game_id = await self._seed(
"chiamata-call-timeout-1", "CH0002", seed
)
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 ws:
first = await ws.receive_json()
self.assertEqual(PHASE_CALLING, first["game"]["phase"])
called = None
for _ in range(20):
try:
called = await asyncio.wait_for(
ws.receive_json(), timeout=2
)
except asyncio.TimeoutError:
break
if (
called.get("type") == "state"
and called["game"]["phase"] == PHASE_PLAYING
):
break
self.assertIsNotNone(called)
assert called is not None
self.assertEqual(PHASE_PLAYING, called["game"]["phase"])
card = called["game"]["called_card"]
self.assertIsNotNone(card)
self.assertTrue(card.startswith("02"))
played = None
for _ in range(20):
try:
played = await asyncio.wait_for(
ws.receive_json(), timeout=2
)
except asyncio.TimeoutError:
break
if (
played.get("type") == "state"
and len(played["game"]["trick"]) == 1
):
break
self.assertIsNotNone(played)
assert played is not None
# Seat 1 was on turn and was auto-played for.
self.assertEqual(1, played["game"]["trick"][0]["seat"])
if __name__ == "__main__":
unittest.main()
View Git Blame Copy Permalink