tavolo/server/tests/test_briscola_chiamata.py

"""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_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()