woggioni
551c4c9c28
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.
106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
"""Keyset pagination tests: cursor values are restored to Python types.
|
|
|
|
Cursors are JSON, so a cursor carrying a datetime reaches the next
|
|
request as a string. Postgres/asyncpg rejects strings for timestamp
|
|
parameters (sqlite silently coerces them), so ``paginate`` must run the
|
|
decoded values through their model fields before building the WHERE
|
|
clause.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any, List
|
|
|
|
from tavolo.platform.models import Match
|
|
from tavolo.platform.pagination import (
|
|
Cursor,
|
|
PageParamsError,
|
|
decode_cursor,
|
|
encode_cursor,
|
|
paginate,
|
|
parse_page_params,
|
|
)
|
|
|
|
from helpers import async_test
|
|
|
|
SORT = [("finished_at", "DESC"), ("id", "ASC")]
|
|
|
|
|
|
class _CapturingQuerySet:
|
|
"""Minimal stand-in for a Tortoise QuerySet, recording its filter."""
|
|
|
|
model = Match
|
|
|
|
def __init__(self) -> None:
|
|
self.queryset_filter: Any = None
|
|
|
|
def order_by(self, *orderings: str) -> "_CapturingQuerySet":
|
|
return self
|
|
|
|
def filter(self, q: Any) -> "_CapturingQuerySet":
|
|
self.queryset_filter = q
|
|
return self
|
|
|
|
async def limit(self, count: int) -> List[Any]:
|
|
return []
|
|
|
|
|
|
def _flatten_q(q: Any) -> List[Any]:
|
|
nodes = [q]
|
|
for child in q.children:
|
|
nodes.extend(_flatten_q(child))
|
|
return nodes
|
|
|
|
|
|
def _filter_values(q: Any, key: str) -> List[Any]:
|
|
return [node.filters[key] for node in _flatten_q(q) if key in node.filters]
|
|
|
|
|
|
class PaginateCursorTypesTest(unittest.TestCase):
|
|
@async_test
|
|
async def test_cursor_values_are_restored_before_filtering(self) -> None:
|
|
match_id = uuid.uuid4()
|
|
finished_at = datetime(2026, 9, 19, 5, 2, 50, 547904, tzinfo=timezone.utc)
|
|
# Round-trip through the wire format, exactly as a request does.
|
|
after = decode_cursor(
|
|
encode_cursor({"finished_at": finished_at, "id": match_id})
|
|
)
|
|
assert after is not None
|
|
|
|
qs = _CapturingQuerySet()
|
|
await paginate(qs, SORT, Cursor(limit=1, after=after))
|
|
|
|
assert qs.queryset_filter is not None
|
|
for key in ("finished_at", "finished_at__lt"):
|
|
values = _filter_values(qs.queryset_filter, key)
|
|
self.assertEqual(1, len(values))
|
|
self.assertIsInstance(values[0], datetime)
|
|
self.assertEqual(finished_at, values[0])
|
|
id_values = _filter_values(qs.queryset_filter, "id__gt")
|
|
self.assertEqual(1, len(id_values))
|
|
self.assertIsInstance(id_values[0], uuid.UUID)
|
|
self.assertEqual(match_id, id_values[0])
|
|
|
|
|
|
class PageParamsTest(unittest.TestCase):
|
|
def test_defaults(self) -> None:
|
|
self.assertEqual((1, 20), parse_page_params(""))
|
|
|
|
def test_page_and_limit(self) -> None:
|
|
self.assertEqual((3, 5), parse_page_params("page=3&limit=5"))
|
|
|
|
def test_limit_is_clamped(self) -> None:
|
|
self.assertEqual((1, 100), parse_page_params("limit=500"))
|
|
self.assertEqual((1, 1), parse_page_params("limit=0"))
|
|
|
|
def test_invalid_values_raise(self) -> None:
|
|
for query in ("page=0", "page=-2", "page=abc", "limit=abc"):
|
|
with self.assertRaises(PageParamsError, msg=query):
|
|
parse_page_params(query)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|