90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from govoplan_access.backend.api.v1.routes import (
|
|
_decode_full_delta_cursor,
|
|
_encode_full_delta_cursor,
|
|
_full_delta_page,
|
|
_page_query,
|
|
)
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, rows: list[int]) -> None:
|
|
self._rows = rows
|
|
self._offset = 0
|
|
self._limit: int | None = None
|
|
|
|
def order_by(self, *_args: object) -> FakeQuery:
|
|
return self
|
|
|
|
def count(self) -> int:
|
|
return len(self._rows)
|
|
|
|
def offset(self, value: int) -> FakeQuery:
|
|
clone = FakeQuery(self._rows)
|
|
clone._offset = value
|
|
clone._limit = self._limit
|
|
return clone
|
|
|
|
def limit(self, value: int) -> FakeQuery:
|
|
clone = FakeQuery(self._rows)
|
|
clone._offset = self._offset
|
|
clone._limit = value
|
|
return clone
|
|
|
|
def all(self) -> list[int]:
|
|
end = None if self._limit is None else self._offset + self._limit
|
|
return self._rows[self._offset:end]
|
|
|
|
|
|
class AdminPaginationTests(unittest.TestCase):
|
|
def test_page_query_returns_page_and_metadata(self) -> None:
|
|
rows, metadata = _page_query(FakeQuery(list(range(7))), page=2, page_size=3)
|
|
|
|
self.assertEqual(rows, [3, 4, 5])
|
|
self.assertEqual(metadata, {"total": 7, "page": 2, "page_size": 3, "pages": 3})
|
|
|
|
def test_full_delta_page_returns_cursor_until_last_page(self) -> None:
|
|
rows, metadata, watermark, has_more = _full_delta_page(
|
|
FakeQuery(list(range(7))),
|
|
page=2,
|
|
page_size=3,
|
|
scope="users",
|
|
snapshot_sequence=42,
|
|
)
|
|
|
|
self.assertEqual(rows, [3, 4, 5])
|
|
self.assertEqual(metadata, {"total": 7, "page": 2, "page_size": 3, "pages": 3})
|
|
self.assertEqual(watermark, "full:users:3:42")
|
|
self.assertTrue(has_more)
|
|
|
|
def test_full_delta_page_returns_sequence_watermark_on_last_page(self) -> None:
|
|
rows, metadata, watermark, has_more = _full_delta_page(
|
|
FakeQuery(list(range(7))),
|
|
page=3,
|
|
page_size=3,
|
|
scope="users",
|
|
snapshot_sequence=42,
|
|
)
|
|
|
|
self.assertEqual(rows, [6])
|
|
self.assertEqual(metadata, {"total": 7, "page": 3, "page_size": 3, "pages": 3})
|
|
self.assertEqual(watermark, "seq:42")
|
|
self.assertFalse(has_more)
|
|
|
|
def test_full_delta_cursor_round_trips_and_rejects_wrong_scope(self) -> None:
|
|
cursor = _encode_full_delta_cursor("users", page=3, snapshot_sequence=42)
|
|
|
|
self.assertEqual(_decode_full_delta_cursor(cursor, scope="users"), (3, 42))
|
|
self.assertIsNone(_decode_full_delta_cursor("seq:42", scope="users"))
|
|
with self.assertRaises(HTTPException):
|
|
_decode_full_delta_cursor(cursor, scope="groups")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|