"""Cursor-based pagination for symbol detail — provenance and coupling.

TDD spec — tests are written before implementation.

Provenance:  10 entries per page, cursor = committed_at ISO string of last entry.
Coupling:    15 entries per page, cursor = "shared_commits:address" of last row.

URL contracts
─────────────
  GET /{owner}/{repo}/symbol/{address}                       → page 1 (no cursor)
  GET /{owner}/{repo}/symbol/{address}?history_cursor=<iso>  → provenance page N+1
  GET /{owner}/{repo}/symbol/{address}?coupling_cursor=<sha> → coupling page N+1

Both cursors may appear together (independent pagination).

Tier breakdown
──────────────
  P1xx  Unit — pure pagination helpers
  P2xx  Integration — route returns correct slices with real DB rows
  P3xx  HTML — template wires next-page links correctly
  P4xx  Edge cases — empty, last page, invalid cursor
  P5xx  Performance — no extra DB round-trips per page
  P6xx  Security — cursor injection does not leak data or 500
"""
from __future__ import annotations

import datetime as _dt
import typing
from collections.abc import MutableMapping, Sequence

import pytest
from httpx import AsyncClient
from sqlalchemy.engine import CursorResult
from sqlalchemy.sql.base import Executable
from sqlalchemy.ext.asyncio import AsyncSession
from muse.core.types import blob_id

# ---------------------------------------------------------------------------
# Shared constants
# ---------------------------------------------------------------------------

HISTORY_PAGE   = 10
COUPLING_PAGE  = 15

# ---------------------------------------------------------------------------
# P1xx — Unit tests (pure helpers, no DB)
# ---------------------------------------------------------------------------


class TestHistoryCursorParsing:
    """P101–P104: history cursor encode/decode round-trips."""

    def _encode(self, iso: str) -> str:
        # The cursor IS the ISO timestamp of the last entry on the current page.
        return iso

    def _decode(self, cursor: str) -> _dt.datetime:
        return _dt.datetime.fromisoformat(cursor)

    def test_P101_roundtrip_utc(self) -> None:
        """P101: ISO cursor survives encode→decode for a UTC timestamp."""
        ts = "2026-01-15T10:30:00+00:00"
        assert self._decode(self._encode(ts)).isoformat() == _dt.datetime.fromisoformat(ts).isoformat()

    def test_P102_cursor_is_comparable(self) -> None:
        """P102: decoded cursor compares correctly to committed_at values."""
        cursor_ts = _dt.datetime.fromisoformat("2026-01-15T10:30:00+00:00")
        older     = _dt.datetime.fromisoformat("2026-01-10T00:00:00+00:00")
        newer     = _dt.datetime.fromisoformat("2026-01-20T00:00:00+00:00")
        assert older < cursor_ts
        assert newer > cursor_ts

    def test_P103_page_size_constant_is_10(self) -> None:
        """P103: HISTORY_PAGE == 10."""
        assert HISTORY_PAGE == 10

    def test_P104_coupling_page_size_constant_is_15(self) -> None:
        """P104: COUPLING_PAGE == 15."""
        assert COUPLING_PAGE == 15


class TestCouplingCursorParsing:
    """P105–P107: coupling cursor encode/decode."""

    def _encode(self, shared: int, address: str) -> str:
        return f"{shared}:{address}"

    def _decode(self, cursor: str) -> tuple[int, str]:
        shared_str, _, addr = cursor.partition(":")
        return int(shared_str), addr

    def test_P105_roundtrip(self) -> None:
        """P105: coupling cursor survives encode→decode."""
        shared, addr = 7, "src/auth.py::validate_token"
        cursor = self._encode(shared, addr)
        s, a = self._decode(cursor)
        assert s == shared and a == addr

    def test_P106_address_with_colons_preserved(self) -> None:
        """P106: address containing '::' is preserved through cursor."""
        shared, addr = 3, "lib/core.py::MyClass::method"
        cursor = self._encode(shared, addr)
        s, a = self._decode(cursor)
        assert s == shared and a == addr

    def test_P107_zero_shared_valid(self) -> None:
        """P107: shared_commits == 0 is a valid cursor value."""
        cursor = self._encode(0, "src/x.py::fn")
        s, a = self._decode(cursor)
        assert s == 0 and a == "src/x.py::fn"


