Restore cursor value types before keyset filtering
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.
This commit is contained in:
2 files changed
+109
-2
No files matched your search
@@ -0,0 +1,81 @@
|
||||
"""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()
|
||||
Reference in new issue
Block a user