gabriel / musehub public
test_phase2_stable_route.py python
493 lines 19.4 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 21 days ago
1 """TDD spec for Phase 2 — /intel/stable list page route (issue #12).
2
3 Route:
4 GET /{owner}/{repo_slug}/intel/stable
5 ?since_start=true — filter to eternal-only symbols
6 ?top=N — limit rows (25 / 50 / 100)
7
8 Returns 200 with stat row (eternal, veteran, total, oldest) and a
9 days_stable-DESC ranked list. Empty state when no data exists.
10
11 Seven test tiers
12 ----------------
13 Unit P2_01 Route registered
14 Integration P2_02 – P2_08 HTTP responses, filters, stat counts, HTML content
15 E2E P2_09 – P2_11 Full seed → HTML round-trip
16 Stress P2_12 – P2_13 200-row render, top=100 limit
17 Data Integrity P2_14 – P2_15 Repo isolation, sort order
18 Performance P2_16 – P2_17 Response time, ordering
19 Security P2_18 – P2_20 XSS, path traversal, IDOR
20 """
21 from __future__ import annotations
22
23 import secrets
24 from datetime import datetime, timedelta, timezone
25
26 import pytest
27 import pytest_asyncio
28 from httpx import AsyncClient
29 from sqlalchemy.dialects.postgresql import insert as pg_insert
30 from sqlalchemy.ext.asyncio import AsyncSession
31
32 from muse.core.types import fake_id, long_id
33 from musehub.db.musehub_intel_models import MusehubIntelStable
34 from musehub.db.musehub_repo_models import MusehubRepo
35 from tests.factories import create_repo
36
37
38 def _uid() -> str:
39 return fake_id(secrets.token_hex(16))
40
41
42 _OWNER = "testuser"
43 _SLUG = "stableroute"
44 _REF = long_id("c" * 64)
45
46
47 async def _seed_stable(
48 session: AsyncSession,
49 repo_id: str,
50 *,
51 address: str,
52 days_stable: int = 180,
53 since_start: bool = False,
54 last_changed_commit: str | None = None,
55 symbol_kind: str | None = None,
56 ) -> None:
57 """Insert a ``musehub_intel_stable`` row for test fixtures.
58
59 Parameters
60 ----------
61 session: Active async session.
62 repo_id: Target repository ID.
63 address: Symbol address (``file.py::fn``).
64 days_stable: Days since last modification.
65 since_start: True when symbol has never been modified.
66 last_changed_commit: Commit ID of last modification.
67 symbol_kind: Symbol kind (function, method, class, async_method).
68 """
69 stmt = (
70 pg_insert(MusehubIntelStable)
71 .values(
72 repo_id=repo_id,
73 address=address,
74 days_stable=days_stable,
75 since_start=since_start,
76 last_changed_commit=last_changed_commit,
77 symbol_kind=symbol_kind,
78 ref=_REF,
79 )
80 .on_conflict_do_update(
81 index_elements=["repo_id", "address"],
82 set_={"days_stable": days_stable, "since_start": since_start,
83 "symbol_kind": symbol_kind},
84 )
85 )
86 await session.execute(stmt)
87 await session.flush()
88
89
90 # ---------------------------------------------------------------------------
91 # Fixtures
92 # ---------------------------------------------------------------------------
93
94 @pytest_asyncio.fixture
95 async def route_repo(db_session: AsyncSession) -> MusehubRepo:
96 """Bare repo — no stable rows."""
97 return await create_repo(db_session, owner=_OWNER, slug=_SLUG)
98
99
100 @pytest_asyncio.fixture
101 async def route_repo_with_symbols(db_session: AsyncSession, route_repo: MusehubRepo) -> MusehubRepo:
102 """Repo with a varied set of stable symbols."""
103 repo_id = route_repo.repo_id
104 await db_session.commit()
105 await _seed_stable(db_session, repo_id, address="pkg/core.py::parse",
106 days_stable=400, since_start=False,
107 last_changed_commit=_REF, symbol_kind="function")
108 await _seed_stable(db_session, repo_id, address="pkg/codec.py::pack",
109 days_stable=900, since_start=True,
110 last_changed_commit=None, symbol_kind="function")
111 await _seed_stable(db_session, repo_id, address="pkg/utils.py::sha256",
112 days_stable=200, since_start=False,
113 last_changed_commit=_REF, symbol_kind="method")
114 await db_session.commit()
115 return route_repo
116
117
118 # ---------------------------------------------------------------------------
119 # Tier 1 — Unit
120 # ---------------------------------------------------------------------------
121
122 class TestStableRouteUnit:
123 """Unit test — route registration."""
124
125 def test_P2_01_route_registered(self) -> None:
126 """intel/stable route must be registered in the ui_intel router."""
127 from musehub.api.routes.musehub.ui_intel import router
128 paths = [r.path for r in router.routes]
129 assert any("intel/stable" in p for p in paths)
130
131
132 # ---------------------------------------------------------------------------
133 # Tier 2 — Integration
134 # ---------------------------------------------------------------------------
135
136 class TestStableRouteIntegration:
137 """Integration tests — HTTP responses against the real DB."""
138
139 @pytest.mark.asyncio
140 async def test_P2_02_empty_state_returns_200(
141 self, client: AsyncClient, route_repo: MusehubRepo
142 ) -> None:
143 """Empty stable table → 200 with empty-state message."""
144 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
145 assert resp.status_code == 200
146
147 @pytest.mark.asyncio
148 async def test_P2_03_populated_returns_200(
149 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
150 ) -> None:
151 """Populated repo → 200."""
152 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
153 assert resp.status_code == 200
154
155 @pytest.mark.asyncio
156 async def test_P2_04_since_start_filter(
157 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
158 ) -> None:
159 """?since_start=true shows only eternal symbols."""
160 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable", params={"since_start": "true"})
161 assert resp.status_code == 200
162 # Only pkg/codec.py::pack has since_start=True
163 assert "pack" in resp.text
164 assert "parse" not in resp.text
165
166 @pytest.mark.asyncio
167 async def test_P2_05_top_param_limits_rows(
168 self, client: AsyncClient, db_session: AsyncSession, route_repo: MusehubRepo
169 ) -> None:
170 """?top=2 returns at most 2 symbol rows."""
171 repo_id = route_repo.repo_id
172 await db_session.commit()
173 for i in range(5):
174 await _seed_stable(db_session, repo_id,
175 address=f"pkg/top{i}.py::fn",
176 days_stable=100 + i)
177 await db_session.commit()
178 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable", params={"top": 25})
179 assert resp.status_code == 200
180
181 @pytest.mark.asyncio
182 async def test_P2_06_stat_eternal_count_correct(
183 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
184 ) -> None:
185 """Eternal count stat reflects since_start=True rows only."""
186 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
187 # 1 eternal symbol (pkg/codec.py::pack)
188 assert "1" in resp.text
189
190 @pytest.mark.asyncio
191 async def test_P2_07_address_in_html(
192 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
193 ) -> None:
194 """Symbol address fragments appear in the rendered HTML."""
195 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
196 assert "parse" in resp.text
197 assert "pack" in resp.text
198
199 @pytest.mark.asyncio
200 async def test_P2_08_days_stable_value_in_html(
201 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
202 ) -> None:
203 """days_stable value (400d) appears in the HTML output."""
204 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
205 assert "400" in resp.text
206
207 @pytest.mark.asyncio
208 async def test_P2_08b_kind_filter_restricts_rows(
209 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
210 ) -> None:
211 """?kind=method shows only method symbols (sha256), not function symbols."""
212 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable", params={"kind": "method"})
213 assert resp.status_code == 200
214 # sha256 is a method; parse and pack are functions — their file paths must not appear
215 assert "sha256" in resp.text
216 assert "pkg/core.py" not in resp.text # parse is function
217 assert "pkg/codec.py" not in resp.text # pack is function
218
219 @pytest.mark.asyncio
220 async def test_P2_08c_invalid_kind_returns_all(
221 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
222 ) -> None:
223 """?kind=bogus (invalid) is silently ignored — all rows returned."""
224 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable", params={"kind": "bogus"})
225 assert resp.status_code == 200
226 assert "parse" in resp.text
227 assert "pkg/codec.py" in resp.text # codec.py::pack is unique
228
229 @pytest.mark.asyncio
230 async def test_P2_08d_symbol_kind_appears_in_html(
231 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
232 ) -> None:
233 """symbol_kind values (function, method) appear in the rendered HTML."""
234 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
235 assert "function" in resp.text
236 assert "method" in resp.text
237
238
239 # ---------------------------------------------------------------------------
240 # Tier 3 — E2E
241 # ---------------------------------------------------------------------------
242
243 class TestStableRouteE2E:
244 """End-to-end tests — full seed-to-HTML round-trip."""
245
246 @pytest.mark.asyncio
247 async def test_P2_09_full_round_trip(
248 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
249 ) -> None:
250 """Seeded symbols appear in HTML with days and address."""
251 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
252 text = resp.text
253 assert "pack" in text
254 assert "900" in text
255
256 @pytest.mark.asyncio
257 async def test_P2_10_stat_row_rendered(
258 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
259 ) -> None:
260 """Stat row keywords (Eternal, Veteran, Total, Oldest) present."""
261 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
262 text = resp.text
263 assert "Eternal" in text
264 assert "Total" in text
265
266 @pytest.mark.asyncio
267 async def test_P2_11_filter_url_contains_top(
268 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
269 ) -> None:
270 """Filter bar links contain ?top= parameter."""
271 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
272 assert "top=" in resp.text
273
274
275 # ---------------------------------------------------------------------------
276 # Tier 4 — Stress
277 # ---------------------------------------------------------------------------
278
279 class TestStableRouteStress:
280 """Stress tests — large row counts render without error."""
281
282 @pytest.mark.asyncio
283 async def test_P2_12_200_rows_render_without_error(
284 self, client: AsyncClient, db_session: AsyncSession, route_repo: MusehubRepo
285 ) -> None:
286 """200 stable rows render to a 200 response without error."""
287 repo_id = route_repo.repo_id
288 await db_session.commit()
289 for i in range(200):
290 await _seed_stable(db_session, repo_id,
291 address=f"pkg/stress{i}.py::fn",
292 days_stable=100 + i)
293 await db_session.commit()
294 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable", params={"top": 100})
295 assert resp.status_code == 200
296
297 @pytest.mark.asyncio
298 async def test_P2_13_top_100_returns_correct_count(
299 self, client: AsyncClient, db_session: AsyncSession, route_repo: MusehubRepo
300 ) -> None:
301 """?top=100 with 150 rows renders exactly 100 symbol addresses."""
302 repo_id = route_repo.repo_id
303 await db_session.commit()
304 for i in range(150):
305 await _seed_stable(db_session, repo_id,
306 address=f"pkg/count{i}.py::fn",
307 days_stable=50 + i)
308 await db_session.commit()
309 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable", params={"top": 100})
310 # 100 rows × 2 occurrences each (title attr + symbol href)
311 assert resp.text.count("::fn") == 200
312
313
314 # ---------------------------------------------------------------------------
315 # Tier 5 — Data Integrity
316 # ---------------------------------------------------------------------------
317
318 class TestStableRouteDataIntegrity:
319 """Data integrity tests — repo isolation and sort order."""
320
321 @pytest.mark.asyncio
322 async def test_P2_14_only_this_repo_rows_returned(
323 self, client: AsyncClient, db_session: AsyncSession, route_repo: MusehubRepo
324 ) -> None:
325 """Rows from a different repo are not shown on this repo's page."""
326 repo_b = await create_repo(db_session, owner=_OWNER, slug="stableroute_b")
327 repo_id_b = repo_b.repo_id
328 await db_session.commit()
329 await _seed_stable(db_session, repo_id_b,
330 address="pkg/other.py::secret_fn",
331 days_stable=500)
332 await db_session.commit()
333 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
334 assert "secret_fn" not in resp.text
335
336 @pytest.mark.asyncio
337 async def test_P2_15_sort_order_days_stable_desc(
338 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
339 ) -> None:
340 """Rows are ordered by days_stable DESC — highest first."""
341 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
342 text = resp.text
343 # pack (900d) must appear before parse (400d) in the HTML
344 assert text.index("pack") < text.index("parse")
345
346
347 # ---------------------------------------------------------------------------
348 # Tier 6 — Performance
349 # ---------------------------------------------------------------------------
350
351 class TestStableRoutePerformance:
352 """Performance tests — response time and sort correctness at scale."""
353
354 @pytest.mark.asyncio
355 async def test_P2_16_response_under_500ms(
356 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
357 ) -> None:
358 """Response time for 3 rows < 500ms."""
359 import time
360 start = time.monotonic()
361 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
362 elapsed = time.monotonic() - start
363 assert resp.status_code == 200
364 assert elapsed < 0.5
365
366 @pytest.mark.asyncio
367 async def test_P2_17_invalid_top_falls_back_to_default(
368 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
369 ) -> None:
370 """?top=999 (invalid) silently falls back to default=50, returns 200."""
371 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable", params={"top": 999})
372 assert resp.status_code == 200
373
374
375 # ---------------------------------------------------------------------------
376 # Tier 7 — Security
377 # ---------------------------------------------------------------------------
378
379 class TestStableRouteSecurity:
380 """Security tests — XSS, path traversal, IDOR."""
381
382 @pytest.mark.asyncio
383 async def test_P2_18_xss_in_address_escaped(
384 self, client: AsyncClient, db_session: AsyncSession, route_repo: MusehubRepo
385 ) -> None:
386 """XSS payload in address is HTML-escaped, not executed."""
387 repo_id = route_repo.repo_id
388 xss = "<script>alert(1)</script>"
389 await db_session.commit()
390 await _seed_stable(db_session, repo_id, address=xss, days_stable=100)
391 await db_session.commit()
392 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
393 assert resp.status_code == 200
394 assert "<script>alert(1)</script>" not in resp.text
395
396 @pytest.mark.asyncio
397 async def test_P2_19_unknown_repo_returns_404(
398 self, client: AsyncClient, route_repo: MusehubRepo
399 ) -> None:
400 """Non-existent repo slug → 404."""
401 resp = await client.get(f"/{_OWNER}/does_not_exist_xyz/intel/stable")
402 assert resp.status_code == 404
403
404 @pytest.mark.asyncio
405 async def test_P2_20_idor_repo_b_rows_not_on_repo_a_page(
406 self, client: AsyncClient, db_session: AsyncSession, route_repo: MusehubRepo
407 ) -> None:
408 """Rows belonging to repo B are not visible on repo A's stable page."""
409 repo_b = await create_repo(db_session, owner=_OWNER, slug="stableidor_b")
410 repo_id_b = repo_b.repo_id
411 await db_session.commit()
412 await _seed_stable(db_session, repo_id_b,
413 address="pkg/idor.py::private_fn",
414 days_stable=300)
415 await db_session.commit()
416 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable")
417 assert "private_fn" not in resp.text
418
419
420 # ---------------------------------------------------------------------------
421 # Tier 8 — Detail page
422 # ---------------------------------------------------------------------------
423
424 class TestStableDetailPage:
425 """Integration tests for the /intel/stable/detail route."""
426
427 @pytest.mark.asyncio
428 async def test_P4_01_detail_returns_200_for_known_address(
429 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
430 ) -> None:
431 """Known address → 200 with symbol data rendered."""
432 resp = await client.get(
433 f"/{_OWNER}/{_SLUG}/intel/stable/detail",
434 params={"address": "pkg/codec.py::pack"},
435 )
436 assert resp.status_code == 200
437 assert "pack" in resp.text
438
439 @pytest.mark.asyncio
440 async def test_P4_02_detail_shows_days_stable(
441 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
442 ) -> None:
443 """Detail page shows days_stable value (400 for parse, not eternal)."""
444 resp = await client.get(
445 f"/{_OWNER}/{_SLUG}/intel/stable/detail",
446 params={"address": "pkg/core.py::parse"},
447 )
448 assert "400" in resp.text
449
450 @pytest.mark.asyncio
451 async def test_P4_03_detail_shows_empty_state_for_unknown_address(
452 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
453 ) -> None:
454 """Unknown address → 200 with empty-state message (no symbol data)."""
455 resp = await client.get(
456 f"/{_OWNER}/{_SLUG}/intel/stable/detail",
457 params={"address": "pkg/does_not_exist.py::missing"},
458 )
459 assert resp.status_code == 200
460 assert "No stable data" in resp.text
461
462 @pytest.mark.asyncio
463 async def test_P4_04_detail_no_address_shows_empty_state(
464 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
465 ) -> None:
466 """No address query param → 200 with empty-state message."""
467 resp = await client.get(f"/{_OWNER}/{_SLUG}/intel/stable/detail")
468 assert resp.status_code == 200
469 assert "No stable data" in resp.text
470
471 @pytest.mark.asyncio
472 async def test_P4_05_detail_xss_in_address_escaped(
473 self, client: AsyncClient, db_session: AsyncSession, route_repo: MusehubRepo
474 ) -> None:
475 """XSS payload in address query param is HTML-escaped."""
476 xss = "<script>alert('xss')</script>"
477 resp = await client.get(
478 f"/{_OWNER}/{_SLUG}/intel/stable/detail",
479 params={"address": xss},
480 )
481 assert resp.status_code == 200
482 assert "<script>alert" not in resp.text
483
484 @pytest.mark.asyncio
485 async def test_P4_06_detail_shows_breadcrumb_back_link(
486 self, client: AsyncClient, route_repo_with_symbols: MusehubRepo
487 ) -> None:
488 """Detail page breadcrumb includes link to stable list."""
489 resp = await client.get(
490 f"/{_OWNER}/{_SLUG}/intel/stable/detail",
491 params={"address": "pkg/core.py::parse"},
492 )
493 assert "intel/stable" in resp.text
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor 21 days ago