"""Match-statistics tests: Postgres persistence and the stats endpoints.""" from __future__ import annotations import unittest import uuid from datetime import datetime, timezone from httpx import ASGITransport, AsyncClient from pwo import async_test from tavolo.app import app, tortoise_mixin from tavolo.game import engine from tavolo.game.state import GameState from tavolo.models import Match, MatchPlayer from tavolo.stats import save_match_result from tests.helpers import oidc_user async def _use_app_db(): """Bind the same Tortoise context the app uses for this event loop and return it, so tests can seed rows the route handlers will see.""" await tortoise_mixin._bind() ctx = tortoise_mixin._ctx assert ctx is not None return ctx def _finished_state() -> GameState: # Team A sweeps the (single-card) table with carte + denara and reaches # a target of 2, ending the match. state = GameState( id="stats-game", join_code="STATS1", creator_sub="alice", target_score=2, phase=engine.PHASE_PLAYING, turn=0, table=[engine.parse_card("02C")], ) from tavolo.game.state import PlayerState, Card state.players = [ PlayerState(sub="alice", name="alice", seat=0, hand=[Card.parse("02D")]), PlayerState(sub="bob", name="bob", seat=1), PlayerState(sub="carol", name="carol", seat=2), PlayerState(sub="dave", name="dave", seat=3), ] return state class SaveMatchResultTest(unittest.TestCase): @async_test async def test_finished_match_is_persisted_once(self) -> None: ctx = await _use_app_db() state = _finished_state() engine.play(state, "alice", "02D", ["02C"]) self.assertEqual(engine.PHASE_FINISHED, state.phase) with ctx: await save_match_result(state) await save_match_result(state) # idempotent 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(state.scores[0], match.team_a_score) self.assertEqual("A", match.winner_team) # The game type travels from the live state onto the row. self.assertEqual("scopone_scientifico", match.game_type) winners = await MatchPlayer.filter(won=True) self.assertEqual({"alice", "carol"}, {p.user_sub for p in winners}) async def _seed_two_matches(game_types: tuple = ("scopone_scientifico", "scopone_scientifico")) -> None: ctx = await _use_app_db() with ctx: for index, (a_score, b_score, winner, finished) in enumerate( [ (11, 5, "A", datetime(2026, 1, 1, 10, tzinfo=timezone.utc)), (8, 11, "B", datetime(2026, 1, 2, 10, tzinfo=timezone.utc)), ] ): match = await Match.create( id=uuid.uuid4(), game_type=game_types[index], team_a_score=a_score, team_b_score=b_score, winner_team=winner, target_score=11, hands_played=2 + index, started_at=datetime(2025, 12, 31, tzinfo=timezone.utc), finished_at=finished, ) seats = [ ("alice", 0, "A"), ("bob", 1, "B"), ("carol", 2, "A"), ("dave", 3, "B"), ] for sub, seat, team in seats: await MatchPlayer.create( id=uuid.uuid4(), match=match, user_sub=sub, display_name=sub, seat=seat, team=team, won=(team == winner), ) class StatsRouteTest(unittest.TestCase): @async_test async def test_my_matches_newest_first(self) -> None: await _seed_two_matches() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user("alice"): response = await client.get("/api/me/matches") self.assertEqual(200, response.status_code) results = response.json()["results"] self.assertEqual(2, len(results)) self.assertEqual("B", results[0]["winner_team"]) # newest first self.assertFalse(results[0]["you_won"]) self.assertTrue(results[1]["you_won"]) self.assertEqual(4, len(results[0]["players"])) self.assertIn("next_cursor", response.json()) @async_test async def test_my_matches_pagination(self) -> None: await _seed_two_matches() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user("alice"): first = await client.get("/api/me/matches?limit=1") cursor = first.json()["next_cursor"] self.assertIsNotNone(cursor) second = await client.get(f"/api/me/matches?limit=1&cursor={cursor}") self.assertEqual(1, len(first.json()["results"])) self.assertEqual(1, len(second.json()["results"])) self.assertNotEqual( first.json()["results"][0]["id"], second.json()["results"][0]["id"], ) @async_test async def test_my_matches_requires_auth(self) -> None: transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: response = await client.get("/api/me/matches") self.assertEqual(401, response.status_code) @async_test async def test_leaderboard_aggregates(self) -> None: await _seed_two_matches() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: response = await client.get("/api/leaderboard") self.assertEqual(200, response.status_code) by_sub = {row["user_sub"]: row for row in response.json()["results"]} self.assertEqual(2, by_sub["alice"]["matches"]) self.assertEqual(1, by_sub["alice"]["wins"]) # team A won match 1 self.assertEqual(19, by_sub["alice"]["points"]) self.assertEqual(1, by_sub["bob"]["wins"]) # team B won match 2 self.assertEqual(16, by_sub["bob"]["points"]) # Alice leads on points after tying Bob on wins. self.assertEqual("alice", response.json()["results"][0]["user_sub"]) class GameTypeFilterTest(unittest.TestCase): """Stats endpoints scope results by the match's game type.""" @async_test async def test_my_matches_filter_by_game_type(self) -> None: # The second seed names a game the registry does not know; rows are # written directly, so this only exercises the SQL filter. await _seed_two_matches(game_types=("scopone_scientifico", "other_game")) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: with oidc_user("alice"): all_matches = await client.get("/api/me/matches") scoped = await client.get("/api/me/matches?game_type=scopone_scientifico") unknown = await client.get("/api/me/matches?game_type=briscola") self.assertEqual(2, len(all_matches.json()["results"])) self.assertEqual( {"scopone_scientifico", "other_game"}, {m["game_type"] for m in all_matches.json()["results"]}, ) scoped_results = scoped.json()["results"] self.assertEqual(1, len(scoped_results)) self.assertEqual("scopone_scientifico", scoped_results[0]["game_type"]) self.assertEqual(400, unknown.status_code) @async_test async def test_leaderboard_filter_by_game_type(self) -> None: await _seed_two_matches(game_types=("scopone_scientifico", "other_game")) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client: scoped = await client.get("/api/leaderboard?game_type=scopone_scientifico") unknown = await client.get("/api/leaderboard?game_type=briscola") self.assertEqual(200, scoped.status_code) by_sub = {row["user_sub"]: row for row in scoped.json()["results"]} # Only the first match counts: one match per player, team A won. self.assertEqual(1, by_sub["alice"]["matches"]) self.assertEqual(1, by_sub["alice"]["wins"]) self.assertEqual(0, by_sub["bob"]["wins"]) self.assertEqual(400, unknown.status_code) if __name__ == "__main__": unittest.main()