"""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_principal, 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, make_principal(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=make_principal(PLAYERS[0]), players=[ Seat(user_sub=make_principal(name), display_name=name.capitalize()) for name in PLAYERS ], ) session.state = rules.create_game( make_principal(PLAYERS[0]), "Alice", turn_timeout=1 ) for name in PLAYERS[1:]: rules.join_game(session.state, make_principal(name), name.capitalize()) state_mutator(session.state) await game_store.save(session) return session.id @async_test async def test_auction_timeout_auto_passes(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 1 in update["game"]["auction"]["passed"] ): break self.assertIsNotNone(update) assert update is not None # The absent seat 1 passed: nobody bid and the turn # moved to seat 2. auction = update["game"]["auction"] self.assertIn(1, auction["passed"]) self.assertIsNone(auction["current"]) self.assertEqual(2, auction["turn"]) @async_test async def test_forced_last_bidder_auto_bids_the_ace(self) -> None: def seed(state) -> None: state.phase = PHASE_AUCTION state.auction = Auction( turn=0, active=[0], passed=[1, 2, 3, 4], current=None ) state.turn = 0 state.auction_deadline = ( datetime.now(timezone.utc) + timedelta(seconds=1) ).isoformat() state.call_deadline = None game_id = await self._seed("chiamata-forced-ace-1", "CH0003", seed) 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: first = await ws.receive_json() self.assertEqual(PHASE_AUCTION, first["game"]["phase"]) # The auction times out: seat 0 is forced to bid the # ace; its call then times out too and a random ace # suit is named. 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 played["game"]["phase"] == PHASE_PLAYING ): break self.assertIsNotNone(played) assert played is not None self.assertEqual(PHASE_PLAYING, played["game"]["phase"]) self.assertEqual(0, played["game"]["caller"]) card = played["game"]["called_card"] self.assertIsNotNone(card) self.assertTrue(card.startswith("01")) @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()