woggioni
d4d437c1a1
New tavolo-scopa package with a pure rules engine and the platform adapters for two-player scopa and four-player fixed-partnership scopa a quattro: three cards each and four on the table, capture by rank or sum, sweep for scope and score carte/denara/settebello/primiera. The optional asso pigliatutto creation option turns either variant into scopa d'assi (an ace sweeps the whole table and scores no scopa), reusing scopone's rule semantics. Wire both engines into the app and the Docker install, add the scopa table view to the SPA (duo and four-seat layouts, deck count, captured/scope stats, hand-summary and final screens, player names in the two-player summary), and cover the rules, the platform contract, the REST/WebSocket flows, match persistence and Elo.
336 lines
13 KiB
Python
336 lines
13 KiB
Python
"""Adapter tests: the platform contract over the pure scopa rules."""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from tavolo.platform import GameSession, Seat
|
|
from tavolo.platform.errors import (
|
|
AlreadyJoined,
|
|
GameError,
|
|
GameFinished,
|
|
GameNotStarted,
|
|
IllegalMove,
|
|
)
|
|
from tavolo.scopa import ScopaEngine, ScopaQuattroEngine
|
|
from tavolo.scopa import engine as rules
|
|
from tavolo.scopa.state import (
|
|
PHASE_FINISHED,
|
|
PHASE_HAND_END,
|
|
PHASE_PLAYING,
|
|
ScopaState,
|
|
)
|
|
|
|
ENGINES = (ScopaEngine, ScopaQuattroEngine)
|
|
|
|
|
|
def _session(engine, **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, **options) -> GameSession:
|
|
session = _session(engine, **options)
|
|
for seat in range(1, engine.max_players):
|
|
engine.join(session, f"p{seat}", f"P{seat}")
|
|
return session
|
|
|
|
|
|
def _drive_to_end(engine, session: GameSession) -> None:
|
|
"""Play the match out with the first legal move each turn."""
|
|
moves = 0
|
|
while not engine.is_finished(session) and moves < 200_000:
|
|
state = session.state
|
|
if state.phase == PHASE_HAND_END:
|
|
for player in state.players:
|
|
engine.handle_action(session, player.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, asso_pigliatutto=state.asso_pigliatutto
|
|
)
|
|
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
|
|
|
|
|
|
class CreateTest(unittest.TestCase):
|
|
def test_create_seats_creator_on_side_a(self) -> None:
|
|
for engine_type in ENGINES:
|
|
engine = engine_type()
|
|
session = _session(engine)
|
|
self.assertEqual("A", session.players[0].team)
|
|
self.assertIsInstance(session.state, ScopaState)
|
|
self.assertEqual(11, session.state.target_score)
|
|
self.assertFalse(session.state.asso_pigliatutto)
|
|
|
|
def test_create_options(self) -> None:
|
|
engine = ScopaEngine()
|
|
session = _session(engine, target_score=21, asso_pigliatutto=True)
|
|
self.assertEqual(21, session.state.target_score)
|
|
self.assertTrue(session.state.asso_pigliatutto)
|
|
|
|
def test_create_rejects_bad_options(self) -> None:
|
|
engine = ScopaEngine()
|
|
with self.assertRaises(IllegalMove):
|
|
_session(engine, target_score=0)
|
|
with self.assertRaises(IllegalMove):
|
|
_session(engine, target_score=101)
|
|
with self.assertRaises(IllegalMove):
|
|
_session(engine, target_score="eleven")
|
|
with self.assertRaises(IllegalMove):
|
|
_session(engine, target_score=True)
|
|
with self.assertRaises(IllegalMove):
|
|
_session(engine, asso_pigliatutto="yes")
|
|
|
|
def test_timeouts_come_from_the_engine(self) -> None:
|
|
engine = ScopaEngine(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_starts_two_player_game(self) -> None:
|
|
engine = ScopaEngine()
|
|
session = _session(engine)
|
|
self.assertTrue(engine.in_lobby(session))
|
|
engine.join(session, "bob", "Bob")
|
|
self.assertFalse(engine.in_lobby(session))
|
|
self.assertEqual(PHASE_PLAYING, session.state.phase)
|
|
self.assertEqual(["A", "B"], [p.team for p in session.players])
|
|
|
|
def test_join_starts_four_player_game_with_teams(self) -> None:
|
|
engine = ScopaQuattroEngine()
|
|
session = _session(engine)
|
|
for seat, team in ((1, "B"), (2, "A"), (3, "B")):
|
|
engine.join(session, f"p{seat}", f"P{seat}")
|
|
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:
|
|
engine = ScopaEngine()
|
|
session = _session(engine)
|
|
with self.assertRaises(AlreadyJoined):
|
|
engine.join(session, "alice", "Alice")
|
|
engine.join(session, "bob", "Bob")
|
|
with self.assertRaises(GameNotStarted):
|
|
engine.join(session, "carol", "Carol")
|
|
with self.assertRaises(GameNotStarted):
|
|
engine.join(session, "alice", "Alice")
|
|
|
|
|
|
class ActionTest(unittest.TestCase):
|
|
def test_unknown_action_rejected(self) -> None:
|
|
engine = ScopaEngine()
|
|
session = _started(engine)
|
|
with self.assertRaises(IllegalMove):
|
|
engine.handle_action(session, "alice", "dance", {})
|
|
|
|
def test_play_validates_payload(self) -> None:
|
|
engine = ScopaEngine()
|
|
session = _started(engine)
|
|
with self.assertRaises(IllegalMove):
|
|
engine.handle_action(session, "p1", "play", {"card": 42})
|
|
with self.assertRaises(IllegalMove):
|
|
engine.handle_action(session, "p1", "play", {})
|
|
with self.assertRaises(IllegalMove):
|
|
engine.handle_action(
|
|
session, "p1", "play", {"card": "01D", "capture": "02C"}
|
|
)
|
|
|
|
def test_invalid_card_code_is_illegal_move(self) -> None:
|
|
engine = ScopaEngine()
|
|
session = _started(engine)
|
|
with self.assertRaises(IllegalMove):
|
|
engine.handle_action(session, "p1", "play", {"card": "nope"})
|
|
|
|
def test_finished_match_rejects_actions(self) -> None:
|
|
engine = ScopaEngine()
|
|
session = _started(engine, target_score=1)
|
|
_drive_to_end(engine, session)
|
|
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_hides_other_hands_and_reports_the_deck(self) -> None:
|
|
engine = ScopaEngine()
|
|
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])
|
|
self.assertEqual(2, view["player_count"])
|
|
self.assertEqual(30, view["deck_count"])
|
|
self.assertEqual(4, len(view["table"]))
|
|
# 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_view_for_spectator_hides_all_hands(self) -> None:
|
|
engine = ScopaQuattroEngine()
|
|
session = _started(engine)
|
|
view = engine.view_for(session, None)
|
|
for player in view["players"]:
|
|
self.assertNotIn("hand", player)
|
|
self.assertNotIn("your_turn", view)
|
|
self.assertNotIn("legal_moves", view)
|
|
self.assertIn("table", view)
|
|
self.assertEqual(4, view["player_count"])
|
|
|
|
def test_lobby_view(self) -> None:
|
|
engine = ScopaEngine()
|
|
session = _session(engine, target_score=16, asso_pigliatutto=True)
|
|
lobby = engine.lobby_view(session)
|
|
self.assertEqual("lobby", lobby["phase"])
|
|
self.assertEqual(16, lobby["target_score"])
|
|
self.assertTrue(lobby["asso_pigliatutto"])
|
|
|
|
|
|
class SerializationTest(unittest.TestCase):
|
|
def test_state_roundtrip(self) -> None:
|
|
engine = ScopaQuattroEngine()
|
|
session = _started(engine)
|
|
restored = engine.state_from_json(engine.state_to_json(session.state))
|
|
self.assertIsInstance(restored, ScopaState)
|
|
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],
|
|
)
|
|
self.assertEqual(
|
|
[c.code for c in session.state.deck],
|
|
[c.code for c in restored.deck],
|
|
)
|
|
|
|
|
|
class DeadlineTest(unittest.TestCase):
|
|
def test_no_deadline_in_lobby(self) -> None:
|
|
engine = ScopaEngine()
|
|
self.assertIsNone(engine.next_deadline(_session(engine)))
|
|
|
|
def test_turn_deadline_and_revalidation(self) -> None:
|
|
engine = ScopaEngine()
|
|
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:
|
|
engine = ScopaEngine()
|
|
session = _started(engine)
|
|
deadline = engine.next_deadline(session)
|
|
assert deadline is not None
|
|
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 = ScopaEngine()
|
|
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_two_player_match(self) -> None:
|
|
engine = ScopaEngine()
|
|
session = _started(engine, target_score=1)
|
|
_drive_to_end(engine, session)
|
|
result = engine.result(session)
|
|
self.assertEqual(2, len(result.teams))
|
|
self.assertEqual(1, len(result.teams[0]))
|
|
self.assertEqual(1, len(result.teams[1]))
|
|
self.assertIn(result.winner_team, (0, 1))
|
|
self.assertEqual(2, 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)
|
|
self.assertIn("asso_pigliatutto", result.summary)
|
|
|
|
def test_result_of_finished_four_player_match(self) -> None:
|
|
engine = ScopaQuattroEngine()
|
|
session = _started(engine, target_score=1)
|
|
_drive_to_end(engine, session)
|
|
result = engine.result(session)
|
|
self.assertEqual([2, 2], [len(team) for team in result.teams])
|
|
self.assertEqual(4, len(result.players))
|
|
|
|
def test_result_requires_finished_match(self) -> None:
|
|
engine = ScopaEngine()
|
|
with self.assertRaises(GameError):
|
|
engine.result(_started(engine))
|
|
|
|
def test_game_over_view(self) -> None:
|
|
engine = ScopaEngine()
|
|
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:
|
|
two = ScopaEngine()
|
|
self.assertEqual("scopa", two.id)
|
|
self.assertEqual("Scopa", two.name)
|
|
self.assertEqual(2, two.min_players)
|
|
self.assertEqual(2, two.max_players)
|
|
four = ScopaQuattroEngine()
|
|
self.assertEqual("scopa_a_quattro", four.id)
|
|
self.assertEqual("Scopa a quattro", four.name)
|
|
self.assertEqual(4, four.min_players)
|
|
self.assertEqual(4, four.max_players)
|
|
for engine in (two, four):
|
|
self.assertIn("target_score", engine.options_schema["properties"])
|
|
self.assertIn(
|
|
"asso_pigliatutto", engine.options_schema["properties"]
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|