woggioni
b5219062f2
CI / Build and push docker image (push) Successful in 3m0s
Anonymous users and logged-in non-players can now open /game/<id> and
follow the match: they see the table, the score and each player's
captured cards and scope, but never a hand and never the join code.
Spectator actions are rejected with a forbidden error.
Backend (tavolo-platform / tavolo-scopone):
- /ws/games/{id} accepts anonymous and non-seated connections as
spectators; only unknown games are refused. The state payload carries
spectator: true and withholds join_code unless the viewer is seated.
- GET /api/games/{id} is public and returns the same personalized
(player) or read-only (spectator) view.
- GameEngine.view_for/game_over_view take Optional[str] (None = spectator)
and are documented to hide privileged information for non-seated
viewers.
Frontend (Sycamore):
- Parse the spectator flag and render a spectator table: card backs plus
captured/scope for every seat, no clickable hand, "Spectating" badge.
- A pre-start spectator sees a "game not started yet" waiting room with
the watch link and no join code.
- "Copy watch link" buttons in the waiting room and the game HUD.
- The hand-end summary carries no acknowledge button for spectators and
the game-over overlay hides "My matches".
Tests and docs updated.
311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""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)
|
|
self.assertFalse(session.state.asso_pigliatutto)
|
|
|
|
def test_create_asso_pigliatutto_option(self) -> None:
|
|
engine = ScoponeEngine()
|
|
session = _session(engine, asso_pigliatutto=True)
|
|
self.assertTrue(session.state.asso_pigliatutto)
|
|
|
|
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")
|
|
with self.assertRaises(IllegalMove):
|
|
_session(engine, asso_pigliatutto="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_view_for_spectator_hides_all_hands(self) -> None:
|
|
engine = ScoponeEngine()
|
|
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({"A": 0, "B": 0}, view["scores"])
|
|
|
|
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"])
|
|
self.assertFalse(lobby["asso_pigliatutto"])
|
|
|
|
|
|
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"])
|
|
self.assertIn("asso_pigliatutto", engine.options_schema["properties"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|