woggioni
2bdf247cb3
Cursors are JSON, so a datetime sort key reached the next request as a string. Tortoise's DatetimeField.to_db_value does not parse strings, so asyncpg rejected the string where a timestamptz parameter was expected; sqlite tolerated it, hiding the bug in tests. Coerce decoded cursor values through their model field's to_python_value before building the keyset filter.
82 lines
2.5 KiB
Python
82 lines
2.5 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, decode_cursor, encode_cursor, paginate
|
|
|
|
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])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|