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