class TestPageSlicing:
    """P108–P112: page-slice logic on in-memory lists."""

    @staticmethod
    def _slice(items: Sequence[int], page: int, cursor_idx: int | None = None) -> tuple[list[int], bool, str | None]:
        """Simulate cursor pagination over a pre-sorted list."""
        start = cursor_idx + 1 if cursor_idx is not None else 0
        window = items[start : start + page + 1]
        has_next = len(window) > page
        page_items = window[:page]
        next_cursor = str(start + page - 1) if has_next else None
        return page_items, has_next, next_cursor

    def test_P108_first_page_returns_page_items(self) -> None:
        """P108: first page of 25 items returns exactly HISTORY_PAGE items."""
        items = list(range(25))
        page, has_next, cursor = self._slice(items, HISTORY_PAGE)
        assert len(page) == HISTORY_PAGE
        assert has_next is True
        assert cursor is not None

    def test_P109_last_page_has_no_next(self) -> None:
        """P109: last page has has_next=False and cursor=None."""
        items = list(range(12))       # 12 items, page=10 → second page has 2
        page, has_next, cursor = self._slice(items, HISTORY_PAGE, cursor_idx=9)
        assert has_next is False
        assert cursor is None

    def test_P110_exact_fit_no_next(self) -> None:
        """P110: exactly HISTORY_PAGE items → has_next=False."""
        items = list(range(HISTORY_PAGE))
        page, has_next, _ = self._slice(items, HISTORY_PAGE)
        assert len(page) == HISTORY_PAGE
        assert has_next is False

    def test_P111_empty_set_returns_empty(self) -> None:
        """P111: zero items → empty page, has_next=False."""
        page, has_next, cursor = self._slice([], HISTORY_PAGE)
        assert page == []
        assert has_next is False
        assert cursor is None

    def test_P112_coupling_page_size_15(self) -> None:
        """P112: COUPLING_PAGE slices produce at most 15 items."""
        items = list(range(40))
        page, has_next, cursor = self._slice(items, COUPLING_PAGE)
        assert len(page) == COUPLING_PAGE
        assert has_next is True


