gabriel / musehub public
test_symbol_detail_pagination.py python
661 lines 29.9 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 22 days ago
1 """Cursor-based pagination for symbol detail — provenance and coupling.
2
3 TDD spec — tests are written before implementation.
4
5 Provenance: 10 entries per page, cursor = committed_at ISO string of last entry.
6 Coupling: 15 entries per page, cursor = "shared_commits:address" of last row.
7
8 URL contracts
9 ─────────────
10 GET /{owner}/{repo}/symbol/{address} → page 1 (no cursor)
11 GET /{owner}/{repo}/symbol/{address}?history_cursor=<iso> → provenance page N+1
12 GET /{owner}/{repo}/symbol/{address}?coupling_cursor=<sha> → coupling page N+1
13
14 Both cursors may appear together (independent pagination).
15
16 Tier breakdown
17 ──────────────
18 P1xx Unit — pure pagination helpers
19 P2xx Integration — route returns correct slices with real DB rows
20 P3xx HTML — template wires next-page links correctly
21 P4xx Edge cases — empty, last page, invalid cursor
22 P5xx Performance — no extra DB round-trips per page
23 P6xx Security — cursor injection does not leak data or 500
24 """
25 from __future__ import annotations
26
27 import datetime as _dt
28 import typing
29 from collections.abc import MutableMapping, Sequence
30
31 import pytest
32 from httpx import AsyncClient
33 from sqlalchemy.engine import CursorResult
34 from sqlalchemy.sql.base import Executable
35 from sqlalchemy.ext.asyncio import AsyncSession
36 from muse.core.types import blob_id
37
38 # ---------------------------------------------------------------------------
39 # Shared constants
40 # ---------------------------------------------------------------------------
41
42 HISTORY_PAGE = 10
43 COUPLING_PAGE = 15
44
45 # ---------------------------------------------------------------------------
46 # P1xx — Unit tests (pure helpers, no DB)
47 # ---------------------------------------------------------------------------
48
49
50 class TestHistoryCursorParsing:
51 """P101–P104: history cursor encode/decode round-trips."""
52
53 def _encode(self, iso: str) -> str:
54 # The cursor IS the ISO timestamp of the last entry on the current page.
55 return iso
56
57 def _decode(self, cursor: str) -> _dt.datetime:
58 return _dt.datetime.fromisoformat(cursor)
59
60 def test_P101_roundtrip_utc(self) -> None:
61 """P101: ISO cursor survives encode→decode for a UTC timestamp."""
62 ts = "2026-01-15T10:30:00+00:00"
63 assert self._decode(self._encode(ts)).isoformat() == _dt.datetime.fromisoformat(ts).isoformat()
64
65 def test_P102_cursor_is_comparable(self) -> None:
66 """P102: decoded cursor compares correctly to committed_at values."""
67 cursor_ts = _dt.datetime.fromisoformat("2026-01-15T10:30:00+00:00")
68 older = _dt.datetime.fromisoformat("2026-01-10T00:00:00+00:00")
69 newer = _dt.datetime.fromisoformat("2026-01-20T00:00:00+00:00")
70 assert older < cursor_ts
71 assert newer > cursor_ts
72
73 def test_P103_page_size_constant_is_10(self) -> None:
74 """P103: HISTORY_PAGE == 10."""
75 assert HISTORY_PAGE == 10
76
77 def test_P104_coupling_page_size_constant_is_15(self) -> None:
78 """P104: COUPLING_PAGE == 15."""
79 assert COUPLING_PAGE == 15
80
81
82 class TestCouplingCursorParsing:
83 """P105–P107: coupling cursor encode/decode."""
84
85 def _encode(self, shared: int, address: str) -> str:
86 return f"{shared}:{address}"
87
88 def _decode(self, cursor: str) -> tuple[int, str]:
89 shared_str, _, addr = cursor.partition(":")
90 return int(shared_str), addr
91
92 def test_P105_roundtrip(self) -> None:
93 """P105: coupling cursor survives encode→decode."""
94 shared, addr = 7, "src/auth.py::validate_token"
95 cursor = self._encode(shared, addr)
96 s, a = self._decode(cursor)
97 assert s == shared and a == addr
98
99 def test_P106_address_with_colons_preserved(self) -> None:
100 """P106: address containing '::' is preserved through cursor."""
101 shared, addr = 3, "lib/core.py::MyClass::method"
102 cursor = self._encode(shared, addr)
103 s, a = self._decode(cursor)
104 assert s == shared and a == addr
105
106 def test_P107_zero_shared_valid(self) -> None:
107 """P107: shared_commits == 0 is a valid cursor value."""
108 cursor = self._encode(0, "src/x.py::fn")
109 s, a = self._decode(cursor)
110 assert s == 0 and a == "src/x.py::fn"
111
112
113 class TestPageSlicing:
114 """P108–P112: page-slice logic on in-memory lists."""
115
116 @staticmethod
117 def _slice(items: Sequence[int], page: int, cursor_idx: int | None = None) -> tuple[list[int], bool, str | None]:
118 """Simulate cursor pagination over a pre-sorted list."""
119 start = cursor_idx + 1 if cursor_idx is not None else 0
120 window = items[start : start + page + 1]
121 has_next = len(window) > page
122 page_items = window[:page]
123 next_cursor = str(start + page - 1) if has_next else None
124 return page_items, has_next, next_cursor
125
126 def test_P108_first_page_returns_page_items(self) -> None:
127 """P108: first page of 25 items returns exactly HISTORY_PAGE items."""
128 items = list(range(25))
129 page, has_next, cursor = self._slice(items, HISTORY_PAGE)
130 assert len(page) == HISTORY_PAGE
131 assert has_next is True
132 assert cursor is not None
133
134 def test_P109_last_page_has_no_next(self) -> None:
135 """P109: last page has has_next=False and cursor=None."""
136 items = list(range(12)) # 12 items, page=10 → second page has 2
137 page, has_next, cursor = self._slice(items, HISTORY_PAGE, cursor_idx=9)
138 assert has_next is False
139 assert cursor is None
140
141 def test_P110_exact_fit_no_next(self) -> None:
142 """P110: exactly HISTORY_PAGE items → has_next=False."""
143 items = list(range(HISTORY_PAGE))
144 page, has_next, _ = self._slice(items, HISTORY_PAGE)
145 assert len(page) == HISTORY_PAGE
146 assert has_next is False
147
148 def test_P111_empty_set_returns_empty(self) -> None:
149 """P111: zero items → empty page, has_next=False."""
150 page, has_next, cursor = self._slice([], HISTORY_PAGE)
151 assert page == []
152 assert has_next is False
153 assert cursor is None
154
155 def test_P112_coupling_page_size_15(self) -> None:
156 """P112: COUPLING_PAGE slices produce at most 15 items."""
157 items = list(range(40))
158 page, has_next, cursor = self._slice(items, COUPLING_PAGE)
159 assert len(page) == COUPLING_PAGE
160 assert has_next is True
161
162
163 # ---------------------------------------------------------------------------
164 # P2xx — Integration tests (real DB, route handler)
165 # ---------------------------------------------------------------------------
166
167
168 @pytest.mark.asyncio
169 class TestProvenancePagination:
170 """P201–P208: provenance history pagination via history_cursor query param."""
171
172 async def test_P201_first_page_returns_10_entries(
173 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
174 ) -> None:
175 """P201: no cursor → exactly 10 history entries in HTML."""
176 owner, slug, address = seed_symbol_with_26_history
177 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
178 assert resp.status_code == 200
179 count = resp.content.count(b"sym2-tl-entry")
180 assert count == HISTORY_PAGE
181
182 async def test_P202_first_page_has_next_link(
183 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
184 ) -> None:
185 """P202: history_cursor link present when more entries exist."""
186 owner, slug, address = seed_symbol_with_26_history
187 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
188 assert b"history_cursor=" in resp.content
189
190 async def test_P203_last_page_no_next_link(
191 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
192 ) -> None:
193 """P203: final history page (offset 20) has 6 entries and no next link."""
194 owner, slug, address = seed_symbol_with_26_history
195 import re
196 # Page 2 (offset 10) → 10 entries
197 r2 = await client.get(f"/{owner}/{slug}/symbol/{address}?history_cursor=10")
198 assert r2.status_code == 200
199 assert r2.content.count(b"sym2-tl-entry") == HISTORY_PAGE
200 # Page 3 (offset 20) → 6 remaining, no --next link
201 r3 = await client.get(f"/{owner}/{slug}/symbol/{address}?history_cursor=20")
202 assert r3.status_code == 200
203 assert r3.content.count(b"sym2-tl-entry") == 6
204 assert not re.search(rb'sym2-page-btn--next', r3.content)
205
206 async def test_P204_no_cursor_shows_newest_first(
207 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
208 ) -> None:
209 """P204: first page shows most recent 10 entries (newest → oldest)."""
210 owner, slug, address = seed_symbol_with_26_history
211 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
212 # The newest commit message contains 'entry-25' (seeded newest-last, displayed newest-first)
213 assert b"entry-25" in resp.content
214 assert b"entry-0" not in resp.content
215
216 async def test_P205_cursor_page_does_not_overlap_previous(
217 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
218 ) -> None:
219 """P205: page 2 entries are disjoint from page 1."""
220 owner, slug, address = seed_symbol_with_26_history
221 import re
222 r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
223 m = re.search(rb'history_cursor=([^"&]+)', r1.content)
224 cursor = m.group(1).decode()
225 r2 = await client.get(f"/{owner}/{slug}/symbol/{address}?history_cursor={cursor}")
226 # entry-15 should be on page 2, not page 1
227 assert b"entry-15" not in r1.content
228 assert b"entry-15" in r2.content
229
230 async def test_P206_10_or_fewer_entries_no_pagination(
231 self, client: AsyncClient, seed_symbol: tuple[str, str, str]
232 ) -> None:
233 """P206: symbol with 1 entry shows no history_cursor link."""
234 owner, slug, address = seed_symbol
235 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
236 assert resp.status_code == 200
237 assert b"history_cursor=" not in resp.content
238
239 async def test_P207_history_total_count_in_context(
240 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
241 ) -> None:
242 """P207: page renders total provenance count for 'showing X of N' display."""
243 owner, slug, address = seed_symbol_with_26_history
244 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
245 # change_count drives the vitals quad and narrative
246 assert b"26" in resp.content
247
248 async def test_P208_invalid_history_cursor_returns_200_first_page(
249 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
250 ) -> None:
251 """P208: malformed cursor falls back to first page gracefully."""
252 owner, slug, address = seed_symbol_with_26_history
253 resp = await client.get(
254 f"/{owner}/{slug}/symbol/{address}?history_cursor=not-a-date"
255 )
256 assert resp.status_code == 200
257 # Should render first page normally
258 assert resp.content.count(b"sym2-tl-entry") == HISTORY_PAGE
259
260
261 @pytest.mark.asyncio
262 class TestCouplingPagination:
263 """P211–P218: coupling partners pagination via coupling_cursor query param."""
264
265 async def test_P211_first_page_returns_15_partners(
266 self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
267 ) -> None:
268 """P211: no cursor → exactly 15 coupling rows."""
269 owner, slug, address = seed_symbol_high_coupling_40
270 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
271 assert resp.status_code == 200
272 assert resp.content.count(b"sym2-blast-row") == COUPLING_PAGE
273
274 async def test_P212_first_page_has_coupling_next_link(
275 self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
276 ) -> None:
277 """P212: coupling_cursor link present when more partners exist."""
278 owner, slug, address = seed_symbol_high_coupling_40
279 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
280 assert b"coupling_cursor=" in resp.content
281
282 async def test_P213_second_page_returns_remaining(
283 self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
284 ) -> None:
285 """P213: page 2 returns the remaining 25 partners (capped at 15)."""
286 owner, slug, address = seed_symbol_high_coupling_40
287 import re
288 r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
289 m = re.search(rb'coupling_cursor=([^"&]+)', r1.content)
290 assert m
291 cursor = m.group(1).decode()
292 r2 = await client.get(f"/{owner}/{slug}/symbol/{address}?coupling_cursor={cursor}")
293 assert r2.status_code == 200
294 assert r2.content.count(b"sym2-blast-row") == COUPLING_PAGE
295
296 async def test_P214_last_coupling_page_no_next_link(
297 self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
298 ) -> None:
299 """P214: final coupling page has no coupling_cursor link."""
300 owner, slug, address = seed_symbol_high_coupling_40
301 import re
302 # 40 partners: p1=15 (offset 0), p2=15 (offset 15), p3=10 (offset 30)
303 r3 = await client.get(f"/{owner}/{slug}/symbol/{address}?coupling_cursor=30")
304 assert r3.status_code == 200
305 assert r3.content.count(b"sym2-blast-row") == 10
306 # Coupling section has no "Next ›" link (history may still have "Older ›")
307 assert b"Next \xe2\x80\xba" not in r3.content
308
309 async def test_P215_fewer_than_15_partners_no_pagination(
310 self, client: AsyncClient, seed_symbol: tuple[str, str, str]
311 ) -> None:
312 """P215: symbol with 0 coupling partners shows no coupling_cursor link."""
313 owner, slug, address = seed_symbol
314 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
315 assert b"coupling_cursor=" not in resp.content
316
317 async def test_P216_both_cursors_independent(
318 self, client: AsyncClient, seed_symbol_with_26_history_and_40_coupling: tuple[str, str, str]
319 ) -> None:
320 """P216: history_cursor and coupling_cursor paginate independently."""
321 owner, slug, address = seed_symbol_with_26_history_and_40_coupling
322 import re
323 r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
324 hc = re.search(rb'history_cursor=([^"&]+)', r1.content)
325 cc = re.search(rb'coupling_cursor=([^"&]+)', r1.content)
326 assert hc and cc
327 # Advance only history cursor
328 r2 = await client.get(
329 f"/{owner}/{slug}/symbol/{address}"
330 f"?history_cursor={hc.group(1).decode()}"
331 )
332 assert r2.status_code == 200
333 assert r2.content.count(b"sym2-blast-row") == COUPLING_PAGE
334
335 async def test_P217_invalid_coupling_cursor_returns_200_first_page(
336 self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
337 ) -> None:
338 """P217: malformed coupling cursor falls back to first page."""
339 owner, slug, address = seed_symbol_high_coupling_40
340 resp = await client.get(
341 f"/{owner}/{slug}/symbol/{address}?coupling_cursor=garbage"
342 )
343 assert resp.status_code == 200
344 assert resp.content.count(b"sym2-blast-row") == COUPLING_PAGE
345
346 async def test_P218_coupling_ordered_by_shared_commits_desc(
347 self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
348 ) -> None:
349 """P218: coupling page 1 contains highest-shared partners first."""
350 owner, slug, address = seed_symbol_high_coupling_40
351 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
352 # The fixture seeds partners with shared counts 40..1; highest first
353 assert b"40\xc3\x97" in resp.content or b"40" in resp.content # top partner count
354
355
356 # ---------------------------------------------------------------------------
357 # P3xx — HTML structural tests
358 # ---------------------------------------------------------------------------
359
360
361 @pytest.mark.asyncio
362 class TestPaginationHTML:
363 """P301–P306: template renders pagination controls correctly."""
364
365 async def test_P301_next_history_link_is_anchor(
366 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
367 ) -> None:
368 """P301: 'Load more' history link is an <a> tag with history_cursor param."""
369 owner, slug, address = seed_symbol_with_26_history
370 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
371 assert b'href=' in resp.content
372 assert b'history_cursor=' in resp.content
373
374 async def test_P302_next_coupling_link_is_anchor(
375 self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
376 ) -> None:
377 """P302: 'Load more' coupling link is an <a> tag with coupling_cursor param."""
378 owner, slug, address = seed_symbol_high_coupling_40
379 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
380 assert b'href=' in resp.content
381 assert b'coupling_cursor=' in resp.content
382
383 async def test_P303_page_indicator_present(
384 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
385 ) -> None:
386 """P303: provenance section shows current count and total."""
387 owner, slug, address = seed_symbol_with_26_history
388 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
389 # e.g. "10 of 26" or "Showing 10"
390 assert b"10" in resp.content and b"26" in resp.content
391
392 async def test_P304_no_controls_when_single_page(self, client: AsyncClient, seed_symbol: tuple[str, str, str]) -> None:
393 """P304: no pagination controls rendered for single-page symbol."""
394 owner, slug, address = seed_symbol
395 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
396 assert b"history_cursor=" not in resp.content
397 assert b"coupling_cursor=" not in resp.content
398
399 async def test_P305_cursor_preserved_in_coupling_next_link(
400 self, client: AsyncClient, seed_symbol_with_26_history_and_40_coupling: tuple[str, str, str]
401 ) -> None:
402 """P305: coupling next link preserves active history_cursor in href."""
403 owner, slug, address = seed_symbol_with_26_history_and_40_coupling
404 import re
405 r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
406 hc = re.search(rb'history_cursor=([^"&]+)', r1.content).group(1).decode()
407 # Navigate to history page 2 while on coupling page 1
408 r2 = await client.get(
409 f"/{owner}/{slug}/symbol/{address}?history_cursor={hc}"
410 )
411 assert r2.status_code == 200
412 # coupling next link in r2 must carry history_cursor forward
413 cc_links = [m.group() for m in re.finditer(rb'href="[^"]*coupling_cursor=[^"]*"', r2.content)]
414 assert any(hc.encode() in link for link in cc_links)
415
416 async def test_P306_history_next_link_preserves_coupling_cursor(
417 self, client: AsyncClient, seed_symbol_with_26_history_and_40_coupling: tuple[str, str, str]
418 ) -> None:
419 """P306: history next link preserves active coupling_cursor in href."""
420 owner, slug, address = seed_symbol_with_26_history_and_40_coupling
421 import re
422 r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
423 cc = re.search(rb'coupling_cursor=([^"&]+)', r1.content).group(1).decode()
424 r2 = await client.get(
425 f"/{owner}/{slug}/symbol/{address}?coupling_cursor={cc}"
426 )
427 assert r2.status_code == 200
428 hc_links = [m.group() for m in re.finditer(rb'href="[^"]*history_cursor=[^"]*"', r2.content)]
429 assert any(cc.encode() in link for link in hc_links)
430
431
432 # ---------------------------------------------------------------------------
433 # P4xx — Edge cases
434 # ---------------------------------------------------------------------------
435
436
437 @pytest.mark.asyncio
438 class TestPaginationEdgeCases:
439 """P401–P406: boundary conditions."""
440
441 async def test_P401_exactly_10_history_no_next(self, client: AsyncClient, seed_symbol_with_exactly_10_history: tuple[str, str, str]) -> None:
442 """P401: exactly 10 history entries → no history_cursor link."""
443 owner, slug, address = seed_symbol_with_exactly_10_history
444 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
445 assert resp.status_code == 200
446 assert resp.content.count(b"sym2-tl-entry") == HISTORY_PAGE
447 assert b"history_cursor=" not in resp.content
448
449 async def test_P402_exactly_11_history_has_next(self, client: AsyncClient, seed_symbol_with_11_history: tuple[str, str, str]) -> None:
450 """P402: 11 history entries → first page has next link."""
451 owner, slug, address = seed_symbol_with_11_history
452 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
453 assert resp.content.count(b"sym2-tl-entry") == HISTORY_PAGE
454 assert b"history_cursor=" in resp.content
455
456 async def test_P403_exactly_15_coupling_no_next(self, client: AsyncClient, seed_symbol_with_exactly_15_coupling: tuple[str, str, str]) -> None:
457 """P403: exactly 15 coupling partners → no coupling_cursor link."""
458 owner, slug, address = seed_symbol_with_exactly_15_coupling
459 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
460 assert resp.content.count(b"sym2-blast-row") == COUPLING_PAGE
461 assert b"coupling_cursor=" not in resp.content
462
463 async def test_P404_exactly_16_coupling_has_next(self, client: AsyncClient, seed_symbol_with_16_coupling: tuple[str, str, str]) -> None:
464 """P404: 16 coupling partners → first page has next link."""
465 owner, slug, address = seed_symbol_with_16_coupling
466 resp = await client.get(f"/{owner}/{slug}/symbol/{address}")
467 assert resp.content.count(b"sym2-blast-row") == COUPLING_PAGE
468 assert b"coupling_cursor=" in resp.content
469
470 async def test_P405_past_cursor_returns_empty_page(
471 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
472 ) -> None:
473 """P405: offset past the end of history returns empty timeline."""
474 owner, slug, address = seed_symbol_with_26_history
475 resp = await client.get(
476 f"/{owner}/{slug}/symbol/{address}?history_cursor=10000"
477 )
478 assert resp.status_code == 200
479 assert resp.content.count(b"sym2-tl-entry") == 0
480
481 async def test_P406_both_cursors_on_last_pages_no_links(
482 self, client: AsyncClient, seed_symbol_with_26_history_and_40_coupling: tuple[str, str, str]
483 ) -> None:
484 """P406: when both paginations are on final pages, no cursor links appear.
485
486 Fixture: 26 history entries (3 pages: 10+10+6) + 26 coupling partners
487 (2 pages: 15+11). Exhaust both to confirm no pagination links remain.
488 """
489 owner, slug, address = seed_symbol_with_26_history_and_40_coupling
490 import re
491 # Page 1: has both cursors
492 r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
493 # Fixture: 26 history (3 pages: 0,10,20) + 26 coupling (2 pages: 0,15)
494 # Final request: history page 3 (last) + coupling page 2 (last)
495 r_final = await client.get(
496 f"/{owner}/{slug}/symbol/{address}?history_cursor=20&coupling_cursor=15"
497 )
498 assert r_final.status_code == 200
499 # Neither section has a next button
500 assert b"Older \xe2\x80\xba" not in r_final.content # no history next
501 assert b"Next \xe2\x80\xba" not in r_final.content # no coupling next
502
503
504 # ---------------------------------------------------------------------------
505 # P5xx — Performance
506 # ---------------------------------------------------------------------------
507
508
509 @pytest.mark.asyncio
510 class TestPaginationPerformance:
511 """P501–P503: pagination does not increase DB round-trips."""
512
513 async def test_P501_history_page2_same_query_count_as_page1(
514 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str], db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
515 ) -> None:
516 """P501: paginated request makes no more DB calls than page 1."""
517 import re
518 r1 = await client.get(
519 f"/{seed_symbol_with_26_history[0]}/{seed_symbol_with_26_history[1]}"
520 f"/symbol/{seed_symbol_with_26_history[2]}"
521 )
522 hc = re.search(rb'history_cursor=([^"&]+)', r1.content).group(1).decode()
523
524 call_counts: list[int] = []
525 for cursor in [None, hc]:
526 count = {"n": 0}
527 orig = db_session.execute
528 async def spy(*a: Executable | typing.Any, _c: MutableMapping[str, int] = count, _o: typing.Any = orig, **kw: typing.Any) -> CursorResult[typing.Any]:
529 _c["n"] += 1
530 return await _o(*a, **kw)
531 monkeypatch.setattr(db_session, "execute", spy)
532 url = (f"/{seed_symbol_with_26_history[0]}/{seed_symbol_with_26_history[1]}"
533 f"/symbol/{seed_symbol_with_26_history[2]}")
534 if cursor:
535 url += f"?history_cursor={cursor}"
536 await client.get(url)
537 call_counts.append(count["n"])
538 monkeypatch.undo()
539
540 assert call_counts[1] <= call_counts[0] + 1 # at most 1 extra call
541
542 async def test_P502_coupling_page_uses_offset_not_python_scan(self) -> None:
543 """P502: coupling cursor pagination uses SQL WHERE, not Python list slice."""
544 from sqlalchemy import select, func as sa_func
545 from musehub.db.musehub_intel_models import MusehubSymbolHistoryEntry
546 # Verify that the query accepts a LIMIT and the architecture allows WHERE for cursor
547 stmt = (
548 select(
549 MusehubSymbolHistoryEntry.address,
550 sa_func.count().label("shared"),
551 )
552 .where(MusehubSymbolHistoryEntry.repo_id == "x")
553 .group_by(MusehubSymbolHistoryEntry.address)
554 .order_by(sa_func.count().desc())
555 .limit(COUPLING_PAGE + 1)
556 )
557 compiled = str(stmt.compile(compile_kwargs={"literal_binds": False}))
558 assert "LIMIT" in compiled.upper()
559 assert "GROUP BY" in compiled.upper()
560
561 async def test_P503_render_time_under_300ms(
562 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str], benchmark_timer: typing.Callable[[float], typing.ContextManager[None]]
563 ) -> None:
564 """P503: paginated page renders in < 300ms."""
565 owner, slug, address = seed_symbol_with_26_history
566 import re
567 r1 = await client.get(f"/{owner}/{slug}/symbol/{address}")
568 hc = re.search(rb'history_cursor=([^"&]+)', r1.content).group(1).decode()
569 with benchmark_timer(max_ms=300):
570 resp = await client.get(
571 f"/{owner}/{slug}/symbol/{address}?history_cursor={hc}"
572 )
573 assert resp.status_code == 200
574
575
576 # ---------------------------------------------------------------------------
577 # P6xx — Security
578 # ---------------------------------------------------------------------------
579
580
581 @pytest.mark.asyncio
582 class TestPaginationSecurity:
583 """P601–P604: cursor values cannot leak data or cause server errors."""
584
585 async def test_P601_sql_injection_in_history_cursor(
586 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
587 ) -> None:
588 """P601: SQL injection in history_cursor param → 200 first page, no 500."""
589 owner, slug, address = seed_symbol_with_26_history
590 resp = await client.get(
591 f"/{owner}/{slug}/symbol/{address}"
592 "?history_cursor='; DROP TABLE musehub_symbol_history_entries; --"
593 )
594 assert resp.status_code == 200
595
596 async def test_P602_sql_injection_in_coupling_cursor(
597 self, client: AsyncClient, seed_symbol_high_coupling_40: tuple[str, str, str]
598 ) -> None:
599 """P602: SQL injection in coupling_cursor param → 200 first page, no 500."""
600 owner, slug, address = seed_symbol_high_coupling_40
601 resp = await client.get(
602 f"/{owner}/{slug}/symbol/{address}"
603 "?coupling_cursor=0:x' OR '1'='1"
604 )
605 assert resp.status_code == 200
606
607 async def test_P603_xss_in_history_cursor_escaped(
608 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
609 ) -> None:
610 """P603: XSS payload in history_cursor is never reflected raw in HTML."""
611 owner, slug, address = seed_symbol_with_26_history
612 resp = await client.get(
613 f"/{owner}/{slug}/symbol/{address}"
614 "?history_cursor=<script>alert(1)</script>"
615 )
616 assert b"<script>alert(1)</script>" not in resp.content
617
618 async def test_P604_very_long_cursor_no_500(
619 self, client: AsyncClient, seed_symbol_with_26_history: tuple[str, str, str]
620 ) -> None:
621 """P604: cursor > 512 chars returns 200 (graceful fallback) never 500."""
622 owner, slug, address = seed_symbol_with_26_history
623 long_cursor = "x" * 600
624 resp = await client.get(
625 f"/{owner}/{slug}/symbol/{address}?history_cursor={long_cursor}"
626 )
627 assert resp.status_code == 200
628
629
630 # ---------------------------------------------------------------------------
631 # New fixtures (appended to conftest.py separately)
632 # ---------------------------------------------------------------------------
633 # The fixtures below are declared here as documentation of what conftest.py
634 # must provide. They are moved to conftest.py by the implementation step.
635
636 """
637 Required new fixtures:
638
639 seed_symbol_with_26_history
640 → repo + symbol with 26 history entries spaced 1h apart, newest last.
641 Commit messages contain 'entry-{i}' for i in 0..25.
642
643 seed_symbol_high_coupling_40
644 → repo + symbol with 1 history entry + 40 partner symbols each sharing
645 that 1 commit, with shared_commits values 40..1 (descending by address).
646
647 seed_symbol_with_26_history_and_40_coupling
648 → combines both: 26 history entries + 40 coupling partners.
649
650 seed_symbol_with_exactly_10_history
651 → repo + symbol with exactly 10 history entries.
652
653 seed_symbol_with_11_history
654 → repo + symbol with exactly 11 history entries.
655
656 seed_symbol_with_exactly_15_coupling
657 → repo + symbol with 15 coupling partners.
658
659 seed_symbol_with_16_coupling
660 → repo + symbol with 16 coupling partners.
661 """
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 22 days ago