tavolo/server/tests/test_scopa.py

"""Scopa integration tests: lobby, websocket play, timeouts and summaries.

The per-package suites cover the rules and the engine contract; these
tests wire both scopa 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.scopa.state import (
    PHASE_HAND_END,
    PHASE_PLAYING,
    Card,
    PlayerState,
    ScopaState,
)
from tavolo.platform import GameSession, Seat
from tests.helpers import async_test, make_user, oidc_user, ws_users


class ScopaRoutesTest(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": "scopa",
                        "options": {"target_score": 21, "asso_pigliatutto": True},
                    },
                )
            self.assertEqual(201, created.status_code)
            body = created.json()
            self.assertEqual("scopa", body["game_type"])
            self.assertEqual("lobby", body["phase"])
            self.assertEqual(1, body["seats_open"])
            self.assertEqual(21, body["target_score"])
            self.assertTrue(body["asso_pigliatutto"])
            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(4, len(state["table"]))
            self.assertEqual(30, state["deck_count"])
            self.assertEqual(1, state["turn"])
            self.assertTrue(state["asso_pigliatutto"])
            # 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": "scopa_a_quattro"}
                )
            self.assertEqual(201, created.status_code)
            body = created.json()
            self.assertEqual("scopa_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"]])
            # Partnership scopa defaults to the traditional 21 points.
            self.assertEqual(21, state["target_score"])
            for participant in state["players"]:
                self.assertEqual(3, participant["cards_left"])
            self.assertEqual(4, len(state["table"]))
            self.assertEqual(24, state["deck_count"])

    @async_test
    async def test_preview_shows_scopa_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": "scopa",
                        "options": {"target_score": 21, "asso_pigliatutto": True},
                    },
                )
            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("scopa", body["game_type"])
            self.assertEqual("Scopa", body["game_name"])
            self.assertTrue(body["game_description"])
            self.assertEqual(21, body["target_score"])
            self.assertTrue(body["asso_pigliatutto"])
            self.assertEqual(1, body["seats_open"])

    @async_test
    async def test_create_rejects_bad_options(self) -> None:
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
            for bad in (0, 101, "eleven", True):
                with oidc_user("alice"):
                    response = await client.post(
                        "/api/games",
                        json={"game_type": "scopa", "options": {"target_score": bad}},
                    )
                self.assertEqual(400, response.status_code, bad)
            with oidc_user("alice"):
                response = await client.post(
                    "/api/games",
                    json={"game_type": "scopa", "options": {"asso_pigliatutto": "yes"}},
                )
            self.assertEqual(400, response.status_code)


class ScopaWebSocketTest(unittest.TestCase):
    async def _started_two_player(self, client: AsyncClient) -> dict:
        with oidc_user("alice"):
            created = await client.post("/api/games", json={"game_type": "scopa"})
        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_play_trails_or_captures(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("bob")]):
                    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"])
                        game = first["game"]
                        self.assertEqual(1, game["turn"])
                        self.assertTrue(game.get("your_turn"))
                        self.assertEqual(4, len(game["table"]))
                        card = game["players"][1]["hand"][0]
                        options = (game.get("legal_moves") or {}).get(card)
                        payload = {"action": "play", "card": card}
                        if options:
                            payload["capture"] = options[0]
                        await bob_ws.send_json(payload)

                        update = await bob_ws.receive_json()
                        self.assertEqual("state", update["type"])
                        game = update["game"]
                        self.assertEqual(0, game["turn"])
                        self.assertIsNotNone(game["last_move"])
                        self.assertEqual(1, game["last_move"]["seat"])
                        self.assertEqual(card, game["last_move"]["card"])
                        if options:
                            self.assertEqual(2, game["players"][1]["captured_count"])
                        else:
                            self.assertEqual(5, len(game["table"]))

    @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"])

    @async_test
    async def test_asso_pigliatutto_sweeps_the_table(self) -> None:
        # A controlled scopa d'assi position: Alice's ace sweeps the whole
        # table, scores no scopa and counts as the last capture.
        state = ScopaState(
            player_count=2,
            target_score=100,
            asso_pigliatutto=True,
            phase=PHASE_PLAYING,
            dealer=0,
            turn=0,
            players=[
                PlayerState(
                    sub="alice", name="Alice", seat=0,
                    hand=[Card.parse("01D"), Card.parse("09C")],
                ),
                PlayerState(
                    sub="bob", name="Bob", seat=1,
                    hand=[Card.parse("04D")],
                ),
            ],
            table=[Card.parse("05C"), Card.parse("02D")],
        )
        engine = platform.registry.require("scopa")
        session = GameSession(
            id="scopa-asso-pigliatutto-1",
            game_type="scopa",
            join_code="SA0001",
            creator_sub="alice",
            players=[
                Seat(user_sub="alice", display_name="Alice", team="A"),
                Seat(user_sub="bob", display_name="Bob", team="B"),
            ],
        )
        session.state = state
        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()
                    self.assertEqual(0, first["game"]["turn"])
                    self.assertTrue(first["game"].get("your_turn"))
                    await ws.send_json(
                        {"action": "play", "card": "01D", "capture": ["05C", "02D"]}
                    )
                    update = await ws.receive_json()
                    game = update["game"]
                    self.assertEqual([], game["table"])
                    self.assertEqual(3, game["players"][0]["captured_count"])
                    self.assertEqual(0, game["players"][0]["scope"])
                    self.assertEqual(0, game["scores"]["A"])
                    self.assertTrue(game["last_move"]["pigliatutto"])
                    self.assertFalse(game["last_move"]["scopa"])

    @async_test
    async def test_hand_end_ack_flow(self) -> None:
        # A one-play hand with target 100 so the match continues to the
        # scoring summary.
        state = ScopaState(
            player_count=2,
            target_score=100,
            phase=PHASE_PLAYING,
            dealer=0,
            turn=0,
            players=[
                PlayerState(
                    sub="alice", name="Alice", seat=0,
                    hand=[Card.parse("02D")],
                ),
                PlayerState(sub="bob", name="Bob", seat=1),
            ],
            table=[Card.parse("02C")],
        )
        session = GameSession(
            id="scopa-hand-end-1",
            game_type="scopa",
            join_code="SH0001",
            creator_sub="alice",
            players=[
                Seat(user_sub="alice", display_name="Alice", team="A"),
                Seat(user_sub="bob", display_name="Bob", team="B"),
            ],
        )
        session.state = state
        await game_store.save(session)

        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/{session.id}", ws_client)
                        )
                        for _ in range(2)
                    ]
                    for ws in sockets:
                        await ws.receive_json()

                    # Alice captures the last table card: the hand ends.
                    await sockets[0].send_json(
                        {"action": "play", "card": "02D", "capture": ["02C"]}
                    )
                    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"])
                        hand = summary["game"]["last_hand"]
                        self.assertEqual({"A": 2, "B": 0}, hand["cards"])
                        self.assertEqual("A", hand["award"]["carte"])

                    # Both acknowledge; the second ack deals the next hand.
                    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"])


class ScopaTurnTimeoutTest(unittest.TestCase):
    @async_test
    async def test_turn_timeout_auto_plays_a_card(self) -> None:
        engine = platform.registry.require("scopa")
        session = GameSession(
            id="scopa-turn-timeout-1",
            game_type=engine.id,
            join_code="ST0001",
            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_move"] is not None
                        ):
                            break
                    self.assertIsNotNone(update)
                    assert update is not None
                    # The absent player was auto-played: the move is on the
                    # board and the turn advanced.
                    self.assertEqual(1, update["game"]["last_move"]["seat"])
                    self.assertEqual(0, update["game"]["turn"])
                    players = {p["seat"]: p for p in update["game"]["players"]}
                    self.assertEqual(2, players[1]["cards_left"])
                    self.assertEqual(3, players[0]["cards_left"])


if __name__ == "__main__":
    unittest.main()