# ---------------------------------------------------------------------------
# P2xx — Integration tests (real DB, route handler)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
class TestProvenancePagination:
    """P201–P208: provenance history pagination via history_cursor query param."""

    async def test_P201_first_page_returns_10_entries(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P201: no cursor → exactly 10 history entries in HTML."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert resp.status_code == 200
        count = resp.content.count(b"sym2-tl-entry")
        assert count == HISTORY_PAGE

    async def test_P202_first_page_has_next_link(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P202: history_cursor link present when more entries exist."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert b"history_cursor=" in resp.content

    async def test_P203_last_page_no_next_link(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P203: final history page (offset 20) has 6 entries and no next link."""
        owner, slug, address = seed_symbol_with_26_history
        import re
        # Page 2 (offset 10) → 10 entries
        r2 = await client.get(f"/{owner}/{slug}/symbol/{address}?history_cursor=10")
        assert r2.status_code == 200
        assert r2.content.count(b"sym2-tl-entry") == HISTORY_PAGE
        # Page 3 (offset 20) → 6 remaining, no --next link
        r3 = await client.get(f"/{owner}/{slug}/symbol/{address}?history_cursor=20")
        assert r3.status_code == 200
        assert r3.content.count(b"sym2-tl-entry") == 6
        assert not re.search(rb'sym2-page-btn--next', r3.content)

    async def test_P204_no_cursor_shows_newest_first(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P204: first page shows most recent 10 entries (newest → oldest)."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        # The newest commit message contains 'entry-25' (seeded newest-last, displayed newest-first)
        assert b"entry-25" in resp.content
        assert b"entry-0" not in resp.content

    async def test_P205_cursor_page_does_not_overlap_previous(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P205: page 2 entries are disjoint from page 1."""
        owner, slug, address = seed_symbol_with_26_history
        import re
        r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
        m = re.search(rb'history_cursor=([^"&]+)', r1.content)
        cursor = m.group(1).decode()
        r2 = await client.get(f"/{owner}/{slug}/symbol/{address}?history_cursor={cursor}")
        # entry-15 should be on page 2, not page 1
        assert b"entry-15" not in r1.content
        assert b"entry-15" in r2.content

    async def test_P206_10_or_fewer_entries_no_pagination(
        self, client: AsyncClient, seed_symbol: tuple[str, str, str]
    ) -> None:
        """P206: symbol with 1 entry shows no history_cursor link."""
        owner, slug, address = seed_symbol
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert resp.status_code == 200
        assert b"history_cursor=" not in resp.content

    async def test_P207_history_total_count_in_context(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P207: page renders total provenance count for 'showing X of N' display."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        # change_count drives the vitals quad and narrative
        assert b"26" in resp.content

    async def test_P208_invalid_history_cursor_returns_200_first_page(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P208: malformed cursor falls back to first page gracefully."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(
            f"/{owner}/{slug}/symbol/{address}?history_cursor=not-a-date"
        )
        assert resp.status_code == 200
        # Should render first page normally
        assert resp.content.count(b"sym2-tl-entry") == HISTORY_PAGE


@pytest.mark.asyncio
class TestCouplingPagination:
    """P211–P218: coupling partners pagination via coupling_cursor query param."""

    async def test_P211_first_page_returns_15_partners(
        self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
    ) -> None:
        """P211: no cursor → exactly 15 coupling rows."""
        owner, slug, address = seed_symbol_high_coupling_40
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert resp.status_code == 200
        assert resp.content.count(b"sym2-blast-row") == COUPLING_PAGE

    async def test_P212_first_page_has_coupling_next_link(
        self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
    ) -> None:
        """P212: coupling_cursor link present when more partners exist."""
        owner, slug, address = seed_symbol_high_coupling_40
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert b"coupling_cursor=" in resp.content

    async def test_P213_second_page_returns_remaining(
        self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
    ) -> None:
        """P213: page 2 returns the remaining 25 partners (capped at 15)."""
        owner, slug, address = seed_symbol_high_coupling_40
        import re
        r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
        m = re.search(rb'coupling_cursor=([^"&]+)', r1.content)
        assert m
        cursor = m.group(1).decode()
        r2 = await client.get(f"/{owner}/{slug}/symbol/{address}?coupling_cursor={cursor}")
        assert r2.status_code == 200
        assert r2.content.count(b"sym2-blast-row") == COUPLING_PAGE

    async def test_P214_last_coupling_page_no_next_link(
        self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
    ) -> None:
        """P214: final coupling page has no coupling_cursor link."""
        owner, slug, address = seed_symbol_high_coupling_40
        import re
        # 40 partners: p1=15 (offset 0), p2=15 (offset 15), p3=10 (offset 30)
        r3 = await client.get(f"/{owner}/{slug}/symbol/{address}?coupling_cursor=30")
        assert r3.status_code == 200
        assert r3.content.count(b"sym2-blast-row") == 10
        # Coupling section has no "Next ›" link (history may still have "Older ›")
        assert b"Next \xe2\x80\xba" not in r3.content

    async def test_P215_fewer_than_15_partners_no_pagination(
        self, client: AsyncClient, seed_symbol: tuple[str, str, str]
    ) -> None:
        """P215: symbol with 0 coupling partners shows no coupling_cursor link."""
        owner, slug, address = seed_symbol
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert b"coupling_cursor=" not in resp.content

    async def test_P216_both_cursors_independent(
        self, client: AsyncClient, seed_symbol_with_26_history_and_40_coupling: tuple[str, str, str]
    ) -> None:
        """P216: history_cursor and coupling_cursor paginate independently."""
        owner, slug, address = seed_symbol_with_26_history_and_40_coupling
        import re
        r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
        hc = re.search(rb'history_cursor=([^"&]+)', r1.content)
        cc = re.search(rb'coupling_cursor=([^"&]+)', r1.content)
        assert hc and cc
        # Advance only history cursor
        r2 = await client.get(
            f"/{owner}/{slug}/symbol/{address}"
            f"?history_cursor={hc.group(1).decode()}"
        )
        assert r2.status_code == 200
        assert r2.content.count(b"sym2-blast-row") == COUPLING_PAGE

    async def test_P217_invalid_coupling_cursor_returns_200_first_page(
        self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
    ) -> None:
        """P217: malformed coupling cursor falls back to first page."""
        owner, slug, address = seed_symbol_high_coupling_40
        resp = await client.get(
            f"/{owner}/{slug}/symbol/{address}?coupling_cursor=garbage"
        )
        assert resp.status_code == 200
        assert resp.content.count(b"sym2-blast-row") == COUPLING_PAGE

    async def test_P218_coupling_ordered_by_shared_commits_desc(
        self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
    ) -> None:
        """P218: coupling page 1 contains highest-shared partners first."""
        owner, slug, address = seed_symbol_high_coupling_40
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        # The fixture seeds partners with shared counts 40..1; highest first
        assert b"40\xc3\x97" in resp.content or b"40" in resp.content  # top partner count


# ---------------------------------------------------------------------------
# P3xx — HTML structural tests
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
class TestPaginationHTML:
    """P301–P306: template renders pagination controls correctly."""

    async def test_P301_next_history_link_is_anchor(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P301: 'Load more' history link is an <a> tag with history_cursor param."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert b'href=' in resp.content
        assert b'history_cursor=' in resp.content

    async def test_P302_next_coupling_link_is_anchor(
        self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
    ) -> None:
        """P302: 'Load more' coupling link is an <a> tag with coupling_cursor param."""
        owner, slug, address = seed_symbol_high_coupling_40
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert b'href=' in resp.content
        assert b'coupling_cursor=' in resp.content

    async def test_P303_page_indicator_present(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P303: provenance section shows current count and total."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        # e.g. "10 of 26" or "Showing 10"
        assert b"10" in resp.content and b"26" in resp.content

    async def test_P304_no_controls_when_single_page(self, client: AsyncClient, seed_symbol: tuple[str, str, str]) -> None:
        """P304: no pagination controls rendered for single-page symbol."""
        owner, slug, address = seed_symbol
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert b"history_cursor=" not in resp.content
        assert b"coupling_cursor=" not in resp.content

    async def test_P305_cursor_preserved_in_coupling_next_link(
        self, client: AsyncClient, seed_symbol_with_26_history_and_40_coupling: tuple[str, str, str]
    ) -> None:
        """P305: coupling next link preserves active history_cursor in href."""
        owner, slug, address = seed_symbol_with_26_history_and_40_coupling
        import re
        r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
        hc = re.search(rb'history_cursor=([^"&]+)', r1.content).group(1).decode()
        # Navigate to history page 2 while on coupling page 1
        r2 = await client.get(
            f"/{owner}/{slug}/symbol/{address}?history_cursor={hc}"
        )
        assert r2.status_code == 200
        # coupling next link in r2 must carry history_cursor forward
        cc_links = [m.group() for m in re.finditer(rb'href="[^"]*coupling_cursor=[^"]*"', r2.content)]
        assert any(hc.encode() in link for link in cc_links)

    async def test_P306_history_next_link_preserves_coupling_cursor(
        self, client: AsyncClient, seed_symbol_with_26_history_and_40_coupling: tuple[str, str, str]
    ) -> None:
        """P306: history next link preserves active coupling_cursor in href."""
        owner, slug, address = seed_symbol_with_26_history_and_40_coupling
        import re
        r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
        cc = re.search(rb'coupling_cursor=([^"&]+)', r1.content).group(1).decode()
        r2 = await client.get(
            f"/{owner}/{slug}/symbol/{address}?coupling_cursor={cc}"
        )
        assert r2.status_code == 200
        hc_links = [m.group() for m in re.finditer(rb'href="[^"]*history_cursor=[^"]*"', r2.content)]
        assert any(cc.encode() in link for link in hc_links)


# ---------------------------------------------------------------------------
# P4xx — Edge cases
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
class TestPaginationEdgeCases:
    """P401–P406: boundary conditions."""

    async def test_P401_exactly_10_history_no_next(self, client: AsyncClient, seed_symbol_with_exactly_10_history: tuple[str, str, str]) -> None:
        """P401: exactly 10 history entries → no history_cursor link."""
        owner, slug, address = seed_symbol_with_exactly_10_history
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert resp.status_code == 200
        assert resp.content.count(b"sym2-tl-entry") == HISTORY_PAGE
        assert b"history_cursor=" not in resp.content

    async def test_P402_exactly_11_history_has_next(self, client: AsyncClient, seed_symbol_with_11_history: tuple[str, str, str]) -> None:
        """P402: 11 history entries → first page has next link."""
        owner, slug, address = seed_symbol_with_11_history
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert resp.content.count(b"sym2-tl-entry") == HISTORY_PAGE
        assert b"history_cursor=" in resp.content

    async def test_P403_exactly_15_coupling_no_next(self, client: AsyncClient, seed_symbol_with_exactly_15_coupling: tuple[str, str, str]) -> None:
        """P403: exactly 15 coupling partners → no coupling_cursor link."""
        owner, slug, address = seed_symbol_with_exactly_15_coupling
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert resp.content.count(b"sym2-blast-row") == COUPLING_PAGE
        assert b"coupling_cursor=" not in resp.content

    async def test_P404_exactly_16_coupling_has_next(self, client: AsyncClient, seed_symbol_with_16_coupling: tuple[str, str, str]) -> None:
        """P404: 16 coupling partners → first page has next link."""
        owner, slug, address = seed_symbol_with_16_coupling
        resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
        assert resp.content.count(b"sym2-blast-row") == COUPLING_PAGE
        assert b"coupling_cursor=" in resp.content

    async def test_P405_past_cursor_returns_empty_page(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P405: offset past the end of history returns empty timeline."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(
            f"/{owner}/{slug}/symbol/{address}?history_cursor=10000"
        )
        assert resp.status_code == 200
        assert resp.content.count(b"sym2-tl-entry") == 0

    async def test_P406_both_cursors_on_last_pages_no_links(
        self, client: AsyncClient, seed_symbol_with_26_history_and_40_coupling: tuple[str, str, str]
    ) -> None:
        """P406: when both paginations are on final pages, no cursor links appear.

        Fixture: 26 history entries (3 pages: 10+10+6) + 26 coupling partners
        (2 pages: 15+11).  Exhaust both to confirm no pagination links remain.
        """
        owner, slug, address = seed_symbol_with_26_history_and_40_coupling
        import re
        # Page 1: has both cursors
        r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
        # Fixture: 26 history (3 pages: 0,10,20) + 26 coupling (2 pages: 0,15)
        # Final request: history page 3 (last) + coupling page 2 (last)
        r_final = await client.get(
            f"/{owner}/{slug}/symbol/{address}?history_cursor=20&coupling_cursor=15"
        )
        assert r_final.status_code == 200
        # Neither section has a next button
        assert b"Older \xe2\x80\xba" not in r_final.content   # no history next
        assert b"Next \xe2\x80\xba" not in r_final.content    # no coupling next


# ---------------------------------------------------------------------------
# P5xx — Performance
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
class TestPaginationPerformance:
    """P501–P503: pagination does not increase DB round-trips."""

    async def test_P501_history_page2_same_query_count_as_page1(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str], db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """P501: paginated request makes no more DB calls than page 1."""
        import re
        r1 = await client.get(
            f"/{seed_symbol_with_26_history[0]}/{seed_symbol_with_26_history[1]}"
            f"/symbol/{seed_symbol_with_26_history[2]}"
        )
        hc = re.search(rb'history_cursor=([^"&]+)', r1.content).group(1).decode()

        call_counts: list[int] = []
        for cursor in [None, hc]:
            count = {"n": 0}
            orig = db_session.execute
            async def spy(*a: Executable | typing.Any, _c: MutableMapping[str, int] = count, _o: typing.Any = orig, **kw: typing.Any) -> CursorResult[typing.Any]:
                _c["n"] += 1
                return await _o(*a, **kw)
            monkeypatch.setattr(db_session, "execute", spy)
            url = (f"/{seed_symbol_with_26_history[0]}/{seed_symbol_with_26_history[1]}"
                   f"/symbol/{seed_symbol_with_26_history[2]}")
            if cursor:
                url += f"?history_cursor={cursor}"
            await client.get(url)
            call_counts.append(count["n"])
            monkeypatch.undo()

        assert call_counts[1] <= call_counts[0] + 1  # at most 1 extra call

    async def test_P502_coupling_page_uses_offset_not_python_scan(self) -> None:
        """P502: coupling cursor pagination uses SQL WHERE, not Python list slice."""
        from sqlalchemy import select, func as sa_func
        from musehub.db.musehub_intel_models import MusehubSymbolHistoryEntry
        # Verify that the query accepts a LIMIT and the architecture allows WHERE for cursor
        stmt = (
            select(
                MusehubSymbolHistoryEntry.address,
                sa_func.count().label("shared"),
            )
            .where(MusehubSymbolHistoryEntry.repo_id == "x")
            .group_by(MusehubSymbolHistoryEntry.address)
            .order_by(sa_func.count().desc())
            .limit(COUPLING_PAGE + 1)
        )
        compiled = str(stmt.compile(compile_kwargs={"literal_binds": False}))
        assert "LIMIT" in compiled.upper()
        assert "GROUP BY" in compiled.upper()

    async def test_P503_render_time_under_300ms(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str], benchmark_timer: typing.Callable[[float], typing.ContextManager[None]]
    ) -> None:
        """P503: paginated page renders in < 300ms."""
        owner, slug, address = seed_symbol_with_26_history
        import re
        r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
        hc = re.search(rb'history_cursor=([^"&]+)', r1.content).group(1).decode()
        with benchmark_timer(max_ms=300):
            resp = await client.get(
                f"/{owner}/{slug}/symbol/{address}?history_cursor={hc}"
            )
        assert resp.status_code == 200


# ---------------------------------------------------------------------------
# P6xx — Security
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
class TestPaginationSecurity:
    """P601–P604: cursor values cannot leak data or cause server errors."""

    async def test_P601_sql_injection_in_history_cursor(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P601: SQL injection in history_cursor param → 200 first page, no 500."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(
            f"/{owner}/{slug}/symbol/{address}"
            "?history_cursor='; DROP TABLE musehub_symbol_history_entries; --"
        )
        assert resp.status_code == 200

    async def test_P602_sql_injection_in_coupling_cursor(
        self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
    ) -> None:
        """P602: SQL injection in coupling_cursor param → 200 first page, no 500."""
        owner, slug, address = seed_symbol_high_coupling_40
        resp = await client.get(
            f"/{owner}/{slug}/symbol/{address}"
            "?coupling_cursor=0:x' OR '1'='1"
        )
        assert resp.status_code == 200

    async def test_P603_xss_in_history_cursor_escaped(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P603: XSS payload in history_cursor is never reflected raw in HTML."""
        owner, slug, address = seed_symbol_with_26_history
        resp = await client.get(
            f"/{owner}/{slug}/symbol/{address}"
            "?history_cursor=<script>alert(1)</script>"
        )
        assert b"<script>alert(1)</script>" not in resp.content

    async def test_P604_very_long_cursor_no_500(
        self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
    ) -> None:
        """P604: cursor > 512 chars returns 200 (graceful fallback) never 500."""
        owner, slug, address = seed_symbol_with_26_history
        long_cursor = "x" * 600
        resp = await client.get(
            f"/{owner}/{slug}/symbol/{address}?history_cursor={long_cursor}"
        )
        assert resp.status_code == 200


# ---------------------------------------------------------------------------
# New fixtures (appended to conftest.py separately)
# ---------------------------------------------------------------------------
# The fixtures below are declared here as documentation of what conftest.py
# must provide. They are moved to conftest.py by the implementation step.

"""
Required new fixtures:

  seed_symbol_with_26_history
    → repo + symbol with 26 history entries spaced 1h apart, newest last.
      Commit messages contain 'entry-{i}' for i in 0..25.

  seed_symbol_high_coupling_40
    → repo + symbol with 1 history entry + 40 partner symbols each sharing
      that 1 commit, with shared_commits values 40..1 (descending by address).

  seed_symbol_with_26_history_and_40_coupling
    → combines both: 26 history entries + 40 coupling partners.

  seed_symbol_with_exactly_10_history
    → repo + symbol with exactly 10 history entries.

  seed_symbol_with_11_history
    → repo + symbol with exactly 11 history entries.

  seed_symbol_with_exactly_15_coupling
    → repo + symbol with 15 coupling partners.

  seed_symbol_with_16_coupling
    → repo + symbol with 16 coupling partners.
"""