- Implement cursor-based pagination in pagination.py - Update response models to standardize pagination structure - Add cursor pagination utilities for repositories - Update HistoryArchiveRepository and ImportLogRepository with new pagination - Add comprehensive tests for cursor pagination - Update documentation for backend development and task tracking Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
127 lines
4.8 KiB
Python
127 lines
4.8 KiB
Python
"""Tests for cursor-based pagination utilities."""
|
|
|
|
import pytest
|
|
|
|
from app.utils.pagination import decode_cursor, encode_cursor
|
|
|
|
|
|
class TestEncodeCursor:
|
|
"""Test encode_cursor() function."""
|
|
|
|
def test_encodes_valid_row_id(self) -> None:
|
|
"""encode_cursor encodes a valid positive row ID."""
|
|
cursor = encode_cursor(42)
|
|
assert isinstance(cursor, str)
|
|
assert len(cursor) > 0
|
|
|
|
def test_encoded_cursor_is_decodable(self) -> None:
|
|
"""Encoded cursor can be decoded back to original ID."""
|
|
original_id = 12345
|
|
cursor = encode_cursor(original_id)
|
|
decoded_id = decode_cursor(cursor)
|
|
assert decoded_id == original_id
|
|
|
|
def test_raises_for_zero_id(self) -> None:
|
|
"""encode_cursor raises ValueError for row_id < 1."""
|
|
with pytest.raises(ValueError, match="row_id must be >= 1"):
|
|
encode_cursor(0)
|
|
|
|
def test_raises_for_negative_id(self) -> None:
|
|
"""encode_cursor raises ValueError for negative row_id."""
|
|
with pytest.raises(ValueError, match="row_id must be >= 1"):
|
|
encode_cursor(-5)
|
|
|
|
def test_different_ids_produce_different_cursors(self) -> None:
|
|
"""Different row IDs produce different cursor strings."""
|
|
cursor1 = encode_cursor(1)
|
|
cursor2 = encode_cursor(2)
|
|
assert cursor1 != cursor2
|
|
|
|
def test_encoding_is_deterministic(self) -> None:
|
|
"""encode_cursor produces the same output for the same input."""
|
|
cursor1 = encode_cursor(999)
|
|
cursor2 = encode_cursor(999)
|
|
assert cursor1 == cursor2
|
|
|
|
|
|
class TestDecodeCursor:
|
|
"""Test decode_cursor() function."""
|
|
|
|
def test_decodes_valid_cursor(self) -> None:
|
|
"""decode_cursor correctly decodes a valid cursor."""
|
|
original_id = 555
|
|
cursor = encode_cursor(original_id)
|
|
decoded_id = decode_cursor(cursor)
|
|
assert decoded_id == 555
|
|
|
|
def test_raises_for_invalid_base64(self) -> None:
|
|
"""decode_cursor raises ValueError for invalid base64."""
|
|
with pytest.raises(ValueError, match="Invalid cursor format"):
|
|
decode_cursor("not-valid-base64!!!")
|
|
|
|
def test_raises_for_invalid_json(self) -> None:
|
|
"""decode_cursor raises ValueError when JSON is invalid."""
|
|
import base64
|
|
invalid_json = base64.b64encode(b"not json").decode("ascii")
|
|
with pytest.raises(ValueError, match="Invalid cursor format"):
|
|
decode_cursor(invalid_json)
|
|
|
|
def test_raises_for_missing_id_field(self) -> None:
|
|
"""decode_cursor raises ValueError when 'id' field is missing."""
|
|
import base64
|
|
import json
|
|
cursor_data = {"other_field": 42}
|
|
invalid_cursor = base64.b64encode(json.dumps(cursor_data).encode()).decode("ascii")
|
|
with pytest.raises(ValueError, match="Invalid cursor format"):
|
|
decode_cursor(invalid_cursor)
|
|
|
|
def test_raises_for_non_integer_id(self) -> None:
|
|
"""decode_cursor raises ValueError when 'id' is not an integer."""
|
|
import base64
|
|
import json
|
|
cursor_data = {"id": "not-an-int"}
|
|
invalid_cursor = base64.b64encode(json.dumps(cursor_data).encode()).decode("ascii")
|
|
with pytest.raises(ValueError, match="Invalid cursor format"):
|
|
decode_cursor(invalid_cursor)
|
|
|
|
def test_raises_for_invalid_id_value(self) -> None:
|
|
"""decode_cursor raises ValueError when 'id' is < 1."""
|
|
import base64
|
|
import json
|
|
cursor_data = {"id": 0}
|
|
invalid_cursor = base64.b64encode(json.dumps(cursor_data).encode()).decode("ascii")
|
|
with pytest.raises(ValueError, match="Invalid cursor format"):
|
|
decode_cursor(invalid_cursor)
|
|
|
|
def test_roundtrip_large_id(self) -> None:
|
|
"""Roundtrip encoding/decoding works for large row IDs."""
|
|
large_id = 999999999
|
|
cursor = encode_cursor(large_id)
|
|
decoded_id = decode_cursor(cursor)
|
|
assert decoded_id == large_id
|
|
|
|
|
|
class TestCursorPaginationIntegration:
|
|
"""Integration tests for cursor pagination workflow."""
|
|
|
|
def test_pagination_workflow_first_page(self) -> None:
|
|
"""Simulate pagination workflow: start with no cursor."""
|
|
page_size = 10
|
|
# First page: no cursor
|
|
cursor = None
|
|
# ... fetch items and get last_id = 100
|
|
cursor = encode_cursor(100)
|
|
assert isinstance(cursor, str)
|
|
|
|
def test_pagination_workflow_subsequent_pages(self) -> None:
|
|
"""Simulate pagination workflow: decode cursor for next page."""
|
|
# Previous page ended at ID 100
|
|
cursor = encode_cursor(100)
|
|
# Decode to get WHERE clause: WHERE id < 100
|
|
last_id = decode_cursor(cursor)
|
|
assert last_id == 100
|
|
# Fetch next page with WHERE id < 100
|
|
# ... mock fetch returns items ending at ID 50
|
|
next_cursor = encode_cursor(50)
|
|
assert decode_cursor(next_cursor) == 50
|