woggioni/tavolo
2
0
Fork 0
Code Actions Packages Releases Activity

Split backend into tavolo-platform, tavolo-scopone and tavolo-app packages

Browse Source 
Move the game-independent machinery (lobby, live-game store, websocket,
deadline scheduler, match history, leaderboards) into a new
tavolo-platform distribution behind a GameEngine contract, the scopone
scientifico rules plus a platform adapter into tavolo-scopone, and keep
only the composition root in tavolo-app. The three distributions share
the tavolo namespace (PEP 420, kaya-style monorepo).

Match history becomes fully generic: Match carries the engine's result
JSON and MatchPlayer points/details instead of scopone-shaped team
columns (migration 3 backfills existing rows). Lobby creation takes an
opaque per-game options object and websocket actions dispatch to the
session's engine.

Tests: platform suite runs against a DummyEngine toy game, scopone
keeps the rules tests plus new adapter tests, server/tests covers the
wired stack end to end (194 tests, was 143).
This commit is contained in:
woggioni
2026-09-19 07:28:58 +00:00
parent b2b514ab91
commit 5a73601ddf
72 changed files with 4490 additions and 2102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/packages/tavolo-scopone/README.md
+36
View File
@@ -0,0 +1,36 @@
# tavolo-scopone
Scopone scientifico — the four-player, fixed-partnership Italian card
game — as a [`tavolo-platform`](../tavolo-platform/README.md) game
implementation.
## Contents
- `state.py` — `ScoponeState` (pure game data: phases, players, hands,
table, scores, deadlines), `PlayerState`, `Card`, `Move`, with JSON
(de)serialization. No session envelope, no transport, no I/O.
- `engine.py` — the pure rules engine: deck, legal captures, plays,
auto-play, hand scoring (carte, denara, settebello, primiera, scope,
napola), hand-end acknowledgements. Deterministic and I/O-free apart
from logging, so the whole rule set is unit-testable.
- `errors.py` — scopone-specific errors (`CardNotInHand`); every other
failure mode is a shared `tavolo.platform.errors` subclass.
- `plugin.py` — `ScoponeEngine(GameEngine)`: the platform-facing adapter.
It translates create/join/websocket actions/deadlines into rules-engine
calls and back, validates the `target_score`/`napola` creation options,
and extracts the `MatchResult` (teams, winner, per-player scores and the
match summary persisted as the match's JSON `result`).
Timeouts are constructor arguments (`turn_timeout_seconds`,
`hand_ack_timeout_seconds`), wired from the environment by the
application composition root.
## Development (from `server/`)
```sh
.venv/bin/python -m unittest discover -s packages/tavolo-scopone/tests
.venv/bin/python -m mypy -p tavolo.scopone
```
`test_engine.py` covers the pure rules; `test_plugin.py` covers the
platform contract (actions, deadlines, serialization, results).
server/packages/tavolo-scopone/pyproject.toml
+22
View File
@@ -0,0 +1,22 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "tavolo-scopone"
version = "0.1.0"
description = "Scopone scientifico game implementation for the tavolo platform"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"tavolo-platform",
]
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
[tool.mypy]
python_version = "3.12"
ignore_missing_imports = true
plugins = []
server/packages/tavolo-scopone/src/tavolo/scopone/__init__.py
+13
View File
@@ -0,0 +1,13 @@
"""Scopone scientifico: tavolo's first game implementation.
The pure rules (:mod:`tavolo.scopone.engine`, :mod:`tavolo.scopone.state`)
know nothing about HTTP, Redis or Postgres; :mod:`tavolo.scopone.plugin`
adapts them to the platform's
:class:`~tavolo.platform.engine.GameEngine` contract so the
game-independent platform can host them.
"""
from __future__ import annotations
from .plugin import ScoponeEngine
__all__ = ["ScoponeEngine"]
server/packages/tavolo-scopone/src/tavolo/scopone/engine.py
+554
View File
@@ -0,0 +1,554 @@
"""Pure rules engine for scopone scientifico.
Every function here is deterministic and I/O-free (the only side effect is
debug logging): it mutates (or reads)
:class:`~tavolo.scopone.state.ScoponeState` and raises
:class:`~tavolo.platform.errors.GameError` subclasses on rule violations.
This makes the whole rule set unit-testable without Redis, Postgres or
HTTP. The platform-facing adapter is
:class:`~tavolo.scopone.plugin.ScoponeEngine`.
Rules implemented
-----------------
* 40-card Italian deck (4 suits x ranks 1-10), ten cards per player, empty
table at the start of every hand.
* A card captures either a **single card of equal rank** or a **combination
of cards whose ranks sum to its own**. When an equal-ranked card is on the
table that capture is mandatory; the player may not take an alternative
combination instead.
* Emptying the table with a capture is a **scopa** (+1), except on the very
last play of a hand.
* At the end of a hand the remaining table cards go to the player who made
the last capture.
* Hand points: ``carte`` (most captured cards), ``denara`` (most diamond
cards), ``settebello`` (the 7 of diamonds), ``primiera`` (best
seven/five/four/three card of each suit, all four suits required), plus
one point per ``scopa``. Ties on carte/denara/primiera award nothing.
* Optional ``napola`` rule (enabled by default): the longest run of
consecutive denari starting from the ace scores one point per card when
it reaches at least three cards (A-2-3 = 3, A-2-3-4 = 4, ...). A team
capturing the whole denari suit (ace to king) wins the match instantly.
* The match ends when a team reaches the target score with a clear lead; a
tie at or above the target is broken by playing another hand.
"""
from __future__ import annotations
import random
from datetime import datetime, timedelta, timezone
from itertools import combinations
from logging import getLogger
from typing import Dict, List, Optional, Sequence, Tuple
from tavolo.platform.errors import (
AlreadyJoined,
GameFinished,
GameNotStarted,
IllegalMove,
LobbyFull,
NotYourTurn,
)
from .errors import CardNotInHand
from .state import (
DEFAULT_TARGET_SCORE,
PHASE_FINISHED,
PHASE_HAND_END,
PHASE_LOBBY,
PHASE_PLAYING,
SUITS,
TEAM_NAMES,
Card,
Move,
PlayerState,
ScoponeState,
parse_card,
)
log = getLogger(__name__)
# Number of cards dealt to each player at the start of a hand.
HAND_SIZE = 10
PLAYERS = 4
# Default seconds the hand-end summary waits before dealing anyway. Games
# carry their own copy in ``ScoponeState.hand_ack_timeout``.
DEFAULT_HAND_ACK_TIMEOUT_SECONDS = 30
# Default seconds a player has to play before the server plays a random
# legal card for them. Games carry their own copy in
# ``ScoponeState.turn_timeout``.
DEFAULT_TURN_TIMEOUT_SECONDS = 30
# Primiera card values: sevens are best, then sixes, then aces, then the
# remaining ranks in descending order. All of 8/9/10 are worth 10.
PRIMIERA_VALUES: Dict[int, int] = {
7: 21,
6: 18,
1: 16,
5: 15,
4: 14,
3: 13,
2: 12,
8: 10,
9: 10,
10: 10,
}
_rng = random.SystemRandom()
def full_deck() -> List[Card]:
"""Return the 40 cards of the Italian deck in canonical order."""
return [Card(rank=rank, suit=suit) for suit in SUITS for rank in range(1, 11)]
def shuffled_deck(rng: Optional[random.Random] = None) -> List[Card]:
"""Return a shuffled deck. Pass ``rng`` for deterministic tests."""
deck = full_deck()
(rng or _rng).shuffle(deck)
return deck
def legal_captures(table: Sequence[Card], card: Card) -> List[List[Card]]:
"""Return every legal capture (a list of card sets) for ``card``.
If an equal-ranked card is on the table, only those single-card
captures are returned (the rule forbids taking a combination instead).
Otherwise every subset of the table whose ranks sum to ``card.rank`` is
returned.
"""
equal = [c for c in table if c.rank == card.rank]
if equal:
return [[c] for c in equal]
candidates = [c for c in table if c.rank <= card.rank]
captures: List[List[Card]] = []
# A sum-equal capture needs at least two cards (single non-equal cards
# cannot sum to the played card).
for size in range(2, len(candidates) + 1):
for combo in combinations(candidates, size):
if sum(c.rank for c in combo) == card.rank:
captures.append(list(combo))
return captures
def create_game(
creator_sub: str,
creator_name: str,
target_score: int = DEFAULT_TARGET_SCORE,
hand_ack_timeout: int = DEFAULT_HAND_ACK_TIMEOUT_SECONDS,
turn_timeout: int = DEFAULT_TURN_TIMEOUT_SECONDS,
napola: bool = True,
) -> ScoponeState:
"""Create a lobby game with the creator seated first."""
if target_score < 1 or target_score > 100:
raise IllegalMove("target_score must be between 1 and 100")
return ScoponeState(
target_score=target_score,
napola=napola,
phase=PHASE_LOBBY,
players=[PlayerState(sub=creator_sub, name=creator_name, seat=0)],
hand_ack_timeout=hand_ack_timeout,
turn_timeout=turn_timeout,
)
def join_game(state: ScoponeState, sub: str, name: str) -> None:
"""Seat ``sub`` in the next free chair, starting the match when full."""
if state.phase != PHASE_LOBBY:
raise GameNotStarted("game has already started")
if state.seated(sub):
raise AlreadyJoined("already joined this game")
if len(state.players) >= PLAYERS:
raise LobbyFull("game is full")
seat = len(state.players)
state.players.append(PlayerState(sub=sub, name=name, seat=seat))
if len(state.players) == PLAYERS:
start_game(state)
def start_game(state: ScoponeState) -> None:
"""Deal the first hand and switch the game to playing."""
if len(state.players) != PLAYERS:
raise GameNotStarted("need exactly four players to start")
state.phase = PHASE_PLAYING
_deal_hand(state)
def _set_turn_deadline(state: ScoponeState) -> None:
"""Arm the auto-play deadline for whoever is on turn."""
deadline = datetime.now(timezone.utc) + timedelta(seconds=state.turn_timeout)
state.turn_deadline = deadline.isoformat()
def _deal_hand(state: ScoponeState) -> None:
deck = shuffled_deck()
for player in state.players:
player.hand = []
player.captured = []
player.scope = 0
state.table = []
state.last_taker = None
# Dealer rotates each hand; the first card is played by the player to
# the dealer's left.
state.turn = (state.dealer + 1) % PLAYERS
_set_turn_deadline(state)
for offset in range(HAND_SIZE):
for seat in range(PLAYERS):
player = _player_at(state, (state.dealer + 1 + seat) % PLAYERS)
player.hand.append(deck.pop())
log.debug("hand %d dealt (dealer seat %d)", state.hand_number, state.dealer)
def _player_at(state: ScoponeState, seat: int) -> PlayerState:
for player in state.players:
if player.seat == seat:
return player
raise IllegalMove(f"no player in seat {seat}")
def play(
state: ScoponeState,
sub: str,
card_code: str,
capture_codes: Optional[Sequence[str]] = None,
) -> None:
"""Apply one move by the player identified by ``sub``.
``capture_codes`` selects which table cards to capture; it must be a
legal capture (see :func:`legal_captures`) when one exists and empty
otherwise. Raises a :class:`~tavolo.platform.errors.GameError` subclass
on any violation.
"""
if state.phase == PHASE_FINISHED:
raise GameFinished("the match is over")
if state.phase == PHASE_HAND_END:
raise IllegalMove("the hand is over; acknowledge the summary to continue")
if state.phase != PHASE_PLAYING:
raise GameNotStarted("the game has not started yet")
player = state.player_for(sub)
if player is None or player.seat != state.turn:
raise NotYourTurn("it is not your turn")
card = parse_card(card_code)
if card not in player.hand:
raise CardNotInHand(f"card {card.code} is not in your hand")
# Remove the card now so the scopa check below can tell whether this
# was the last play of the hand.
player.hand.remove(card)
requested = [parse_card(c) for c in (capture_codes or [])]
options = legal_captures(state.table, card)
taken: List[Card] = []
scopa = False
if not options:
if requested:
raise IllegalMove("no capture is possible with that card")
state.table.append(card)
else:
chosen = _match_option(options, requested)
if chosen is None:
raise IllegalMove("the requested capture is not legal")
taken = chosen
for captured in chosen:
state.table.remove(captured)
player.captured.append(captured)
player.captured.append(card)
state.last_taker = player.seat
# A scopa scores only if cards remain to be played this hand.
hands_empty = all(not p.hand for p in state.players)
if not state.table and not hands_empty:
player.scope += 1
scopa = True
state.last_move = Move(
seat=player.seat,
name=player.name,
card=card.code,
captured=[c.code for c in taken],
scopa=scopa,
)
if all(not p.hand for p in state.players):
_end_hand(state)
else:
state.turn = (state.turn + 1) % PLAYERS
_set_turn_deadline(state)
def _match_option(
options: Sequence[Sequence[Card]], requested: Sequence[Card]
) -> Optional[List[Card]]:
"""Return the option matching ``requested`` exactly, if any."""
wanted = sorted(c.code for c in requested)
if not wanted:
return None
for option in options:
if sorted(c.code for c in option) == wanted:
return list(option)
return None
def auto_play(state: ScoponeState, rng: Optional[random.Random] = None) -> None:
"""Play a random legal move for the player currently on turn.
A card is drawn at random from that player's hand; if it can capture,
one of the legal captures is chosen at random (the rules require a
capture when one exists). Delegates to :func:`play`, so the move is
fully validated and can end the hand or the match. Pass ``rng`` for
deterministic tests.
"""
if state.phase != PHASE_PLAYING:
raise GameNotStarted("the game has not started yet")
player = _player_at(state, state.turn)
if not player.hand:
raise IllegalMove("the player on turn has no cards")
chooser = rng or _rng
card = chooser.choice(player.hand)
options = legal_captures(state.table, card)
capture = chooser.choice(options) if options else None
play(
state,
player.sub,
card.code,
[c.code for c in capture] if capture else None,
)
def _end_hand(state: ScoponeState) -> None:
"""Sweep the table and score the hand.
If the match continues, the game pauses in the ``hand_end`` phase so
every player can read the scoring summary; the next hand is dealt by
:func:`acknowledge_hand` once all four players have acknowledged (or
by the hand-end timeout fired by the platform's deadline scheduler).
If the match is over the game goes to ``finished`` immediately.
"""
state.turn_deadline = None
if state.table and state.last_taker is not None:
taker = _player_at(state, state.last_taker)
taker.captured.extend(state.table)
state.table = []
points, details = hand_points(state)
for team in (0, 1):
state.scores[team] += points[team]
details["hand"] = state.hand_number
details["team_a_points"] = points[0]
details["team_b_points"] = points[1]
state.hand_scores.append(details)
a, b = state.scores
log.debug(
"hand %d scored A+%d B+%d (totals %d-%d)",
state.hand_number,
points[0],
points[1],
a,
b,
)
# A full napola (the whole denari suit) wins the match outright,
# regardless of the score.
napola = details.get("napola")
if isinstance(napola, dict):
for team, name in enumerate(TEAM_NAMES):
if napola.get(name) == 10:
state.phase = PHASE_FINISHED
state.winner = team
log.info("team %s swept the denari (napola) and wins", name)
return
reached = max(a, b) >= state.target_score
if reached and a != b:
state.phase = PHASE_FINISHED
state.winner = 0 if a > b else 1
log.debug("match ended, team %s wins", "A" if state.winner == 0 else "B")
return
# Pause for the scoring summary instead of dealing immediately.
state.phase = PHASE_HAND_END
state.acked = []
deadline = datetime.now(timezone.utc) + timedelta(seconds=state.hand_ack_timeout)
state.hand_end_deadline = deadline.isoformat()
def acknowledge_hand(state: ScoponeState, sub: str) -> None:
"""Record that ``sub`` has read the hand-end summary.
When all four players have acknowledged, the next hand is dealt.
Acknowledging twice is a no-op; acknowledging outside the ``hand_end``
phase raises an error.
"""
if state.phase != PHASE_HAND_END:
raise IllegalMove("no hand summary is waiting for acknowledgement")
player = state.player_for(sub)
if player is None:
raise NotYourTurn("you are not seated in this game")
if player.seat in state.acked:
return
state.acked.append(player.seat)
if len(state.acked) < PLAYERS:
return
state.hand_number += 1
state.dealer = (state.dealer + 1) % PLAYERS
state.acked = []
state.hand_end_deadline = None
state.last_move = None
state.phase = PHASE_PLAYING
_deal_hand(state)
def primiera_score(captured: Sequence[Card]) -> int:
"""Return the primiera value of a capture pile (0 if a suit is absent)."""
best: Dict[str, int] = {}
for card in captured:
value = PRIMIERA_VALUES[card.rank]
if card.suit not in best or value > best[card.suit]:
best[card.suit] = value
if len(best) < len(SUITS):
return 0
return sum(best.values())
def napola_score(captured: Sequence[Card]) -> int:
"""Return the napola value of a capture pile.
The longest run of consecutive denari starting from the ace scores one
point per card once it reaches three cards (A-2-3 = 3, A-2-3-4 = 4,
...), so the whole suit (ace to king) is worth 10. Shorter runs score
nothing. Only one team can score a napola: the ace of denari belongs
to exactly one capture pile.
"""
ranks = {card.rank for card in captured if card.suit == "D"}
run = 0
while run + 1 in ranks:
run += 1
return run if run >= 3 else 0
def hand_points(state: ScoponeState) -> Tuple[List[int], Dict[str, object]]:
"""Compute the hand points for both teams (index 0 = team A)."""
piles: List[List[Card]] = [[], []]
scope: List[int] = [0, 0]
for player in state.players:
piles[player.team].extend(player.captured)
scope[player.team] += player.scope
points = [0, 0]
award: Dict[str, Optional[str]] = {}
# Carte: most captured cards. Ties award nothing.
cards = [len(piles[0]), len(piles[1])]
if cards[0] != cards[1]:
winner = 0 if cards[0] > cards[1] else 1
points[winner] += 1
award["carte"] = TEAM_NAMES[winner]
else:
award["carte"] = None
# Denara: most diamond cards. Ties award nothing.
coins = [
sum(1 for c in piles[t] if c.suit == "D") for t in (0, 1)
]
if coins[0] != coins[1]:
winner = 0 if coins[0] > coins[1] else 1
points[winner] += 1
award["denara"] = TEAM_NAMES[winner]
else:
award["denara"] = None
# Settebello: the 7 of diamonds always belongs to someone.
settebello = [
any(c.rank == 7 and c.suit == "D" for c in piles[t]) for t in (0, 1)
]
if settebello[0] != settebello[1]:
winner = 0 if settebello[0] else 1
points[winner] += 1
award["settebello"] = TEAM_NAMES[winner]
# Primiera: highest value, only if the team holds all four suits.
primiera = [primiera_score(piles[t]) for t in (0, 1)]
if primiera[0] != primiera[1]:
winner = 0 if primiera[0] > primiera[1] else 1
points[winner] += 1
award["primiera"] = TEAM_NAMES[winner]
else:
award["primiera"] = None
# Scope: one point each.
points[0] += scope[0]
points[1] += scope[1]
details: Dict[str, object] = {
"cards": {"A": cards[0], "B": cards[1]},
"denara": {"A": coins[0], "B": coins[1]},
"settebello": {"A": settebello[0], "B": settebello[1]},
"primiera": {"A": primiera[0], "B": primiera[1]},
"scope": {"A": scope[0], "B": scope[1]},
"award": award,
}
# Napola (optional rule): consecutive denari from the ace. A run of 10
# means the team swept the whole suit and wins the match instantly.
if state.napola:
napola = [napola_score(piles[t]) for t in (0, 1)]
for team in (0, 1):
points[team] += napola[team]
award["napola"] = next(
(TEAM_NAMES[t] for t in (0, 1) if napola[t] > 0), None
)
details["napola"] = {"A": napola[0], "B": napola[1]}
return points, details
def state_for_player(state: ScoponeState, sub: str) -> Dict[str, object]:
"""Serialize ``state`` hiding other players' hands.
Hands are reduced to a count, except for the requesting player's own
hand. Non-seated viewers simply see no hand at all. The platform
merges the session envelope (``id``, ``join_code``, ``game_type``)
into the view itself.
"""
viewer = state.player_for(sub)
players: List[Dict[str, object]] = []
for player in state.players:
view: Dict[str, object] = {
"sub": player.sub,
"name": player.name,
"seat": player.seat,
"team": TEAM_NAMES[player.team],
"cards_left": len(player.hand),
"captured_count": len(player.captured),
"scope": player.scope,
}
if viewer is not None and viewer.seat == player.seat:
view["hand"] = [c.code for c in player.hand]
players.append(view)
payload: Dict[str, object] = {
"phase": state.phase,
"target_score": state.target_score,
"napola": state.napola,
"hand_number": state.hand_number,
"dealer": state.dealer,
"turn": state.turn,
"scores": {"A": state.scores[0], "B": state.scores[1]},
"winner": None if state.winner is None else TEAM_NAMES[state.winner],
"table": [c.code for c in state.table],
"players": players,
"last_hand": state.hand_scores[-1] if state.hand_scores else None,
"last_move": state.last_move.to_json() if state.last_move else None,
"acknowledged": list(state.acked),
"hand_end_deadline": state.hand_end_deadline,
"turn_deadline": state.turn_deadline,
}
if viewer is not None and state.phase == PHASE_PLAYING and viewer.seat == state.turn:
payload["your_turn"] = True
# Only the player on turn receives their legal captures, so all the
# rule logic stays server-side.
legal_moves: Dict[str, List[List[str]]] = {}
for hand_card in viewer.hand:
options = legal_captures(state.table, hand_card)
if options:
legal_moves[hand_card.code] = [
[c.code for c in option] for option in options
]
payload["legal_moves"] = legal_moves
return payload
server/packages/tavolo-scopone/src/tavolo/scopone/errors.py
+34
View File
@@ -0,0 +1,34 @@
"""Scopone scientifico errors.
Most failure modes are game-independent and live in
:mod:`tavolo.platform.errors`; only genuinely scopone-specific errors
are defined here.
"""
from __future__ import annotations
from tavolo.platform.errors import (
AlreadyJoined,
GameError,
GameFinished,
GameNotFound,
GameNotStarted,
IllegalMove,
LobbyFull,
NotYourTurn,
)
__all__ = [
"AlreadyJoined",
"CardNotInHand",
"GameError",
"GameFinished",
"GameNotFound",
"GameNotStarted",
"IllegalMove",
"LobbyFull",
"NotYourTurn",
]
class CardNotInHand(GameError):
"""The played card is not held by the player."""
server/packages/tavolo-scopone/src/tavolo/scopone/plugin.py
+266
View File
@@ -0,0 +1,266 @@
"""The platform-facing adapter for scopone scientifico.
:class:`ScoponeEngine` implements
:class:`~tavolo.platform.engine.GameEngine` on top of the pure rules in
:mod:`tavolo.scopone.engine`: it translates platform calls (create, join,
websocket actions, deadlines) into rules-engine calls and back, keeping
all scopone knowledge inside this package. Nothing outside
``tavolo.scopone`` needs to know about phases, turns or hand scoring.
"""
from __future__ import annotations
from datetime import datetime, timezone
from logging import getLogger
from typing import Any, Dict, Mapping, Optional
from tavolo.platform.engine import (
Deadline,
GameEngine,
GameSession,
MatchResult,
PlayerResult,
Seat,
)
from tavolo.platform.errors import GameError, IllegalMove
from . import engine
from .state import (
DEFAULT_TARGET_SCORE,
PHASE_FINISHED,
PHASE_HAND_END,
PHASE_LOBBY,
PHASE_PLAYING,
TEAM_NAMES,
ScoponeState,
)
log = getLogger(__name__)
def _deadline_ms(iso: Optional[str]) -> Optional[int]:
"""Epoch milliseconds for an ISO-8601 deadline, ``None`` when absent
or unparseable."""
if not iso:
return None
try:
return int(datetime.fromisoformat(iso).timestamp() * 1000)
except ValueError:
return None
class ScoponeEngine(GameEngine):
"""Scopone scientifico as a platform game engine."""
id = "scopone_scientifico"
name = "Scopone scientifico"
description = (
"Four players in fixed partnerships, ten cards each and an empty "
"table. First team to the target score wins."
)
min_players = 4
max_players = 4
options_schema = {
"type": "object",
"properties": {
"target_score": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"default": DEFAULT_TARGET_SCORE,
"description": "Match points the winning team must reach.",
},
"napola": {
"type": "boolean",
"default": True,
"description": "Score the napola rule; a full denari "
"sweep wins the match",
},
},
}
def __init__(
self,
turn_timeout_seconds: int = engine.DEFAULT_TURN_TIMEOUT_SECONDS,
hand_ack_timeout_seconds: int = engine.DEFAULT_HAND_ACK_TIMEOUT_SECONDS,
) -> None:
self._turn_timeout = turn_timeout_seconds
self._hand_ack_timeout = hand_ack_timeout_seconds
# -- lobby ------------------------------------------------------------
def create(self, session: GameSession, options: Mapping[str, Any]) -> None:
target_score = options.get("target_score", DEFAULT_TARGET_SCORE)
if isinstance(target_score, bool) or not isinstance(target_score, int):
raise IllegalMove("target_score must be an integer")
napola = options.get("napola", True)
if not isinstance(napola, bool):
raise IllegalMove("napola must be a boolean")
creator = session.players[0]
session.state = engine.create_game(
creator_sub=creator.user_sub,
creator_name=creator.display_name,
target_score=target_score,
hand_ack_timeout=self._hand_ack_timeout,
turn_timeout=self._turn_timeout,
napola=napola,
)
# The creator takes seat 0, i.e. team A.
session.players[0] = Seat(
user_sub=creator.user_sub,
display_name=creator.display_name,
team=TEAM_NAMES[0],
)
def join(self, session: GameSession, user_sub: str, display_name: str) -> None:
seat = len(session.players)
engine.join_game(session.state, user_sub, display_name)
# join_game raises before appending on any violation, so the seat
# list stays in sync with the engine's players.
session.players.append(
Seat(
user_sub=user_sub,
display_name=display_name,
team=TEAM_NAMES[seat % 2],
)
)
def in_lobby(self, session: GameSession) -> bool:
return session.state.phase == PHASE_LOBBY
def lobby_view(self, session: GameSession) -> Dict[str, Any]:
state: ScoponeState = session.state
return {
"phase": state.phase,
"target_score": state.target_score,
"napola": state.napola,
}
# -- play -------------------------------------------------------------
def handle_action(
self, session: GameSession, user_sub: str, action: str, payload: Mapping[str, Any]
) -> None:
state: ScoponeState = session.state
if action == "play":
card = payload.get("card")
capture = payload.get("capture")
if not isinstance(card, str):
raise IllegalMove("'card' must be a card code string")
if capture is not None and (
not isinstance(capture, list)
or any(not isinstance(item, str) for item in capture)
):
raise IllegalMove("'capture' must be a list of card codes")
try:
engine.play(state, user_sub, card, capture)
except ValueError:
raise IllegalMove("invalid card code")
elif action == "ack":
engine.acknowledge_hand(state, user_sub)
else:
raise IllegalMove(f"unknown action: {action!r}")
def view_for(self, session: GameSession, user_sub: str) -> Dict[str, Any]:
return engine.state_for_player(session.state, user_sub)
def is_finished(self, session: GameSession) -> bool:
return session.state.phase == PHASE_FINISHED
def game_over_view(self, session: GameSession, user_sub: str) -> Dict[str, Any]:
state: ScoponeState = session.state
return {
"scores": {"A": state.scores[0], "B": state.scores[1]},
"winner": None if state.winner is None else TEAM_NAMES[state.winner],
}
# -- (de)serialization -------------------------------------------------
def state_to_json(self, state: Any) -> Dict[str, Any]:
assert isinstance(state, ScoponeState)
return state.to_json()
def state_from_json(self, data: Mapping[str, Any]) -> Any:
return ScoponeState.from_json(dict(data))
# -- deadlines ----------------------------------------------------------
def next_deadline(self, session: GameSession) -> Optional[Deadline]:
state: ScoponeState = session.state
if state.phase == PHASE_PLAYING and state.turn_deadline:
due_ms = _deadline_ms(state.turn_deadline)
if due_ms is None:
return None
return Deadline(
kind="turn",
due_at=datetime.fromtimestamp(due_ms / 1000, tz=timezone.utc),
token=f"turn:{state.hand_number}:{state.turn}:{due_ms}",
)
if state.phase == PHASE_HAND_END and state.hand_end_deadline:
due_ms = _deadline_ms(state.hand_end_deadline)
if due_ms is None:
return None
return Deadline(
kind="hand_end",
due_at=datetime.fromtimestamp(due_ms / 1000, tz=timezone.utc),
token=f"hand_end:{state.hand_number}:{due_ms}",
)
return None
def fire_deadline(self, session: GameSession, kind: str, token: str) -> None:
current = self.next_deadline(session)
if current is None or current.kind != kind or current.token != token:
# Overtaken by events (a play landed in time, the hand was
# acknowledged, the deadline moved): nothing to do.
raise GameError("stale deadline")
state: ScoponeState = session.state
if kind == "turn":
engine.auto_play(state)
actor = state.last_move.seat if state.last_move is not None else None
log.info(
"auto-played for seat %s (turn timeout, hand %d)",
actor,
state.hand_number,
)
elif kind == "hand_end":
for player in state.players:
engine.acknowledge_hand(state, player.sub)
log.info(
"hand %d auto-advanced after the acknowledgement timeout",
state.hand_number,
)
else: # pragma: no cover - next_deadline never emits other kinds
raise GameError(f"unknown deadline kind: {kind!r}")
# -- results -------------------------------------------------------------
def result(self, session: GameSession) -> MatchResult:
state: ScoponeState = session.state
if state.winner is None:
raise GameError("no result: the match is not finished")
teams = [
[p.sub for p in state.players if p.team == 0],
[p.sub for p in state.players if p.team == 1],
]
return MatchResult(
teams=teams,
winner_team=state.winner,
players=[
PlayerResult(
user_sub=player.sub,
seat=player.seat,
won=player.team == state.winner,
team=TEAM_NAMES[player.team],
score=float(state.scores[player.team]),
details={"scope": player.scope},
)
for player in state.players
],
summary={
"team_a_score": state.scores[0],
"team_b_score": state.scores[1],
"winner_team": TEAM_NAMES[state.winner],
"target_score": state.target_score,
"hands_played": state.hand_number,
"hand_scores": state.hand_scores,
},
)
server/packages/tavolo-scopone/src/tavolo/scopone/py.typed
View File
server/packages/tavolo-scopone/src/tavolo/scopone/state.py
+237
View File
@@ -0,0 +1,237 @@
"""In-memory representation of a scopone scientifico game.
The whole mutable game lives in :class:`ScoponeState`, which is serialized
to and from plain JSON for storage inside the platform's session envelope
(see :mod:`tavolo.platform.store`). Keeping the representation JSON-native
means the store needs no custom codecs and the state is inspectable with
``redis-cli``.
The state contains only game data: the platform owns the session envelope
(id, join code, seats, timestamps, stats persistence) — see
:class:`tavolo.platform.engine.GameSession`.
Deck convention: a 40-card Italian deck. Suits are ``D`` (denari),
``C`` (coppe), ``S`` (spade) and ``B`` (bastoni); ranks are ``1``..``10``.
A card is rendered as ``RRSUIT`` (e.g. ``07D`` is the settebello).
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
SUITS = ("D", "C", "S", "B")
RANKS = tuple(range(1, 11))
# Teams are derived from the seat: seats 0 and 2 form team A (index 0),
# seats 1 and 3 form team B (index 1). Team pairs always sit opposite each
# other, as in real scopone scientifico.
TEAM_A = 0
TEAM_B = 1
TEAM_NAMES = ("A", "B")
PHASE_LOBBY = "lobby"
PHASE_PLAYING = "playing"
# Between hands of an unfinished match: scoring summary shown to every
# player; the next hand is dealt once all four acknowledge (or the
# hand-end timeout elapses).
PHASE_HAND_END = "hand_end"
PHASE_FINISHED = "finished"
DEFAULT_TARGET_SCORE = 11
def team_of(seat: int) -> int:
return seat % 2
@dataclass(frozen=True)
class Card:
rank: int
suit: str
def __post_init__(self) -> None:
if self.suit not in SUITS:
raise ValueError(f"invalid suit: {self.suit!r}")
if self.rank not in RANKS:
raise ValueError(f"invalid rank: {self.rank!r}")
@property
def code(self) -> str:
return f"{self.rank:02d}{self.suit}"
@staticmethod
def parse(code: str) -> "Card":
code = str(code).upper()
if len(code) != 3 or not code[:2].isdigit():
raise ValueError(f"invalid card code: {code!r}")
return Card(rank=int(code[:2]), suit=code[2])
def to_json(self) -> str:
return self.code
@staticmethod
def from_json(value: Any) -> "Card":
return Card.parse(str(value))
def parse_card(code: Any) -> Card:
"""Parse a card code, raising :class:`ValueError` on malformed input."""
try:
return Card.parse(str(code))
except ValueError:
raise
@dataclass
class Move:
"""Record of a single play, broadcast so every client can show who
played which card and what it captured."""
seat: int
name: str
card: str
captured: List[str] = field(default_factory=list)
scopa: bool = False
def to_json(self) -> Dict[str, Any]:
return {
"seat": self.seat,
"name": self.name,
"card": self.card,
"captured": list(self.captured),
"scopa": self.scopa,
}
@staticmethod
def from_json(data: Dict[str, Any]) -> "Move":
return Move(
seat=int(data["seat"]),
name=str(data["name"]),
card=str(data["card"]),
captured=[str(c) for c in data.get("captured", [])],
scopa=bool(data.get("scopa", False)),
)
@dataclass
class PlayerState:
sub: str
name: str
seat: int
hand: List[Card] = field(default_factory=list)
captured: List[Card] = field(default_factory=list)
scope: int = 0
@property
def team(self) -> int:
return team_of(self.seat)
def to_json(self) -> Dict[str, Any]:
return {
"sub": self.sub,
"name": self.name,
"seat": self.seat,
"hand": [c.to_json() for c in self.hand],
"captured": [c.to_json() for c in self.captured],
"scope": self.scope,
}
@staticmethod
def from_json(data: Dict[str, Any]) -> "PlayerState":
return PlayerState(
sub=str(data["sub"]),
name=str(data["name"]),
seat=int(data["seat"]),
hand=[Card.from_json(c) for c in data.get("hand", [])],
captured=[Card.from_json(c) for c in data.get("captured", [])],
scope=int(data.get("scope", 0)),
)
@dataclass
class ScoponeState:
target_score: int = DEFAULT_TARGET_SCORE
# Whether the napola rule is scored (denari run from the ace; a full
# suit wins the match instantly). Default on.
napola: bool = True
phase: str = PHASE_LOBBY
players: List[PlayerState] = field(default_factory=list)
table: List[Card] = field(default_factory=list)
dealer: int = 0
turn: int = 0
hand_number: int = 1
scores: List[int] = field(default_factory=lambda: [0, 0])
winner: Optional[int] = None
last_taker: Optional[int] = None
# Per-hand points awarded, for a compact audit trail in the API.
hand_scores: List[Dict[str, Any]] = field(default_factory=list)
# The most recent play in the current hand, for move announcements.
last_move: Optional[Move] = None
# While phase == "hand_end": seats that acknowledged the summary, and
# when the auto-continue timeout fires.
acked: List[int] = field(default_factory=list)
hand_end_deadline: Optional[str] = None
# Seconds the hand-end summary waits before dealing anyway.
hand_ack_timeout: int = 30
# While phase == "playing": when the server plays a random legal card
# for the player on turn.
turn_deadline: Optional[str] = None
turn_timeout: int = 30
# -- serialization ----------------------------------------------------
def to_json(self) -> Dict[str, Any]:
return {
"target_score": self.target_score,
"napola": self.napola,
"phase": self.phase,
"players": [p.to_json() for p in self.players],
"table": [c.to_json() for c in self.table],
"dealer": self.dealer,
"turn": self.turn,
"hand_number": self.hand_number,
"scores": list(self.scores),
"winner": self.winner,
"last_taker": self.last_taker,
"hand_scores": list(self.hand_scores),
"last_move": self.last_move.to_json() if self.last_move else None,
"acked": list(self.acked),
"hand_end_deadline": self.hand_end_deadline,
"hand_ack_timeout": self.hand_ack_timeout,
"turn_deadline": self.turn_deadline,
"turn_timeout": self.turn_timeout,
}
@staticmethod
def from_json(data: Dict[str, Any]) -> "ScoponeState":
return ScoponeState(
target_score=int(data.get("target_score", DEFAULT_TARGET_SCORE)),
napola=bool(data.get("napola", True)),
phase=str(data.get("phase", PHASE_LOBBY)),
players=[PlayerState.from_json(p) for p in data.get("players", [])],
table=[Card.from_json(c) for c in data.get("table", [])],
dealer=int(data.get("dealer", 0)),
turn=int(data.get("turn", 0)),
hand_number=int(data.get("hand_number", 1)),
scores=[int(x) for x in data.get("scores", [0, 0])],
winner=data.get("winner"),
last_taker=data.get("last_taker"),
hand_scores=list(data.get("hand_scores", [])),
last_move=Move.from_json(data["last_move"]) if data.get("last_move") else None,
acked=[int(s) for s in data.get("acked", [])],
hand_end_deadline=data.get("hand_end_deadline"),
hand_ack_timeout=int(data.get("hand_ack_timeout", 30)),
turn_deadline=data.get("turn_deadline"),
turn_timeout=int(data.get("turn_timeout", 30)),
)
# -- helpers ----------------------------------------------------------
def player_for(self, sub: str) -> Optional[PlayerState]:
for player in self.players:
if player.sub == sub:
return player
return None
def seated(self, sub: str) -> bool:
return self.player_for(sub) is not None
server/packages/tavolo-scopone/tests/test_engine.py
+535
View File
@@ -0,0 +1,535 @@
"""Rule engine tests: captures, scope, scoring and full-match simulation."""
from __future__ import annotations
import random
import unittest
from tavolo.scopone import engine
from tavolo.scopone.errors import (
CardNotInHand,
GameFinished,
GameNotStarted,
IllegalMove,
NotYourTurn,
)
from tavolo.scopone.state import (
PHASE_FINISHED,
PHASE_PLAYING,
Card,
ScoponeState,
PlayerState,
)
def card(code: str) -> Card:
return Card.parse(code)
def make_state(
hands,
table,
turn: int = 0,
*,
captured=None,
scope=None,
target: int = 11,
last_taker=None,
) -> ScoponeState:
"""Build a controlled game state directly (bypassing the deal)."""
state = ScoponeState(
target_score=target,
phase=PHASE_PLAYING,
turn=turn,
last_taker=last_taker,
)
for seat, hand in enumerate(hands):
state.players.append(
PlayerState(sub=f"p{seat}", name=f"p{seat}", seat=seat,
hand=[card(c) for c in hand])
)
if captured is not None:
for player, codes in zip(state.players, captured):
player.captured = [card(c) for c in codes]
if scope is not None:
for player, value in zip(state.players, scope):
player.scope = value
state.table = [card(c) for c in table]
return state
class DeckTest(unittest.TestCase):
def test_full_deck_has_40_unique_cards(self) -> None:
deck = engine.full_deck()
self.assertEqual(40, len(deck))
self.assertEqual(40, len({c.code for c in deck}))
self.assertEqual(4, len({c.suit for c in deck}))
self.assertEqual(4, sum(1 for c in deck if c.rank == 7))
def test_shuffled_deck_is_permutation(self) -> None:
deck = engine.shuffled_deck()
self.assertEqual(
sorted(c.code for c in engine.full_deck()),
sorted(c.code for c in deck),
)
class CaptureTest(unittest.TestCase):
def test_equal_card_is_mandatory(self) -> None:
table = [card("05C"), card("02D"), card("03S")]
options = engine.legal_captures(table, card("05D"))
self.assertEqual([["05C"]], [[c.code for c in o] for o in options])
def test_sum_combination(self) -> None:
table = [card("01C"), card("03C"), card("02S")]
options = engine.legal_captures(table, card("04D"))
self.assertEqual([["01C", "03C"]], [[c.code for c in o] for o in options])
def test_multiple_equal_cards_each_a_separate_option(self) -> None:
table = [card("05C"), card("05S")]
options = engine.legal_captures(table, card("05D"))
self.assertEqual(
[["05C"], ["05S"]], sorted([[c.code for c in o] for o in options])
)
def test_no_capture(self) -> None:
table = [card("09C"), card("08S")]
self.assertEqual([], engine.legal_captures(table, card("02D")))
def test_play_without_capture_places_card_on_table(self) -> None:
state = make_state([["02D", "03D"], ["01C"], ["01S"], ["01B"]],
table=["09C"])
engine.play(state, "p0", "02D")
self.assertIn("02D", [c.code for c in state.table])
self.assertNotIn("02D", [c.code for c in state.players[0].hand])
self.assertEqual(1, state.turn)
def test_play_capture_and_scopa(self) -> None:
state = make_state([["02D", "09C"], ["01C"], ["01S"], ["01B"]],
table=["02C"])
engine.play(state, "p0", "02D", ["02C"])
self.assertEqual(1, state.players[0].scope)
self.assertEqual([], state.table)
self.assertEqual(
["02C", "02D"], [c.code for c in state.players[0].captured]
)
# The move is recorded for the "who played what" announcement.
assert state.last_move is not None
self.assertEqual(0, state.last_move.seat)
self.assertEqual("p0", state.last_move.name)
self.assertEqual("02D", state.last_move.card)
self.assertEqual(["02C"], state.last_move.captured)
self.assertTrue(state.last_move.scopa)
def test_play_without_capture_records_move(self) -> None:
state = make_state([["02D", "03D"], ["01C"], ["01S"], ["01B"]],
table=["09C"])
engine.play(state, "p0", "02D")
assert state.last_move is not None
self.assertEqual("02D", state.last_move.card)
self.assertEqual([], state.last_move.captured)
self.assertFalse(state.last_move.scopa)
def test_illegal_combination_when_equal_card_present(self) -> None:
state = make_state([["05D"], ["01C"], ["01S"], ["01B"]],
table=["05C", "02D", "03S"])
with self.assertRaises(IllegalMove):
engine.play(state, "p0", "05D", ["02D", "03S"])
def test_illegal_capture_rejected(self) -> None:
state = make_state([["04D"], ["01C"], ["01S"], ["01B"]],
table=["02C", "03S"])
with self.assertRaises(IllegalMove):
engine.play(state, "p0", "04D", ["02C"])
def test_no_capture_requested_when_capture_possible(self) -> None:
state = make_state([["05D"], ["01C"], ["01S"], ["01B"]],
table=["05C"])
with self.assertRaises(IllegalMove):
engine.play(state, "p0", "05D")
def test_not_your_turn(self) -> None:
state = make_state([["02D"], ["03D"], ["04D"], ["05D"]],
table=[], turn=1)
with self.assertRaises(NotYourTurn):
engine.play(state, "p0", "02D")
def test_card_not_in_hand(self) -> None:
state = make_state([["02D"], ["03D"], ["04D"], ["05D"]], table=[])
with self.assertRaises(CardNotInHand):
engine.play(state, "p0", "07D")
def test_finished_game_rejects_moves(self) -> None:
state = make_state([["02D"], ["03D"], ["04D"], ["05D"]], table=[])
state.phase = PHASE_FINISHED
with self.assertRaises(GameFinished):
engine.play(state, "p0", "02D")
class LastPlayTest(unittest.TestCase):
def test_no_scopa_on_last_play_of_hand(self) -> None:
# p0 plays the last card of the hand (everyone else is already
# empty): the capture empties the table but must NOT count as a
# scopa. Team A still reaches the target of 2 with carte + denara.
state = make_state([["02D"], [], [], []],
table=["02C"], target=2)
engine.play(state, "p0", "02D", ["02C"])
self.assertEqual(PHASE_FINISHED, state.phase)
self.assertEqual(0, state.winner)
self.assertEqual(0, state.hand_scores[-1]["scope"]["A"])
def test_table_swept_to_last_taker(self) -> None:
# target 2 so the game ends on this hand and the capture piles are
# not reset by the next deal.
state = make_state([["02D"], [], [], []],
table=["05C", "04D"], last_taker=1, target=2)
engine.play(state, "p0", "02D")
captured = {c.code for c in state.players[1].captured}
self.assertEqual({"05C", "04D", "02D"}, captured)
self.assertEqual(PHASE_FINISHED, state.phase)
self.assertEqual(1, state.winner)
class ScoringTest(unittest.TestCase):
def test_primiera_values_and_all_suits_requirement(self) -> None:
self.assertEqual(70, engine.primiera_score(
[card(c) for c in ["07D", "06C", "01S", "05B"]]))
self.assertEqual(0, engine.primiera_score(
[card(c) for c in ["07D", "06C", "01S"]]))
self.assertEqual(40, engine.primiera_score(
[card(c) for c in ["08D", "09C", "10S", "10B"]]))
def test_hand_points_carte_denara_settebello_primiera_scope(self) -> None:
state = make_state(
[[], [], [], []],
table=[],
captured=[
["07D", "06C", "01S", "05B"], # seat 0, team A
["03D", "04C", "07S", "02B"], # seat 1, team B
["02D"], # seat 2, team A
["10D", "10C", "10S", "10B"], # seat 3, team B
],
scope=[1, 0, 0, 2],
)
points, details = engine.hand_points(state)
self.assertEqual([3, 3], points)
self.assertEqual({"A": 5, "B": 8}, details["cards"])
self.assertEqual({"A": 2, "B": 2}, details["denara"])
self.assertEqual({"A": True, "B": False}, details["settebello"])
self.assertEqual({"A": 70, "B": 60}, details["primiera"])
self.assertEqual({"A": 1, "B": 2}, details["scope"])
def test_ties_award_nothing(self) -> None:
state = make_state(
[[], [], [], []],
table=[],
captured=[
["06C", "01S", "05B", "02D"],
["06S", "01B", "05D", "02C"],
[],
[],
],
scope=[0, 0, 0, 0],
)
points, _ = engine.hand_points(state)
# Equal cards, equal denara, equal primiera and no settebello:
# everything ties, so no points at all.
self.assertEqual([0, 0], points)
class NapolaTest(unittest.TestCase):
def test_napola_score_runs(self) -> None:
self.assertEqual(0, engine.napola_score(
[card(c) for c in ["02D", "03D", "04D"]])) # no ace
self.assertEqual(0, engine.napola_score(
[card(c) for c in ["01D", "02D"]])) # too short
self.assertEqual(3, engine.napola_score(
[card(c) for c in ["03D", "01D", "02D"]])) # order-independent
self.assertEqual(4, engine.napola_score(
[card(c) for c in ["01D", "02D", "03D", "04D", "07C"]]))
self.assertEqual(3, engine.napola_score(
[card(c) for c in ["01D", "02D", "03D", "05D"]])) # broken run
self.assertEqual(10, engine.napola_score(
[card(f"{rank:02d}D") for rank in range(1, 11)]))
def test_hand_points_napola(self) -> None:
state = make_state(
[[], [], [], []],
table=[],
captured=[
["01D", "02D", "03D", "04C"], # seat 0, team A
["05D", "06D", "07D", "08D"], # seat 1, team B
["09D", "10D", "01C", "02C"], # seat 2, team A
["03C", "05C", "06C", "07C"], # seat 3, team B
],
)
points, details = engine.hand_points(state)
# Team A has the ace-led run 01D-03D (3 points); team B's denari
# start at the 5, so no napola. Carte tie (8 each), denara to A
# (5 vs 4), settebello to B, primiere tied at 0 (missing suits).
self.assertEqual({"A": 3, "B": 0}, details["napola"])
self.assertEqual("A", details["award"]["napola"])
self.assertEqual([4, 1], points)
def test_napola_disabled(self) -> None:
state = make_state(
[[], [], [], []],
table=[],
captured=[
["01D", "02D", "03D", "04C"],
["05D", "06D", "07D", "08D"],
["09D", "10D", "01C", "02C"],
["03C", "05C", "06C", "07C"],
],
)
state.napola = False
points, details = engine.hand_points(state)
self.assertNotIn("napola", details)
self.assertEqual([1, 1], points)
def test_full_denari_sweep_wins_match_instantly(self) -> None:
# Team A already captured the whole denari suit; the last play of
# the hand cannot capture. Team B leads 50-0, yet the napola ends
# the match in team A's favour, well below the target of 100.
state = make_state(
[["02C"], [], [], []],
table=[],
target=100,
captured=[
[f"{rank:02d}D" for rank in range(1, 11)],
[],
[],
[],
],
)
state.scores = [0, 50]
engine.play(state, "p0", "02C")
self.assertEqual(PHASE_FINISHED, state.phase)
self.assertEqual(0, state.winner)
self.assertLess(state.scores[0], 100)
self.assertEqual(10, state.hand_scores[-1]["napola"]["A"])
def test_napola_serialization_roundtrip(self) -> None:
state = make_state([["02D"], [], [], []], table=[])
self.assertTrue(state.napola)
state.napola = False
self.assertFalse(ScoponeState.from_json(state.to_json()).napola)
# States serialized before the option existed default to enabled.
data = state.to_json()
del data["napola"]
self.assertTrue(ScoponeState.from_json(data).napola)
def test_create_game_napola_default_and_override(self) -> None:
self.assertTrue(engine.create_game("p0", "p0").napola)
self.assertFalse(
engine.create_game("p0", "p0", napola=False).napola
)
class MatchFlowTest(unittest.TestCase):
def test_join_starts_when_full(self) -> None:
state = engine.create_game("p0", "p0", target_score=11)
self.assertEqual(1, len(state.players))
engine.join_game(state, "p1", "p1")
engine.join_game(state, "p2", "p2")
self.assertEqual("lobby", state.phase)
engine.join_game(state, "p3", "p3")
self.assertEqual(PHASE_PLAYING, state.phase)
self.assertEqual(4, len(state.players))
for player in state.players:
self.assertEqual(10, len(player.hand))
self.assertEqual([], state.table)
self.assertEqual(1, state.turn) # dealer is seat 0
def test_match_ends_when_target_reached(self) -> None:
state = make_state([["02D"], [], [], []],
table=["02C"], target=1)
engine.play(state, "p0", "02D", ["02C"])
self.assertEqual(PHASE_FINISHED, state.phase)
self.assertEqual(0, state.winner)
self.assertGreaterEqual(state.scores[0], 1)
def test_state_for_player_hides_other_hands(self) -> None:
state = make_state([["02D", "03C"], ["04D"], ["05D"], ["06D"]],
table=["07C"])
view = engine.state_for_player(state, "p0")
players = {p["seat"]: p for p in view["players"]}
self.assertEqual(["02D", "03C"], players[0]["hand"])
self.assertNotIn("hand", players[1])
self.assertEqual(1, players[1]["cards_left"])
self.assertEqual(["07C"], view["table"])
self.assertTrue(view.get("your_turn"))
def test_legal_moves_only_for_player_on_turn(self) -> None:
state = make_state([["02D", "03C"], ["04D"], ["05D"], ["06D"]],
table=["07C"])
view = engine.state_for_player(state, "p0")
legal = view["legal_moves"]
# 02D can capture nothing; 03C has no combination either (only 07C
# on the table).
self.assertEqual({}, legal)
state = make_state([["09D"], ["04D"], ["05D"], ["06D"]],
table=["07C", "02S"])
view = engine.state_for_player(state, "p0")
self.assertEqual({"09D": [["07C", "02S"]]}, view["legal_moves"])
# A player who is not on turn gets no legal_moves key.
other = engine.state_for_player(state, "p1")
self.assertNotIn("legal_moves", other)
self.assertNotIn("your_turn", other)
def test_full_random_match_reaches_completion(self) -> None:
state = engine.create_game("p0", "p0", target_score=11)
for i in range(1, 4):
engine.join_game(state, f"p{i}", f"p{i}")
moves = 0
while state.phase != PHASE_FINISHED and moves < 200000:
if state.phase == "hand_end":
for p in state.players:
engine.acknowledge_hand(state, p.sub)
continue
player = next(p for p in state.players if p.seat == state.turn)
played = player.hand[0]
options = engine.legal_captures(state.table, played)
capture = [c.code for c in options[0]] if options else None
engine.play(state, player.sub, played.code, capture)
moves += 1
self.assertEqual(PHASE_FINISHED, state.phase)
self.assertIn(state.winner, (0, 1))
# At the end all 40 cards are captured and no hand is left.
self.assertEqual([], state.table)
self.assertTrue(all(not p.hand for p in state.players))
self.assertEqual(40, sum(len(p.captured) for p in state.players))
class HandEndAckTest(unittest.TestCase):
def _hand_end_state(self) -> ScoponeState:
"""Drive a game into the hand_end phase with a one-card hand."""
state = make_state([["02D"], [], [], []],
table=["02C"], target=11)
engine.play(state, "p0", "02D", ["02C"])
return state
def test_end_of_hand_pauses_for_acknowledgement(self) -> None:
state = self._hand_end_state()
self.assertEqual("hand_end", state.phase)
# Nobody has acknowledged yet, and no new hand was dealt.
self.assertEqual([], state.acked)
self.assertEqual(1, state.hand_number)
self.assertTrue(state.hand_end_deadline)
# Capture piles stay visible during the summary.
self.assertEqual(["02C", "02D"],
[c.code for c in state.players[0].captured])
# The summary carries the award map.
summary = state.hand_scores[-1]
self.assertEqual(1, summary["hand"])
self.assertIn("award", summary)
def test_play_during_hand_end_is_rejected(self) -> None:
state = self._hand_end_state()
with self.assertRaises(IllegalMove):
engine.play(state, "p0", "02D")
def test_ack_all_four_deals_next_hand(self) -> None:
state = self._hand_end_state()
dealer_before = state.dealer
for i, sub in enumerate(("p0", "p1", "p2")):
engine.acknowledge_hand(state, sub)
self.assertEqual(list(range(i + 1)), state.acked)
self.assertEqual("hand_end", state.phase)
engine.acknowledge_hand(state, "p3")
self.assertEqual("playing", state.phase)
self.assertEqual(2, state.hand_number)
self.assertEqual((dealer_before + 1) % 4, state.dealer)
self.assertEqual([], state.acked)
self.assertIsNone(state.hand_end_deadline)
self.assertIsNone(state.last_move)
for player in state.players:
self.assertEqual(10, len(player.hand))
self.assertEqual([], player.captured)
self.assertEqual((dealer_before + 2) % 4, state.turn)
def test_double_ack_is_idempotent(self) -> None:
state = self._hand_end_state()
engine.acknowledge_hand(state, "p0")
engine.acknowledge_hand(state, "p0")
self.assertEqual([0], state.acked)
def test_ack_outside_hand_end_is_rejected(self) -> None:
state = make_state([["02D"], ["03D"], ["04D"], ["05D"]], table=[])
with self.assertRaises(IllegalMove):
engine.acknowledge_hand(state, "p0")
def test_ack_by_non_player_is_rejected(self) -> None:
state = self._hand_end_state()
with self.assertRaises(NotYourTurn):
engine.acknowledge_hand(state, "mallory")
def test_state_exposes_ack_progress(self) -> None:
state = self._hand_end_state()
engine.acknowledge_hand(state, "p1")
view = engine.state_for_player(state, "p0")
self.assertEqual([1], view["acknowledged"])
self.assertTrue(view["hand_end_deadline"])
self.assertIsNotNone(view["last_hand"])
self.assertIn("award", view["last_hand"])
class AutoPlayTest(unittest.TestCase):
def test_auto_play_plays_a_card_and_advances_turn(self) -> None:
state = make_state([["02D", "03C"], ["04D"], ["05D"], ["06D"]],
table=["09B"])
state.turn_deadline = "2000-01-01T00:00:00+00:00"
engine.auto_play(state, random.Random(7))
self.assertEqual(1, state.turn)
self.assertEqual(1, len(state.players[0].hand))
# The played card could not capture the nine, so the table grew.
self.assertEqual(2, len(state.table))
self.assertIsNotNone(state.last_move)
assert state.last_move is not None
self.assertEqual(0, state.last_move.seat)
self.assertNotEqual("2000-01-01T00:00:00+00:00", state.turn_deadline)
def test_auto_play_takes_a_mandatory_capture(self) -> None:
# p0 holds only the five of denari, which must capture the equal
# five of coppe instead of the unrelated nine on the table.
state = make_state([["05D"], ["04D"], ["05D"], ["06D"]],
table=["05C", "09B"])
engine.auto_play(state)
self.assertIsNotNone(state.last_move)
assert state.last_move is not None
self.assertEqual("05D", state.last_move.card)
self.assertEqual(["05C"], state.last_move.captured)
self.assertEqual(["09B"], [c.code for c in state.table])
self.assertEqual(["05C", "05D"],
[c.code for c in state.players[0].captured])
def test_auto_play_can_end_the_hand_and_clears_deadline(self) -> None:
state = make_state([["02D"], [], [], []], table=["02C"])
state.turn_deadline = "2000-01-01T00:00:00+00:00"
engine.auto_play(state)
self.assertEqual("hand_end", state.phase)
self.assertIsNone(state.turn_deadline)
self.assertTrue(state.hand_end_deadline)
def test_auto_play_requires_playing_phase(self) -> None:
state = make_state([["02D"], ["04D"], ["05D"], ["06D"]], table=[])
state.phase = "hand_end"
with self.assertRaises(GameNotStarted):
engine.auto_play(state)
def test_create_game_copies_turn_timeout_and_arms_deadline(self) -> None:
state = engine.create_game("p0", "p0", turn_timeout=7)
self.assertEqual(7, state.turn_timeout)
for i in range(1, 4):
engine.join_game(state, f"p{i}", f"p{i}")
self.assertEqual(PHASE_PLAYING, state.phase)
self.assertTrue(state.turn_deadline)
view = engine.state_for_player(state, "p0")
self.assertTrue(view["turn_deadline"])
if __name__ == "__main__":
unittest.main()
server/packages/tavolo-scopone/tests/test_plugin.py
+289
View File
@@ -0,0 +1,289 @@
"""ScoponeEngine adapter tests: the platform contract over the pure rules."""
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from tavolo.platform import GameSession, Seat
from tavolo.platform.errors import GameError, IllegalMove
from tavolo.scopone import ScoponeEngine
from tavolo.scopone.state import PHASE_FINISHED, PHASE_HAND_END, PHASE_PLAYING, ScoponeState
def _session(engine: ScoponeEngine, **options) -> GameSession:
session = GameSession(
id="s1",
game_type=engine.id,
join_code="CODE01",
creator_sub="alice",
players=[Seat(user_sub="alice", display_name="Alice")],
)
engine.create(session, options)
return session
def _started(engine: ScoponeEngine, **options) -> GameSession:
session = _session(engine, **options)
for name in ("bob", "carol", "dave"):
engine.join(session, name, name.capitalize())
return session
class CreateTest(unittest.TestCase):
def test_create_seats_creator_on_team_a(self) -> None:
engine = ScoponeEngine()
session = _session(engine)
self.assertEqual("A", session.players[0].team)
self.assertIsInstance(session.state, ScoponeState)
self.assertEqual(11, session.state.target_score)
self.assertTrue(session.state.napola)
def test_create_options(self) -> None:
engine = ScoponeEngine()
session = _session(engine, target_score=16, napola=False)
self.assertEqual(16, session.state.target_score)
self.assertFalse(session.state.napola)
def test_create_rejects_bad_options(self) -> None:
engine = ScoponeEngine()
with self.assertRaises(IllegalMove):
_session(engine, target_score=0)
with self.assertRaises(IllegalMove):
_session(engine, target_score="eleven")
with self.assertRaises(IllegalMove):
_session(engine, napola="yes")
def test_timeouts_come_from_the_engine(self) -> None:
engine = ScoponeEngine(turn_timeout_seconds=7, hand_ack_timeout_seconds=9)
session = _session(engine)
self.assertEqual(7, session.state.turn_timeout)
self.assertEqual(9, session.state.hand_ack_timeout)
class JoinTest(unittest.TestCase):
def test_join_assigns_teams_and_starts(self) -> None:
engine = ScoponeEngine()
session = _session(engine)
self.assertTrue(engine.in_lobby(session))
for name, team in (("bob", "B"), ("carol", "A"), ("dave", "B")):
engine.join(session, name, name.capitalize())
self.assertEqual(team, session.players[-1].team)
self.assertFalse(engine.in_lobby(session))
self.assertEqual(PHASE_PLAYING, session.state.phase)
def test_join_errors(self) -> None:
from tavolo.platform.errors import AlreadyJoined, GameNotStarted
engine = ScoponeEngine()
session = _session(engine)
with self.assertRaises(AlreadyJoined):
engine.join(session, "alice", "Alice")
for name in ("bob", "carol", "dave"):
engine.join(session, name, name.capitalize())
# The lobby filled up and the match started: late joins and even
# re-joins are rejected as "already started".
with self.assertRaises(GameNotStarted):
engine.join(session, "erin", "Erin")
with self.assertRaises(GameNotStarted):
engine.join(session, "alice", "Alice")
class ActionTest(unittest.TestCase):
def test_unknown_action_rejected(self) -> None:
engine = ScoponeEngine()
session = _started(engine)
with self.assertRaises(IllegalMove):
engine.handle_action(session, "alice", "dance", {})
def test_play_validates_payload(self) -> None:
engine = ScoponeEngine()
session = _started(engine)
with self.assertRaises(IllegalMove):
engine.handle_action(session, "bob", "play", {"card": 42})
with self.assertRaises(IllegalMove):
engine.handle_action(session, "bob", "play", {})
with self.assertRaises(IllegalMove):
engine.handle_action(session, "bob", "play", {"card": "01D", "capture": "02C"})
def test_invalid_card_code_is_illegal_move(self) -> None:
engine = ScoponeEngine()
session = _started(engine)
with self.assertRaises(IllegalMove):
engine.handle_action(session, "bob", "play", {"card": "nope"})
def test_finished_match_rejects_actions(self) -> None:
from tavolo.platform.errors import GameFinished
engine = ScoponeEngine()
session = _started(engine, target_score=1)
# Drive to completion: keep playing legal moves until finished.
from tavolo.scopone import engine as rules
moves = 0
while not engine.is_finished(session) and moves < 200000:
state = session.state
if state.phase == "hand_end":
for p in state.players:
engine.handle_action(session, p.sub, "ack", {})
continue
player = next(p for p in state.players if p.seat == state.turn)
played = player.hand[0]
options = rules.legal_captures(state.table, played)
payload = {"card": played.code}
if options:
payload["capture"] = [c.code for c in options[0]]
engine.handle_action(session, player.sub, "play", payload)
moves += 1
self.assertTrue(engine.is_finished(session))
with self.assertRaises(GameFinished):
engine.handle_action(session, "alice", "play", {"card": "01D"})
class ViewTest(unittest.TestCase):
def test_view_for_hides_other_hands(self) -> None:
engine = ScoponeEngine()
session = _started(engine)
view = engine.view_for(session, "alice")
players = {p["seat"]: p for p in view["players"]}
self.assertIn("hand", players[0])
self.assertNotIn("hand", players[1])
# The envelope is the platform's job, not the view's.
self.assertNotIn("id", view)
self.assertNotIn("join_code", view)
self.assertNotIn("game_type", view)
def test_lobby_view(self) -> None:
engine = ScoponeEngine()
session = _session(engine, target_score=16)
lobby = engine.lobby_view(session)
self.assertEqual("lobby", lobby["phase"])
self.assertEqual(16, lobby["target_score"])
self.assertTrue(lobby["napola"])
class SerializationTest(unittest.TestCase):
def test_state_roundtrip(self) -> None:
engine = ScoponeEngine()
session = _started(engine)
restored = engine.state_from_json(engine.state_to_json(session.state))
self.assertIsInstance(restored, ScoponeState)
self.assertEqual(session.state.phase, restored.phase)
self.assertEqual(session.state.turn, restored.turn)
self.assertEqual(
[p.sub for p in session.state.players],
[p.sub for p in restored.players],
)
class DeadlineTest(unittest.TestCase):
def test_no_deadline_in_lobby(self) -> None:
engine = ScoponeEngine()
self.assertIsNone(engine.next_deadline(_session(engine)))
def test_turn_deadline_and_revalidation(self) -> None:
engine = ScoponeEngine()
session = _started(engine)
deadline = engine.next_deadline(session)
assert deadline is not None
self.assertEqual("turn", deadline.kind)
# A forged token is stale.
with self.assertRaises(GameError):
engine.fire_deadline(session, "turn", "turn:1:1:0")
turn_before = session.state.turn
engine.fire_deadline(session, deadline.kind, deadline.token)
self.assertNotEqual(turn_before, session.state.turn)
def test_stale_deadline_after_play(self) -> None:
from tavolo.scopone import engine as rules
engine = ScoponeEngine()
session = _started(engine)
deadline = engine.next_deadline(session)
assert deadline is not None
# A play lands in time: the armed deadline is overtaken.
state = session.state
player = next(p for p in state.players if p.seat == state.turn)
played = player.hand[0]
options = rules.legal_captures(state.table, played)
payload = {"card": played.code}
if options:
payload["capture"] = [c.code for c in options[0]]
engine.handle_action(session, player.sub, "play", payload)
with self.assertRaises(GameError):
engine.fire_deadline(session, deadline.kind, deadline.token)
def test_hand_end_deadline_advances(self) -> None:
engine = ScoponeEngine()
session = _started(engine)
state = session.state
# Force the hand-end phase with an imminent deadline.
state.phase = PHASE_HAND_END
state.hand_end_deadline = (
datetime.now(timezone.utc) + timedelta(seconds=60)
).isoformat()
deadline = engine.next_deadline(session)
assert deadline is not None
self.assertEqual("hand_end", deadline.kind)
engine.fire_deadline(session, deadline.kind, deadline.token)
self.assertEqual(PHASE_PLAYING, session.state.phase)
self.assertEqual(2, session.state.hand_number)
class ResultTest(unittest.TestCase):
def test_result_of_finished_match(self) -> None:
from tavolo.scopone import engine as rules
engine = ScoponeEngine()
session = _started(engine, target_score=1)
moves = 0
while not engine.is_finished(session) and moves < 200000:
state = session.state
if state.phase == "hand_end":
for p in state.players:
engine.handle_action(session, p.sub, "ack", {})
continue
player = next(p for p in state.players if p.seat == state.turn)
played = player.hand[0]
options = rules.legal_captures(state.table, played)
payload = {"card": played.code}
if options:
payload["capture"] = [c.code for c in options[0]]
engine.handle_action(session, player.sub, "play", payload)
moves += 1
result = engine.result(session)
self.assertEqual(2, len(result.teams))
self.assertIn(result.winner_team, (0, 1))
self.assertEqual(4, len(result.players))
for player in result.players:
self.assertEqual(
player.won, player.team == ("A" if result.winner_team == 0 else "B")
)
self.assertIn("team_a_score", result.summary)
self.assertIn("hands_played", result.summary)
def test_result_requires_finished_match(self) -> None:
engine = ScoponeEngine()
with self.assertRaises(GameError):
engine.result(_started(engine))
def test_game_over_view(self) -> None:
engine = ScoponeEngine()
session = _started(engine)
session.state.phase = PHASE_FINISHED
session.state.winner = 1
session.state.scores = [3, 11]
over = engine.game_over_view(session, "alice")
self.assertEqual({"A": 3, "B": 11}, over["scores"])
self.assertEqual("B", over["winner"])
def test_registry_metadata(self) -> None:
engine = ScoponeEngine()
self.assertEqual("scopone_scientifico", engine.id)
self.assertEqual(4, engine.min_players)
self.assertEqual(4, engine.max_players)
self.assertIn("target_score", engine.options_schema["properties"])
self.assertIn("napola", engine.options_schema["properties"])
if __name__ == "__main__":
unittest.main()