"""Scopone result-persistence tests: engine result to Postgres to API. The platform suite covers the generic machinery against a toy game; these tests pin the scopone-specific shape: the match ``result`` summary, per-player scores/teams, Elo deltas and what the history and leaderboard endpoints expose for a finished scopone match. """ from __future__ import annotations import unittest from httpx import ASGITransport, AsyncClient from tavolo.app import app, platform, tortoise_mixin from tavolo.briscola.chiamata import PHASE_AUCTION, PHASE_CALLING from tavolo.briscola.state import Card from tavolo.platform import GameSession, Seat from tavolo.platform.elo import INITIAL_RATING from tavolo.platform.errors import IllegalMove from tavolo.platform.models import Match, MatchPlayer, PlayerRating from tavolo.platform.stats import save_match_result from tavolo.scopa import engine as scopa_rules from tavolo.scopone import engine as rules from tests.helpers import async_test, oidc_user PLAYERS = ("alice", "bob", "carol", "dave") async def _finished_session(target_score: int = 1) -> GameSession: engine = platform.registry.require("scopone_scientifico") session = GameSession( id="stats-scope-1", game_type=engine.id, join_code="SS0001", creator_sub="alice", players=[Seat(user_sub="alice", display_name="Alice")], ) engine.create(session, {"target_score": target_score}) for name in PLAYERS[1:]: engine.join(session, name, name.capitalize()) moves = 0 while not engine.is_finished(session) and moves < 200000: state = session.state if state.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) 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 assert engine.is_finished(session) return session class ScoponeStatsTest(unittest.TestCase): @async_test async def test_finished_match_persisted_with_scopone_summary(self) -> None: await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None engine = platform.registry.require("scopone_scientifico") session = await _finished_session() with ctx: await save_match_result(session, engine) self.assertEqual(1, await Match.all().count()) self.assertEqual(4, await MatchPlayer.all().count()) match = await Match.all().first() assert match is not None self.assertEqual("scopone_scientifico", match.game_type) summary = match.result self.assertEqual(1, summary["target_score"]) self.assertIn("team_a_score", summary) self.assertIn("team_b_score", summary) self.assertIn("winner_team", summary) self.assertIn("hands_played", summary) self.assertIn("hand_scores", summary) players = {p.user_sub: p for p in await MatchPlayer.all()} winner_team = summary["winner_team"] for sub, row in players.items(): if row.won: self.assertEqual(winner_team, row.team) else: self.assertNotEqual(winner_team, row.team) self.assertEqual( summary["team_a_score"] if row.team == "A" else summary["team_b_score"], row.score, ) # Winners share a team, losers the other. winners = {sub for sub, row in players.items() if row.won} self.assertEqual(2, len(winners)) @async_test async def test_finished_match_updates_elo(self) -> None: await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None engine = platform.registry.require("scopone_scientifico") session = await _finished_session() with ctx: await save_match_result(session, engine) ratings = {r.user_sub: r.rating for r in await PlayerRating.all()} self.assertEqual(4, len(ratings)) self.assertEqual( {INITIAL_RATING + 16, INITIAL_RATING - 16}, set(ratings.values()) ) @async_test async def test_history_and_leaderboard_expose_scopone_result(self) -> None: await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None engine = platform.registry.require("scopone_scientifico") session = await _finished_session() with ctx: await save_match_result(session, engine) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user("alice"): history = await client.get("/api/me/matches") self.assertEqual(200, history.status_code) results = history.json()["results"] self.assertEqual(1, len(results)) self.assertEqual("scopone_scientifico", results[0]["game_type"]) self.assertIn("team_a_score", results[0]["result"]) self.assertIn("your_elo_delta", results[0]) self.assertEqual(4, len(results[0]["players"])) with oidc_user("alice"): board = await client.get("/api/leaderboard") self.assertEqual(200, board.status_code) body = board.json() by_sub = {r["user_sub"]: r for r in body["results"]} self.assertEqual(4, len(by_sub)) self.assertTrue(all(r["matches"] == 1 for r in by_sub.values())) self.assertEqual(1, body["page"]) self.assertEqual(1, body["pages"]) self.assertEqual(4, body["total"]) # The caller's own rank rides along on the response. self.assertIsNotNone(body["your_rank"]) self.assertEqual(1, body["your_page"]) paged = await client.get("/api/leaderboard?limit=1&page=2") self.assertEqual(200, paged.status_code) self.assertEqual(4, paged.json()["pages"]) self.assertEqual(2, paged.json()["page"]) self.assertEqual(1, len(paged.json()["results"])) async def _finished_briscola_session() -> GameSession: engine = platform.registry.require("briscola") session = GameSession( id="stats-briscola-1", game_type=engine.id, join_code="BS0001", creator_sub="alice", players=[Seat(user_sub="alice", display_name="Alice")], ) engine.create(session, {}) engine.join(session, "bob", "Bob") moves = 0 while not engine.is_finished(session) and moves < 200000: state = session.state if state.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) engine.handle_action( session, player.sub, "play", {"card": player.hand[0].code} ) moves += 1 assert engine.is_finished(session) return session class BriscolaStatsTest(unittest.TestCase): @async_test async def test_finished_briscola_match_persisted_with_summary(self) -> None: await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None engine = platform.registry.require("briscola") session = await _finished_briscola_session() with ctx: await save_match_result(session, engine) self.assertEqual(1, await Match.all().count()) self.assertEqual(2, await MatchPlayer.all().count()) match = await Match.all().first() assert match is not None self.assertEqual("briscola", match.game_type) summary = match.result self.assertIn("team_a_score", summary) self.assertIn("team_b_score", summary) self.assertIn("winner_team", summary) self.assertIn("target_score", summary) self.assertIn("deals_played", summary) self.assertIn("deal_scores", summary) winner_team = summary["winner_team"] players = {p.user_sub: p for p in await MatchPlayer.all()} for row in players.values(): if row.won: self.assertEqual(winner_team, row.team) else: self.assertNotEqual(winner_team, row.team) # A two-player match moves Elo by 16 points each. ratings = {r.user_sub: r.rating for r in await PlayerRating.all()} self.assertEqual(2, len(ratings)) self.assertEqual( {INITIAL_RATING + 16, INITIAL_RATING - 16}, set(ratings.values()) ) @async_test async def test_history_exposes_briscola_result(self) -> None: await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None engine = platform.registry.require("briscola") session = await _finished_briscola_session() with ctx: await save_match_result(session, engine) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user("alice"): history = await client.get("/api/me/matches") self.assertEqual(200, history.status_code) results = history.json()["results"] self.assertEqual(1, len(results)) self.assertEqual("briscola", results[0]["game_type"]) self.assertIn("team_a_score", results[0]["result"]) self.assertEqual(2, len(results[0]["players"])) async def _finished_chiamata_session() -> GameSession: engine = platform.registry.require("briscola_chiamata") session = GameSession( id="stats-chiamata-1", game_type=engine.id, join_code="BC0001", creator_sub="alice", players=[ Seat(user_sub=name, display_name=name.capitalize()) for name in ("alice", "bob", "carol", "dave", "erin") ], ) engine.create(session, {}) for name in ("bob", "carol", "dave", "erin"): engine.join(session, name, name.capitalize()) moves = 0 while not engine.is_finished(session) and moves < 200000: state = session.state if state.phase == PHASE_AUCTION: player = next(p for p in state.players if p.seat == state.turn) try: engine.handle_action(session, player.sub, "pass", {}) except IllegalMove: engine.handle_action(session, player.sub, "bid", {"rank": 2}) elif state.phase == PHASE_CALLING: assert state.auction is not None and state.auction.current is not None caller = next(p for p in state.players if p.seat == state.caller) card = Card(rank=state.auction.current.rank, suit="D") engine.handle_action(session, caller.sub, "call", {"card": card.code}) else: player = next(p for p in state.players if p.seat == state.turn) engine.handle_action( session, player.sub, "play", {"card": player.hand[0].code} ) moves += 1 assert engine.is_finished(session) return session class ChiamataStatsTest(unittest.TestCase): @async_test async def test_finished_chiamata_match_persisted_with_summary(self) -> None: await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None engine = platform.registry.require("briscola_chiamata") session = await _finished_chiamata_session() with ctx: await save_match_result(session, engine) self.assertEqual(1, await Match.all().count()) self.assertEqual(5, await MatchPlayer.all().count()) match = await Match.all().first() assert match is not None self.assertEqual("briscola_chiamata", match.game_type) summary = match.result for key in ( "team_a_score", "team_b_score", "winner_team", "declared_target", "called_card", "caller", "solo", "deals_played", "deal_scores", ): self.assertIn(key, summary) self.assertIn(summary["winner_team"], ("Chiamante", "Difensori")) players = await MatchPlayer.all() roles = {p.team for p in players} self.assertTrue(roles.issubset({"Chiamante", "Socio", "Difensori"})) # Roles match the sides: exactly one caller, the rest defenders # plus at most one partner. self.assertEqual(1, sum(1 for p in players if p.team == "Chiamante")) winner_team = summary["winner_team"] for row in players: on_caller_side = row.team in ("Chiamante", "Socio") self.assertEqual(row.won, on_caller_side == (winner_team == "Chiamante")) # Mean-team Elo: 2v3 or 1v4 moves every winner by +16. ratings = {r.user_sub: r.rating for r in await PlayerRating.all()} self.assertEqual(5, len(ratings)) winners = {p.user_sub for p in players if p.won} self.assertEqual( {INITIAL_RATING + 16}, {ratings[sub] for sub in winners} ) self.assertEqual( {INITIAL_RATING - 16}, {ratings[sub] for sub in ratings if sub not in winners}, ) class ScopaStatsTest(unittest.TestCase): async def _finished_session( self, player_count: int = 2, asso_pigliatutto: bool = False ) -> GameSession: engine = platform.registry.require( "scopa" if player_count == 2 else "scopa_a_quattro" ) session = GameSession( id=f"stats-scopa-{player_count}", game_type=engine.id, join_code="SC0001", creator_sub="alice", players=[Seat(user_sub="alice", display_name="Alice")], ) engine.create( session, {"target_score": 1, "asso_pigliatutto": asso_pigliatutto}, ) for name in PLAYERS[1:player_count]: engine.join(session, name, name.capitalize()) moves = 0 while not engine.is_finished(session) and moves < 200000: state = session.state if state.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 = scopa_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 assert engine.is_finished(session) return session @async_test async def test_finished_two_player_match_persisted_with_summary(self) -> None: await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None engine = platform.registry.require("scopa") session = await self._finished_session(2, asso_pigliatutto=True) with ctx: await save_match_result(session, engine) self.assertEqual(1, await Match.all().count()) self.assertEqual(2, await MatchPlayer.all().count()) match = await Match.all().first() assert match is not None self.assertEqual("scopa", match.game_type) summary = match.result for key in ( "team_a_score", "team_b_score", "winner_team", "target_score", "asso_pigliatutto", "hands_played", "hand_scores", ): self.assertIn(key, summary) self.assertTrue(summary["asso_pigliatutto"]) winner_team = summary["winner_team"] players = {p.user_sub: p for p in await MatchPlayer.all()} for row in players.values(): self.assertEqual(row.won, row.team == winner_team) # A two-player match moves Elo by 16 points each. ratings = {r.user_sub: r.rating for r in await PlayerRating.all()} self.assertEqual( {INITIAL_RATING + 16, INITIAL_RATING - 16}, set(ratings.values()) ) @async_test async def test_finished_four_player_match_persisted_with_teams(self) -> None: await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None engine = platform.registry.require("scopa_a_quattro") session = await self._finished_session(4) with ctx: await save_match_result(session, engine) self.assertEqual(4, await MatchPlayer.all().count()) match = await Match.all().first() assert match is not None self.assertEqual("scopa_a_quattro", match.game_type) players = await MatchPlayer.all() self.assertEqual({"A", "B"}, {p.team for p in players}) # Partnership Elo: the winning side gains 16 each. ratings = {r.user_sub: r.rating for r in await PlayerRating.all()} winners = {p.user_sub for p in players if p.won} self.assertEqual( {INITIAL_RATING + 16}, {ratings[sub] for sub in winners} ) self.assertEqual( {INITIAL_RATING - 16}, {ratings[sub] for sub in ratings if sub not in winners}, ) @async_test async def test_history_exposes_scopa_result(self) -> None: await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None engine = platform.registry.require("scopa") session = await self._finished_session(2) with ctx: await save_match_result(session, engine) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user("alice"): history = await client.get("/api/me/matches") self.assertEqual(200, history.status_code) results = history.json()["results"] self.assertEqual(1, len(results)) self.assertEqual("scopa", results[0]["game_type"]) self.assertIn("team_a_score", results[0]["result"]) self.assertIn("your_elo_delta", results[0]) self.assertEqual(2, len(results[0]["players"])) if __name__ == "__main__": unittest.main()