Add hand-end scoring summary screen with acknowledgement

After each hand of an unfinished match the game now pauses in a new
hand_end phase instead of dealing immediately:

- engine: hand_points gains an 'award' map (which team won each category),
  _end_hand stops at hand_end with a deadline, new acknowledge_hand deals
  the next hand once all four players have acked; plays are rejected while
  the summary is up
- state: acked seats, hand_end_deadline and hand_ack_timeout are persisted
  and exposed in the personalized view (also on the finished state, so the
  final hand is explained before the result)
- ws: new {"action": "ack"}; a per-hand timer force-deals the next hand
  after HAND_ACK_TIMEOUT_SECONDS (new env var, default 30s) so an away
  player cannot stall the match
- web: modal explaining each category in plain language with icons (card
  images for denara/settebello/primiera), team-coloured rows, running
  totals with progress bars, an 'Understood — next hand' button that turns
  into 'Waiting for …' plus an auto-continue countdown; the final screen
  shows the last hand's breakdown too

Verified in the browser against the compose stack: hand played to
completion, summary rendered (including a carte tie), ack from all four
players dealt the next hand live, and the auto-continue path fired when
nobody acked. 60 backend tests + mypy + cargo tests green.
This commit is contained in:
2026-09-16 21:46:01 +08:00
parent b6a3a95f52
commit d9cdba33a1
19 changed files with 873 additions and 13 deletions
+112 -1
View File
@@ -1,6 +1,7 @@
"""WebSocket live-play tests via kaya's ASGI websocket transport."""
from __future__ import annotations
import asyncio
import unittest
from httpx import ASGITransport, AsyncClient
@@ -8,7 +9,9 @@ from httpx_ws import WebSocketDisconnect, aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport
from pwo import async_test
from scopa.app import app
from scopa.app import app, game_store
from scopa.game import engine
from scopa.game.state import Card, GameState, PlayerState
from tests.helpers import make_user, oidc_user, ws_users
PLAYERS = ("alice", "bob", "carol", "dave")
@@ -127,5 +130,113 @@ class WebSocketTest(unittest.TestCase):
self.assertEqual(4401, caught.exception.code)
async def _seed_last_play_state(hand_ack_timeout: int = 30) -> str:
"""Seed a game where a single play ends the hand: p0 holds the only
card left and can capture the only table card."""
state = GameState(
id="hand-end-1",
join_code="HEND01",
creator_sub="alice",
target_score=11,
phase="playing",
turn=0,
table=[Card.parse("02C")],
)
state.players = [
PlayerState(sub="alice", name="Alice", seat=0, hand=[Card.parse("02D")]),
PlayerState(sub="bob", name="Bob", seat=1),
PlayerState(sub="carol", name="Carol", seat=2),
PlayerState(sub="dave", name="Dave", seat=3),
]
state.hand_ack_timeout = hand_ack_timeout
await game_store.save(state)
return state.id
class HandEndWebSocketTest(unittest.TestCase):
@async_test
async def test_hand_end_ack_flow(self) -> None:
import contextlib
game_id = await _seed_last_play_state()
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(name) for name in PLAYERS]):
sockets = [
await stack.enter_async_context(
aconnect_ws(f"/ws/games/{game_id}", ws_client)
)
for _ in PLAYERS
]
for ws in sockets:
await ws.receive_json() # initial state
# Alice plays the last card: the hand ends and the game
# pauses for acknowledgements.
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("hand_end", summary["game"]["phase"])
self.assertEqual([], summary["game"]["acknowledged"])
self.assertIsNotNone(summary["game"]["hand_end_deadline"])
award = summary["game"]["last_hand"]["award"]
self.assertEqual("A", award["carte"])
self.assertEqual("A", award["denara"])
# Everyone acknowledges; the fourth ack deals the next hand.
for i, ws in enumerate(sockets):
await ws.send_json({"action": "ack"})
updates = [await other.receive_json() for other in sockets]
for update in updates:
if i < 3:
self.assertEqual("hand_end", update["game"]["phase"])
self.assertEqual(
list(range(i + 1)),
update["game"]["acknowledged"],
)
else:
self.assertEqual("playing", update["game"]["phase"])
self.assertEqual(2, update["game"]["hand_number"])
self.assertEqual(
10, update["game"]["players"][i]["cards_left"]
)
@async_test
async def test_hand_end_timeout_deals_next_hand(self) -> None:
game_id = await _seed_last_play_state(hand_ack_timeout=1)
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:
await ws.receive_json() # initial state
await ws.send_json(
{"action": "play", "card": "02D", "capture": ["02C"]}
)
summary = await ws.receive_json()
self.assertEqual("hand_end", summary["game"]["phase"])
# Nobody acks: the timer must deal the next hand.
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"]["phase"] == "playing"
):
break
self.assertIsNotNone(update)
assert update is not None
self.assertEqual("playing", update["game"]["phase"])
self.assertEqual(2, update["game"]["hand_number"])
if __name__ == "__main__":
unittest.main()