Logan CusanoandClaude Opus 5.5
40014a47a3
/calls/search scans a 200-row window and returns 50, but the next cursor was always the last SCANNED row — so with an unfiltered list each page jumped past the 150 matches it had already read and not shown. Resume after the last RETURNED row when matches overflow the page; keep the last-scanned cursor only when the page holds every match (the sparse- filter case that cursor exists for). Same fix for /calls/eval-queue. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
"""calls._parse_ts — cursor/date bounds must reach Firestore as datetimes.
|
|
|
|
A raw ISO string compared against a timestamp field sorts by type, not time,
|
|
which made the Archive's "Load more" return the first page again.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.routers.calls import _next_cursor, _parse_ts
|
|
|
|
|
|
def test_empty_is_none():
|
|
assert _parse_ts(None, "cursor") is None
|
|
assert _parse_ts("", "cursor") is None
|
|
|
|
|
|
def test_z_suffix_parses_as_utc():
|
|
assert _parse_ts("2026-09-20T12:00:00Z", "date_from") == datetime(2026, 9, 20, 12, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_naive_is_assumed_utc():
|
|
assert _parse_ts("2026-09-20T12:00:00", "date_to").tzinfo == timezone.utc
|
|
|
|
|
|
def test_round_trips_isoformat_cursor():
|
|
dt = datetime(2026, 9, 20, 12, 30, 5, 123456, tzinfo=timezone.utc)
|
|
assert _parse_ts(dt.isoformat(), "cursor") == dt
|
|
|
|
|
|
def test_garbage_is_400():
|
|
with pytest.raises(HTTPException) as exc:
|
|
_parse_ts("yesterday", "date_from")
|
|
assert exc.value.status_code == 400
|
|
|
|
|
|
# ── _next_cursor ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def _rows(n):
|
|
return [{"started_at": datetime(2026, 9, 20, 12, i // 60, i % 60, tzinfo=timezone.utc)} for i in range(n)]
|
|
|
|
|
|
def test_cursor_resumes_after_last_returned_row_when_matches_overflow():
|
|
rows = _rows(200)
|
|
page = rows[:50]
|
|
assert _next_cursor(rows, rows, page, 200) == page[-1]["started_at"].isoformat()
|
|
|
|
|
|
def test_cursor_resumes_after_window_when_page_holds_every_match():
|
|
rows = _rows(200)
|
|
matches = rows[:3]
|
|
assert _next_cursor(rows, matches, matches, 200) == rows[-1]["started_at"].isoformat()
|
|
|
|
|
|
def test_short_window_is_the_end():
|
|
rows = _rows(20)
|
|
assert _next_cursor(rows, rows[:5], rows[:5], 200) is None
|