woggioni
ae7ab5ce60
CI / Build and push docker image (push) Successful in 2m6s
Partnership games go to 21: scopone scientifico (the tournament target) and scopa a quattro, while two-player scopa keeps its usual 11. The scopa package resolves the default from the player count and exposes a per variant options_schema default, so the lobby creation form shows the right value without frontend changes.
415 lines
19 KiB
Python
415 lines
19 KiB
Python
"""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()
|