Files
tavolo/server/packages/tavolo-platform/tests/test_stats.py
T
woggioni 1ab748a34f
CI / Build and push docker image (push) Successful in 1m19s
Paginate the leaderboard from materialized player ratings
Store each player's leaderboard totals (wins, accumulated points and the
latest display name) on player_rating alongside the Elo rating, updated
in the same transaction as the match, so the board is a single indexed
table query instead of aggregating every participation on each request.

GET /api/leaderboard is now page-number paginated (page/limit) and
returns the page, page count and total, plus the caller's own rank and
page when authenticated; focus=me returns the caller's page directly.
The migration adds the columns, backfills them from the match history
(inserting rows for players from pre-Elo matches at the initial rating)
and replaces the index. backfill_elo rebuilds the aggregates too.

The web page opens on the page containing the signed-in player, marks
their row and adds a numbered footer pager.
2026-10-09 16:42:47 +08:00

527 lines
23 KiB
Python

"""Match-statistics tests: Postgres persistence and the stats endpoints."""
from __future__ import annotations
import unittest
import uuid
from datetime import datetime, timezone
from typing import Any, Dict
from unittest import mock
from httpx import ASGITransport, AsyncClient
from tavolo.platform.elo import INITIAL_RATING
from tavolo.platform.models import Match, MatchPlayer, PlayerRating
from tavolo.platform.stats import save_match_result
from helpers import DummyEngine, async_test, make_platform, oidc_user, use_db
def _finished_session(target: int = 2):
"""A started dummy session one play short of completion."""
from tavolo.platform import GameSession, Seat
engine = DummyEngine()
session = GameSession(
id="stats-game",
game_type=engine.id,
join_code="STATS1",
creator_sub="alice",
players=[Seat(user_sub="alice", display_name="alice", team="A")],
)
engine.create(session, {"target": target})
engine.join(session, "bob", "bob")
engine.handle_action(session, "alice", "play", {})
return engine, session
class SaveMatchResultTest(unittest.TestCase):
@async_test
async def test_finished_match_is_persisted_once(self) -> None:
_, _, tortoise_mixin = make_platform()
ctx = await use_db(tortoise_mixin)
engine, session = _finished_session()
engine.handle_action(session, "alice", "play", {})
self.assertTrue(engine.is_finished(session))
with ctx:
await save_match_result(session, engine)
await save_match_result(session, engine) # idempotent
self.assertEqual(1, await Match.all().count())
self.assertEqual(2, await MatchPlayer.all().count())
match = await Match.all().first()
assert match is not None
# The game type travels from the session onto the row; the
# engine's summary is stored verbatim as the result.
self.assertEqual("dummy", match.game_type)
self.assertEqual("alice", match.result["winner"])
self.assertEqual(2, match.result["plays"])
winners = await MatchPlayer.filter(won=True)
self.assertEqual({"alice"}, {p.user_sub for p in winners})
scores = {p.user_sub: p.score for p in await MatchPlayer.all()}
self.assertEqual({"alice": 2.0, "bob": 0.0}, scores)
@async_test
async def test_finished_match_updates_elo_ratings(self) -> None:
_, _, tortoise_mixin = make_platform()
ctx = await use_db(tortoise_mixin)
engine, session = _finished_session()
engine.handle_action(session, "alice", "play", {})
with ctx:
await save_match_result(session, engine)
ratings = {
row.user_sub: row for row in await PlayerRating.all()
}
self.assertEqual(2, len(ratings))
# Two players at 1500: winner gains K/2, loser loses it.
self.assertEqual(INITIAL_RATING + 16, ratings["alice"].rating)
self.assertEqual(1, ratings["alice"].matches_played)
self.assertEqual(INITIAL_RATING - 16, ratings["bob"].rating)
self.assertEqual(1, ratings["bob"].matches_played)
# The leaderboard aggregates ride along in the same transaction.
self.assertEqual(1, ratings["alice"].wins)
self.assertEqual(0, ratings["bob"].wins)
self.assertEqual(2.0, ratings["alice"].points)
self.assertEqual(0.0, ratings["bob"].points)
self.assertEqual("alice", ratings["alice"].display_name)
# The per-match delta is recorded on each participation row.
deltas = {
p.user_sub: p.elo_delta for p in await MatchPlayer.all()
}
self.assertEqual({"alice": 16, "bob": -16}, deltas)
@async_test
async def test_elo_ratings_accumulate_across_matches(self) -> None:
_, _, tortoise_mixin = make_platform()
ctx = await use_db(tortoise_mixin)
engine, session = _finished_session()
engine.handle_action(session, "alice", "play", {})
from tavolo.platform import GameSession, Seat
engine2, session2 = DummyEngine(), GameSession(
id="stats-game-2",
game_type="dummy",
join_code="STATS2",
creator_sub="alice",
players=[Seat(user_sub="alice", display_name="alice", team="A")],
)
engine2.create(session2, {"target": 2})
engine2.join(session2, "bob", "bob")
# Bob wins the second match.
engine2.handle_action(session2, "bob", "play", {})
engine2.handle_action(session2, "bob", "play", {})
with ctx:
await save_match_result(session, engine)
await save_match_result(session2, engine2)
ratings = {
row.user_sub: row.rating for row in await PlayerRating.all()
}
# Match 1: even teams, alice wins (+16/-16). Match 2: alice
# is now the favourite (1516 vs 1484), so losing costs 17.
self.assertEqual(INITIAL_RATING - 1, ratings["alice"])
self.assertEqual(INITIAL_RATING + 1, ratings["bob"])
bob = await PlayerRating.get(user_sub="bob")
self.assertEqual(2, bob.matches_played)
@async_test
async def test_unfinished_match_is_not_persisted(self) -> None:
_, _, tortoise_mixin = make_platform()
ctx = await use_db(tortoise_mixin)
engine, session = _finished_session()
with ctx:
await save_match_result(session, engine)
self.assertEqual(0, await Match.all().count())
async def _seed_two_matches(
tortoise_mixin, game_types: tuple = ("dummy", "dummy")
) -> None:
ctx = await use_db(tortoise_mixin)
with ctx:
# The leaderboard reads the player_rating aggregates, so the seeded
# participations are mirrored there (as save_match_result would).
aggregates: Dict[tuple, Dict[str, Any]] = {}
for index, (winner, finished) in enumerate(
[
("alice", datetime(2026, 1, 1, 10, tzinfo=timezone.utc)),
("bob", datetime(2026, 1, 2, 10, tzinfo=timezone.utc)),
]
):
match = await Match.create(
id=uuid.uuid4(),
game_type=game_types[index],
started_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
finished_at=finished,
result={"winner": winner, "plays": 3 + index},
)
for seat, sub in enumerate(("alice", "bob")):
score = 2.0 + index if sub == winner else 1.0
await MatchPlayer.create(
id=uuid.uuid4(),
match=match,
user_sub=sub,
display_name=sub,
seat=seat,
team="A" if seat == 0 else "B",
won=(sub == winner),
score=score,
)
aggregate = aggregates.setdefault(
(sub, game_types[index]),
{"matches": 0, "wins": 0, "points": 0.0},
)
aggregate["matches"] += 1
aggregate["wins"] += 1 if sub == winner else 0
aggregate["points"] += score
for (sub, game_type), aggregate in aggregates.items():
await PlayerRating.create(
id=uuid.uuid4(),
user_sub=sub,
game_type=game_type,
rating=INITIAL_RATING,
matches_played=aggregate["matches"],
wins=aggregate["wins"],
points=aggregate["points"],
display_name=sub,
)
class StatsRouteTest(unittest.TestCase):
@async_test
async def test_my_matches_newest_first(self) -> None:
app, platform, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
with oidc_user(platform.oidc, "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("bob", results[0]["result"]["winner"]) # newest first
self.assertFalse(results[0]["you_won"])
self.assertTrue(results[1]["you_won"])
self.assertEqual(2, len(results[0]["players"]))
self.assertIn("next_cursor", response.json())
@async_test
async def test_my_matches_pagination(self) -> None:
app, platform, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
with oidc_user(platform.oidc, "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_fetches_participants_in_one_query(self) -> None:
app, platform, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin)
transport = ASGITransport(app=app)
with mock.patch.object(
MatchPlayer, "filter", wraps=MatchPlayer.filter
) as spy:
async with AsyncClient(
transport=transport, base_url="http://127.0.0.1"
) as client:
with oidc_user(platform.oidc, "alice"):
response = await client.get("/api/me/matches")
self.assertEqual(200, response.status_code)
# One query for all matches, not one per match row.
self.assertEqual(1, spy.call_count)
results = response.json()["results"]
self.assertEqual(2, len(results))
for result in results:
seats = [player["seat"] for player in result["players"]]
self.assertEqual([0, 1], seats)
@async_test
async def test_my_matches_requires_auth(self) -> None:
app, _, _ = make_platform()
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:
app, _, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin)
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"]) # alice won match 1
self.assertEqual(3.0, by_sub["alice"]["points"]) # 2.0 + 1.0
self.assertEqual(1, by_sub["bob"]["wins"]) # bob won match 2
self.assertEqual(4.0, by_sub["bob"]["points"]) # 1.0 + 3.0
# Bob leads on points after tying Alice on wins.
self.assertEqual("bob", response.json()["results"][0]["user_sub"])
@async_test
async def test_leaderboard_includes_elo_and_sorts_by_it(self) -> None:
app, _, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin)
ctx = await use_db(tortoise_mixin)
with ctx:
# Alice outranks everyone despite Bob leading on points.
await PlayerRating.filter(
user_sub="alice", game_type="dummy"
).update(rating=1600)
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)
results = response.json()["results"]
by_sub = {row["user_sub"]: row for row in results}
self.assertEqual(1600, by_sub["alice"]["elo"])
self.assertEqual(INITIAL_RATING, by_sub["bob"]["elo"])
# Elo outranks wins/points.
self.assertEqual("alice", results[0]["user_sub"])
@async_test
async def test_leaderboard_elo_scoped_by_game_type(self) -> None:
app, platform, tortoise_mixin = make_platform(
engines=(DummyEngine(), SecondEngine())
)
await _seed_two_matches(tortoise_mixin)
ctx = await use_db(tortoise_mixin)
with ctx:
await PlayerRating.filter(
user_sub="bob", game_type="dummy"
).update(rating=1516)
# Bob's rating in another game must not leak into the
# dummy leaderboard.
await PlayerRating.create(
id=uuid.uuid4(),
user_sub="bob",
game_type="second",
rating=1800,
matches_played=1,
)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/api/leaderboard?game_type=dummy")
self.assertEqual(200, response.status_code)
results = response.json()["results"]
by_sub = {row["user_sub"]: row for row in results}
self.assertEqual(1516, by_sub["bob"]["elo"])
self.assertEqual(INITIAL_RATING, by_sub["alice"]["elo"])
self.assertEqual("second", platform.registry.all()[1].id)
@async_test
async def test_leaderboard_pagination(self) -> None:
app, _, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin)
ctx = await use_db(tortoise_mixin)
with ctx:
for index, sub in enumerate(("carol", "dave", "erin", "frank")):
await PlayerRating.create(
id=uuid.uuid4(),
user_sub=sub,
game_type="dummy",
rating=1600 + index,
matches_played=1,
wins=1,
points=2.0,
display_name=sub,
)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
first = await client.get("/api/leaderboard?limit=2")
second = await client.get("/api/leaderboard?limit=2&page=2")
third = await client.get("/api/leaderboard?limit=2&page=3")
beyond = await client.get("/api/leaderboard?limit=2&page=99")
body = first.json()
self.assertEqual(1, body["page"])
self.assertEqual(3, body["pages"])
self.assertEqual(6, body["total"])
self.assertEqual(2, len(body["results"]))
self.assertEqual(
["frank", "erin"], [row["user_sub"] for row in body["results"]]
)
self.assertEqual(
["dave", "carol"],
[row["user_sub"] for row in second.json()["results"]],
)
self.assertEqual(
["bob", "alice"],
[row["user_sub"] for row in third.json()["results"]],
)
# Out-of-range pages clamp to the last page.
self.assertEqual(3, beyond.json()["page"])
self.assertEqual(
["bob", "alice"],
[row["user_sub"] for row in beyond.json()["results"]],
)
@async_test
async def test_leaderboard_focus_me(self) -> None:
app, platform, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin)
ctx = await use_db(tortoise_mixin)
with ctx:
for index, sub in enumerate(("carol", "dave")):
await PlayerRating.create(
id=uuid.uuid4(),
user_sub=sub,
game_type="dummy",
rating=1600 + index,
matches_played=1,
wins=1,
points=2.0,
display_name=sub,
)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
with oidc_user(platform.oidc, "alice"):
mine = await client.get("/api/leaderboard?limit=1&focus=me")
anonymous = await client.get("/api/leaderboard?limit=1&focus=me")
# dave, carol, bob, alice: the caller lands on their own page.
body = mine.json()
self.assertEqual(4, body["page"])
self.assertEqual(4, body["your_rank"])
self.assertEqual(4, body["your_page"])
self.assertEqual("alice", body["results"][0]["user_sub"])
anonymous_body = anonymous.json()
self.assertEqual(1, anonymous_body["page"])
self.assertIsNone(anonymous_body["your_rank"])
self.assertIsNone(anonymous_body["your_page"])
@async_test
async def test_leaderboard_rejects_bad_page_params(self) -> None:
app, _, _ = make_platform()
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
for query in ("page=0", "page=abc", "focus=elsewhere"):
response = await client.get(f"/api/leaderboard?{query}")
self.assertEqual(400, response.status_code, query)
# The limit is clamped, not rejected.
clamped = await client.get("/api/leaderboard?limit=500")
self.assertEqual(200, clamped.status_code)
@async_test
async def test_my_matches_include_elo_delta(self) -> None:
app, platform, tortoise_mixin = make_platform()
ctx = await use_db(tortoise_mixin)
engine, session = _finished_session()
engine.handle_action(session, "alice", "play", {})
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(platform.oidc, "alice"):
response = await client.get("/api/me/matches")
self.assertEqual(200, response.status_code)
players = {
p["user_sub"]: p
for p in response.json()["results"][0]["players"]
}
self.assertEqual(16, players["alice"]["elo_delta"])
self.assertEqual(-16, players["bob"]["elo_delta"])
self.assertEqual(16, response.json()["results"][0]["your_elo_delta"])
@async_test
async def test_my_ratings_requires_auth(self) -> None:
app, _, _ = make_platform()
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/api/me/ratings")
self.assertEqual(401, response.status_code)
@async_test
async def test_my_ratings_returns_only_own_rows(self) -> None:
app, platform, tortoise_mixin = make_platform()
ctx = await use_db(tortoise_mixin)
with ctx:
await PlayerRating.create(
id=uuid.uuid4(),
user_sub="alice",
game_type="dummy",
rating=1516,
matches_played=1,
)
await PlayerRating.create(
id=uuid.uuid4(),
user_sub="bob",
game_type="dummy",
rating=1484,
matches_played=1,
)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
with oidc_user(platform.oidc, "alice"):
response = await client.get("/api/me/ratings")
self.assertEqual(200, response.status_code)
self.assertEqual(
[{"game_type": "dummy", "rating": 1516, "matches_played": 1}],
response.json()["results"],
)
class SecondEngine(DummyEngine):
id = "second"
name = "Second game"
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.
app, platform, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin, game_types=("dummy", "other_game"))
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
with oidc_user(platform.oidc, "alice"):
all_matches = await client.get("/api/me/matches")
scoped = await client.get("/api/me/matches?game_type=dummy")
unknown = await client.get("/api/me/matches?game_type=briscola")
self.assertEqual(2, len(all_matches.json()["results"]))
self.assertEqual(
{"dummy", "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("dummy", scoped_results[0]["game_type"])
self.assertEqual(400, unknown.status_code)
@async_test
async def test_leaderboard_filter_by_game_type(self) -> None:
app, platform, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin, game_types=("dummy", "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=dummy")
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, alice 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()