gabriel / musehub public
test_intel_api_surface.py python
710 lines 29.1 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """API Surface dashboard — full 7-tier test suite (issue #19).
2
3 Tests are written TDD-first: all tests in this file must be RED before
4 Phase 3–5 implementation begins, then GREEN after.
5
6 Tiers:
7 T01–T03 Layer T1 — DB model (composite PK, nullable fields, cascade)
8 T04–T06 Layer T2 — Provider batch performance
9 T07–T15 Layer T3 — Route (unit / integration)
10 T16–T19 Layer T4 — E2E (HTML body assertions)
11 T20–T22 Layer T5 — State integrity
12 T23–T25 Layer T6 — Performance
13 T26–T30 Layer T7 — Security
14 """
15 from __future__ import annotations
16
17 import time
18 from unittest.mock import AsyncMock, patch
19
20 import pytest
21 import sqlalchemy as sa
22 from httpx import AsyncClient
23 from sqlalchemy.dialects.postgresql import insert as pg_insert
24 from sqlalchemy.ext.asyncio import AsyncSession
25
26 from musehub.db import musehub_models as dbm
27 from musehub.types.json_types import JSONObject
28 from tests.factories import create_repo
29 from muse.core.types import long_id
30
31 _REF = long_id("b" * 64)
32
33
34 # ---------------------------------------------------------------------------
35 # Helpers
36 # ---------------------------------------------------------------------------
37
38 async def _insert_as_row(
39 session: AsyncSession,
40 repo_id: str,
41 address: str,
42 kind: str = "function",
43 signature_id: str | None = None,
44 visibility: str = "public",
45 ref: str = _REF,
46 ) -> None:
47 """Upsert one row into musehub_intel_api_surface."""
48 await session.execute(
49 pg_insert(dbm.MusehubIntelApiSurface)
50 .values(
51 repo_id=repo_id,
52 address=address,
53 kind=kind,
54 signature_id=signature_id,
55 visibility=visibility,
56 ref=ref,
57 )
58 .on_conflict_do_update(
59 index_elements=["repo_id", "address"],
60 set_={
61 "kind": kind,
62 "signature_id": signature_id,
63 "visibility": visibility,
64 "ref": ref,
65 },
66 )
67 )
68
69
70 import pytest_asyncio
71
72
73 @pytest_asyncio.fixture
74 async def as_repo(db_session: AsyncSession):
75 """Repo seeded with one symbol of each kind."""
76 repo = await create_repo(db_session, owner="asuser", slug="as-e2e")
77 rid = str(repo.repo_id)
78
79 await _insert_as_row(db_session, rid, "src/billing.py::compute_total",
80 kind="function")
81 await _insert_as_row(db_session, rid, "src/billing.py::async_fetch",
82 kind="async_function")
83 await _insert_as_row(db_session, rid, "src/models.py::UserRecord",
84 kind="class")
85 await _insert_as_row(db_session, rid, "src/models.py::UserRecord.save",
86 kind="method")
87 await _insert_as_row(db_session, rid, "src/models.py::UserRecord.async_load",
88 kind="async_method")
89
90 await db_session.commit()
91 return repo
92
93
94 # ─────────────────────────────────────────────────────────────────────────────
95 # Layer T1 — DB model
96 # ─────────────────────────────────────────────────────────────────────────────
97
98 class TestDBModel:
99
100 def test_T01_model_has_required_columns(self) -> None:
101 """MusehubIntelApiSurface must declare all expected mapped columns."""
102 cols = {c.key for c in sa.inspect(dbm.MusehubIntelApiSurface).mapper.column_attrs}
103 for required in ("repo_id", "address", "kind", "signature_id", "visibility", "ref"):
104 assert required in cols, f"Column '{required}' missing from MusehubIntelApiSurface"
105
106 def test_T02_signature_id_is_nullable(self) -> None:
107 """signature_id must be nullable — not all symbols have a signature object."""
108 col = dbm.MusehubIntelApiSurface.__table__.c["signature_id"]
109 assert col.nullable, "signature_id must be nullable"
110
111 @pytest.mark.asyncio
112 async def test_T03_row_insert_and_cascade_delete(
113 self, db_session: AsyncSession
114 ) -> None:
115 """Row inserts cleanly; deleting the repo cascades to api_surface rows."""
116 repo = await create_repo(db_session, owner="asuser", slug="t03-cascade")
117 rid = str(repo.repo_id)
118 await _insert_as_row(db_session, rid, "src/x.py::fn")
119 await db_session.commit()
120
121 # row present
122 row = await db_session.scalar(
123 sa.select(dbm.MusehubIntelApiSurface).where(
124 dbm.MusehubIntelApiSurface.repo_id == rid,
125 dbm.MusehubIntelApiSurface.address == "src/x.py::fn",
126 )
127 )
128 assert row is not None, "Row not found after insert"
129
130 # cascade delete
131 await db_session.delete(repo)
132 await db_session.commit()
133
134 remaining = (await db_session.execute(
135 sa.select(dbm.MusehubIntelApiSurface).where(
136 dbm.MusehubIntelApiSurface.repo_id == rid
137 )
138 )).scalars().all()
139 assert not remaining, "Cascade delete failed — api_surface rows remain after repo delete"
140
141
142 # ─────────────────────────────────────────────────────────────────────────────
143 # Layer T2 — Provider batch performance
144 # ─────────────────────────────────────────────────────────────────────────────
145
146 async def _seed_snapshot(
147 session: AsyncSession,
148 repo_id: str,
149 manifest: dict[str, str],
150 ) -> str:
151 """Insert a MusehubCommit + MusehubSnapshot and return the snapshot_id."""
152 import msgpack
153 from datetime import datetime, timezone
154
155 snap_id = long_id("c" * 64)
156 commit_id = long_id("d" * 64)
157
158 await session.execute(
159 pg_insert(dbm.MusehubSnapshot)
160 .values(
161 snapshot_id=snap_id,
162 repo_id=repo_id,
163 directories=[],
164 manifest_blob=msgpack.packb(manifest),
165 entry_count=len(manifest),
166 created_at=datetime.now(timezone.utc),
167 )
168 .on_conflict_do_nothing()
169 )
170 await session.execute(
171 pg_insert(dbm.MusehubCommit)
172 .values(
173 commit_id=commit_id,
174 repo_id=repo_id,
175 branch="dev",
176 parent_ids=[],
177 message="test",
178 author="asuser",
179 timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc),
180 snapshot_id=snap_id,
181 )
182 .on_conflict_do_nothing()
183 )
184 await session.commit()
185 return snap_id
186
187
188 def _fake_tree(n: int, prefix: str = "fn") -> JSONObject:
189 """Return a SymbolTree dict with *n* public function symbols."""
190 return {
191 f"src/file.py::{prefix}_{i}": {
192 "kind": "function",
193 "name": f"{prefix}_{i}",
194 "qualified_name": f"{prefix}_{i}",
195 "content_id": long_id("a" * 64),
196 "body_hash": long_id("b" * 64),
197 "signature_id": long_id("c" * 64),
198 "metadata_id": "",
199 "canonical_key": f"src/file.py##function#{prefix}_{i}#1",
200 "lineno": i + 1,
201 "end_lineno": i + 2,
202 }
203 for i in range(n)
204 }
205
206
207 class TestProviderBatch:
208
209 @pytest.mark.asyncio
210 async def test_T04_provider_issues_one_sql_per_chunk(
211 self, db_session: AsyncSession
212 ) -> None:
213 """ApiSurfaceProvider must batch-upsert, not execute one statement per symbol."""
214 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
215
216 repo = await create_repo(db_session, owner="asuser", slug="t04-batch")
217 rid = str(repo.repo_id)
218 await _seed_snapshot(db_session, rid, {"src/file.py": long_id("e" * 64)})
219
220 execute_calls: list[sa.Executable] = []
221 original_execute = db_session.execute
222
223 async def counting_execute(stmt, *args, **kwargs):
224 execute_calls.append(stmt)
225 return await original_execute(stmt, *args, **kwargs)
226
227 mock_backend = AsyncMock()
228 mock_backend.get = AsyncMock(return_value=b"# placeholder")
229
230 with (
231 patch("musehub.services.musehub_intel_providers.get_backend",
232 return_value=mock_backend),
233 patch("musehub.services.musehub_intel_providers.parse_symbols",
234 return_value=_fake_tree(50)),
235 ):
236 db_session.execute = counting_execute # type: ignore[method-assign]
237 await _PROVIDER_REGISTRY["intel.code.api_surface"].compute(
238 db_session, rid, _REF,
239 {"owner": repo.owner, "slug": repo.slug},
240 )
241 db_session.execute = original_execute # type: ignore[method-assign]
242
243 # 50 symbols fit in one chunk — expect exactly 1 INSERT execute
244 insert_calls = [
245 c for c in execute_calls
246 if "insert" in str(type(c).__name__).lower() or "insert" in str(c).lower()
247 ]
248 assert len(insert_calls) == 1, (
249 f"Expected 1 batch upsert for 50 symbols, got {len(insert_calls)}"
250 )
251
252 @pytest.mark.asyncio
253 async def test_T05_provider_uses_ceil_n_over_1000_sql_calls_for_2500_symbols(
254 self, db_session: AsyncSession
255 ) -> None:
256 """2,500 symbols → exactly 3 INSERT statements (ceil(2500/1000) = 3)."""
257 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
258
259 repo = await create_repo(db_session, owner="asuser", slug="t05-chunks")
260 rid = str(repo.repo_id)
261 await _seed_snapshot(db_session, rid, {"src/big.py": long_id("f" * 64)})
262
263 execute_calls: list[sa.Executable] = []
264 original_execute = db_session.execute
265
266 async def counting_execute(stmt, *args, **kwargs):
267 execute_calls.append(stmt)
268 return await original_execute(stmt, *args, **kwargs)
269
270 mock_backend = AsyncMock()
271 mock_backend.get = AsyncMock(return_value=b"# placeholder")
272
273 with (
274 patch("musehub.services.musehub_intel_providers.get_backend",
275 return_value=mock_backend),
276 patch("musehub.services.musehub_intel_providers.parse_symbols",
277 return_value=_fake_tree(2500)),
278 ):
279 db_session.execute = counting_execute # type: ignore[method-assign]
280 result = await _PROVIDER_REGISTRY["intel.code.api_surface"].compute(
281 db_session, rid, _REF,
282 {"owner": repo.owner, "slug": repo.slug},
283 )
284 db_session.execute = original_execute # type: ignore[method-assign]
285
286 insert_calls = [
287 c for c in execute_calls
288 if "insert" in str(type(c).__name__).lower() or "insert" in str(c).lower()
289 ]
290 assert len(insert_calls) == 3, (
291 f"2500 symbols should produce 3 INSERT chunks, got {len(insert_calls)}"
292 )
293 assert result == [("intel.code.api_surface", {"count": 2500})]
294
295 @pytest.mark.asyncio
296 async def test_T06_empty_symbols_returns_empty_list(
297 self, db_session: AsyncSession
298 ) -> None:
299 """Provider must return [] and issue no INSERTs when parse_symbols yields nothing."""
300 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
301
302 repo = await create_repo(db_session, owner="asuser", slug="t06-empty")
303 rid = str(repo.repo_id)
304 await _seed_snapshot(db_session, rid, {"src/empty.py": long_id("a" * 64)})
305
306 execute_calls: list[sa.Executable] = []
307 original_execute = db_session.execute
308
309 async def counting_execute(stmt, *args, **kwargs):
310 execute_calls.append(stmt)
311 return await original_execute(stmt, *args, **kwargs)
312
313 mock_backend = AsyncMock()
314 mock_backend.get = AsyncMock(return_value=b"# no public symbols")
315
316 with (
317 patch("musehub.services.musehub_intel_providers.get_backend",
318 return_value=mock_backend),
319 patch("musehub.services.musehub_intel_providers.parse_symbols",
320 return_value={}),
321 ):
322 db_session.execute = counting_execute # type: ignore[method-assign]
323 result = await _PROVIDER_REGISTRY["intel.code.api_surface"].compute(
324 db_session, rid, _REF,
325 {"owner": repo.owner, "slug": repo.slug},
326 )
327 db_session.execute = original_execute # type: ignore[method-assign]
328
329 assert result == [], "Empty symbols list must return []"
330 insert_calls = [c for c in execute_calls if "insert" in str(c).lower()]
331 assert len(insert_calls) == 0, "No DB writes expected for empty symbol list"
332
333
334 # ─────────────────────────────────────────────────────────────────────────────
335 # Layer T3 — Route (unit / integration)
336 # ─────────────────────────────────────────────────────────────────────────────
337
338 class TestRoute:
339
340 @pytest.mark.asyncio
341 async def test_T07_returns_200_with_empty_repo(
342 self, client: AsyncClient, db_session: AsyncSession
343 ) -> None:
344 """Route must return 200 even when musehub_intel_api_surface has no rows."""
345 await create_repo(db_session, owner="asuser", slug="t07-empty")
346 await db_session.commit()
347 r = await client.get("/asuser/t07-empty/intel/api-surface")
348 assert r.status_code == 200
349
350 @pytest.mark.asyncio
351 async def test_T08_returns_200_with_data(
352 self, client: AsyncClient, as_repo
353 ) -> None:
354 """Route returns 200 when rows exist."""
355 r = await client.get("/asuser/as-e2e/intel/api-surface")
356 assert r.status_code == 200
357
358 @pytest.mark.asyncio
359 async def test_T09_kind_filter_function_only(
360 self, client: AsyncClient, as_repo
361 ) -> None:
362 """?kind=function returns only function symbols, not class or method."""
363 r = await client.get("/asuser/as-e2e/intel/api-surface?kind=function")
364 assert r.status_code == 200
365 assert "compute_total" in r.text
366 assert "UserRecord.save" not in r.text
367 assert "UserRecord" not in r.text or "compute_total" in r.text
368
369 @pytest.mark.asyncio
370 async def test_T10_kind_filter_class_only(
371 self, client: AsyncClient, as_repo
372 ) -> None:
373 """?kind=class returns only class symbols."""
374 r = await client.get("/asuser/as-e2e/intel/api-surface?kind=class")
375 assert r.status_code == 200
376 assert "UserRecord" in r.text
377 assert "compute_total" not in r.text
378
379 @pytest.mark.asyncio
380 async def test_T11_kind_filter_async_function(
381 self, client: AsyncClient, as_repo
382 ) -> None:
383 """?kind=async_function returns only async_function symbols."""
384 r = await client.get("/asuser/as-e2e/intel/api-surface?kind=async_function")
385 assert r.status_code == 200
386 assert "async_fetch" in r.text
387 assert "compute_total" not in r.text
388
389 @pytest.mark.asyncio
390 async def test_T12_unknown_kind_coerced_to_all(
391 self, client: AsyncClient, as_repo
392 ) -> None:
393 """?kind=garbage must return 200 (treated as no filter), not 400/500."""
394 r = await client.get("/asuser/as-e2e/intel/api-surface?kind=garbage")
395 assert r.status_code == 200
396
397 @pytest.mark.asyncio
398 async def test_T13_top_param_limits_results(
399 self, client: AsyncClient, db_session: AsyncSession
400 ) -> None:
401 """?top=20 returns at most 20 symbols even when 25 exist."""
402 repo = await create_repo(db_session, owner="asuser", slug="t13-top")
403 rid = str(repo.repo_id)
404 for i in range(25):
405 await _insert_as_row(db_session, rid,
406 f"src/f{i}.py::fn_{i}", kind="function")
407 await db_session.commit()
408
409 r = await client.get("/asuser/t13-top/intel/api-surface?top=20")
410 assert r.status_code == 200
411 count = sum(1 for i in range(25) if f"src/f{i}.py::fn_{i}" in r.text)
412 assert count <= 20, f"Expected ≤20 results for ?top=20, got {count}"
413
414 @pytest.mark.asyncio
415 async def test_T14_top_invalid_string_returns_422(
416 self, client: AsyncClient, as_repo
417 ) -> None:
418 """?top=notanumber must be rejected with 422 (FastAPI type validation)."""
419 r = await client.get("/asuser/as-e2e/intel/api-surface?top=notanumber")
420 assert r.status_code == 422
421
422 @pytest.mark.asyncio
423 async def test_T15_unknown_repo_returns_404(
424 self, client: AsyncClient
425 ) -> None:
426 """Non-existent repo path must return 404, not 200 or 500."""
427 r = await client.get("/nobody/no-such-repo/intel/api-surface")
428 assert r.status_code in (403, 404)
429
430
431 # ─────────────────────────────────────────────────────────────────────────────
432 # Layer T4 — E2E (HTML body assertions)
433 # ─────────────────────────────────────────────────────────────────────────────
434
435 class TestE2E:
436
437 @pytest.mark.asyncio
438 async def test_T16_total_count_chip_shows_correct_value(
439 self, client: AsyncClient, as_repo
440 ) -> None:
441 """Stat chip for Total must reflect the DB row count (5 symbols seeded)."""
442 r = await client.get("/asuser/as-e2e/intel/api-surface")
443 assert r.status_code == 200
444 # 5 symbols seeded in fixture; total chip must contain "5"
445 assert "5" in r.text
446
447 @pytest.mark.asyncio
448 async def test_T17_kind_breakdown_chips_present(
449 self, client: AsyncClient, as_repo
450 ) -> None:
451 """Kind breakdown stat chips must appear for all five kinds."""
452 r = await client.get("/asuser/as-e2e/intel/api-surface")
453 assert r.status_code == 200
454 body = r.text.lower()
455 for kind_label in ("function", "class", "method"):
456 assert kind_label in body, f"Kind label '{kind_label}' missing from page"
457
458 @pytest.mark.asyncio
459 async def test_T18_symbol_address_split_rendered(
460 self, client: AsyncClient, as_repo
461 ) -> None:
462 """Symbol file and name parts must both appear in the HTML."""
463 r = await client.get("/asuser/as-e2e/intel/api-surface")
464 assert r.status_code == 200
465 # file part
466 assert "src/billing.py" in r.text
467 # name part
468 assert "compute_total" in r.text
469
470 @pytest.mark.asyncio
471 async def test_T19_dashboard_card_links_to_api_surface_page(
472 self, client: AsyncClient, as_repo
473 ) -> None:
474 """Intel dashboard must include a link to /intel/api-surface."""
475 r = await client.get("/asuser/as-e2e/intel")
476 assert r.status_code == 200
477 assert b"/intel/api-surface" in r.content
478
479
480 # ─────────────────────────────────────────────────────────────────────────────
481 # Layer T5 — State integrity
482 # ─────────────────────────────────────────────────────────────────────────────
483
484 class TestStateIntegrity:
485
486 @pytest.mark.asyncio
487 async def test_T20_double_upsert_produces_one_row(
488 self, db_session: AsyncSession
489 ) -> None:
490 """Upserting the same address twice must not create duplicate rows."""
491 repo = await create_repo(db_session, owner="asuser", slug="t20-dup")
492 rid = str(repo.repo_id)
493 addr = "src/a.py::fn"
494
495 for _ in range(2):
496 await _insert_as_row(db_session, rid, addr, kind="function")
497 await db_session.commit()
498
499 rows = (await db_session.execute(
500 sa.select(dbm.MusehubIntelApiSurface).where(
501 dbm.MusehubIntelApiSurface.repo_id == rid
502 )
503 )).scalars().all()
504 assert len(rows) == 1, f"Expected 1 row, got {len(rows)} — upsert created duplicates"
505
506 @pytest.mark.asyncio
507 async def test_T21_second_upsert_overwrites_kind(
508 self, db_session: AsyncSession
509 ) -> None:
510 """A second upsert with a different kind must overwrite the first."""
511 repo = await create_repo(db_session, owner="asuser", slug="t21-overwrite")
512 rid = str(repo.repo_id)
513 addr = "src/a.py::Foo"
514
515 await _insert_as_row(db_session, rid, addr, kind="class")
516 await _insert_as_row(db_session, rid, addr, kind="function")
517 await db_session.commit()
518
519 row = await db_session.scalar(
520 sa.select(dbm.MusehubIntelApiSurface).where(
521 dbm.MusehubIntelApiSurface.repo_id == rid,
522 dbm.MusehubIntelApiSurface.address == addr,
523 )
524 )
525 assert row is not None
526 assert row.kind == "function", (
527 f"Expected kind='function' after second upsert, got '{row.kind}'"
528 )
529
530 @pytest.mark.asyncio
531 async def test_T22_cross_repo_isolation(
532 self, db_session: AsyncSession
533 ) -> None:
534 """Symbols from repo A must not appear under repo B's page URL."""
535 repo_a = await create_repo(db_session, owner="asuser", slug="t22-repo-a")
536 repo_b = await create_repo(db_session, owner="asuser", slug="t22-repo-b")
537
538 await _insert_as_row(db_session, str(repo_a.repo_id),
539 "src/secret.py::private_fn", kind="function")
540 await db_session.commit()
541
542 rows_b = (await db_session.execute(
543 sa.select(dbm.MusehubIntelApiSurface).where(
544 dbm.MusehubIntelApiSurface.repo_id == str(repo_b.repo_id)
545 )
546 )).scalars().all()
547 assert not rows_b, "Repo B must not see Repo A's api_surface symbols"
548
549
550 # ─────────────────────────────────────────────────────────────────────────────
551 # Layer T6 — Performance
552 # ─────────────────────────────────────────────────────────────────────────────
553
554 class TestPerformance:
555
556 @pytest.mark.asyncio
557 async def test_T23_route_responds_under_200ms_for_5k_symbols(
558 self, client: AsyncClient, db_session: AsyncSession
559 ) -> None:
560 """Route must respond in < 200ms for a repo with 5,000 symbol rows."""
561 repo = await create_repo(db_session, owner="asuser", slug="t23-perf")
562 rid = str(repo.repo_id)
563
564 chunk_size = 1000
565 kinds = ["function", "async_function", "class", "method", "async_method"]
566 for start in range(0, 5_000, chunk_size):
567 rows = [
568 {
569 "repo_id": rid,
570 "address": f"src/file{i}.py::sym_{i}",
571 "kind": kinds[i % len(kinds)],
572 "signature_id": None,
573 "visibility": "public",
574 "ref": _REF,
575 }
576 for i in range(start, start + chunk_size)
577 ]
578 await db_session.execute(
579 pg_insert(dbm.MusehubIntelApiSurface)
580 .values(rows)
581 .on_conflict_do_nothing()
582 )
583 await db_session.commit()
584
585 t0 = time.monotonic()
586 r = await client.get("/asuser/t23-perf/intel/api-surface")
587 elapsed = time.monotonic() - t0
588
589 assert r.status_code == 200
590 assert elapsed < 0.2, f"Route took {elapsed:.3f}s for 5k symbols (limit: 0.2s)"
591
592 @pytest.mark.asyncio
593 async def test_T24_db_query_uses_repo_index(
594 self, db_session: AsyncSession
595 ) -> None:
596 """SELECT on musehub_intel_api_surface must use ix_intel_api_surface_repo index."""
597 explain = await db_session.execute(
598 sa.text(
599 "EXPLAIN SELECT * FROM musehub_intel_api_surface WHERE repo_id = 'x'"
600 )
601 )
602 plan = " ".join(row[0] for row in explain.all())
603 assert "ix_intel_api_surface_repo" in plan or "Index" in plan, (
604 f"Query plan does not use ix_intel_api_surface_repo:\n{plan}"
605 )
606
607 @pytest.mark.asyncio
608 async def test_T25_batch_upsert_1000_rows_under_500ms(
609 self, db_session: AsyncSession
610 ) -> None:
611 """Direct batch upsert of 1,000 rows must complete in < 500ms wall time."""
612 repo = await create_repo(db_session, owner="asuser", slug="t25-batch")
613 rid = str(repo.repo_id)
614 rows = [
615 {
616 "repo_id": rid,
617 "address": f"src/f{i}.py::fn",
618 "kind": "function",
619 "signature_id": None,
620 "visibility": "public",
621 "ref": _REF,
622 }
623 for i in range(1000)
624 ]
625 t0 = time.monotonic()
626 await db_session.execute(
627 pg_insert(dbm.MusehubIntelApiSurface)
628 .values(rows)
629 .on_conflict_do_nothing()
630 )
631 await db_session.commit()
632 elapsed = time.monotonic() - t0
633 assert elapsed < 0.5, f"1000-row batch took {elapsed:.3f}s (limit: 0.5s)"
634
635
636 # ─────────────────────────────────────────────────────────────────────────────
637 # Layer T7 — Security
638 # ─────────────────────────────────────────────────────────────────────────────
639
640 class TestSecurity:
641
642 @pytest.mark.asyncio
643 async def test_T26_xss_in_address_is_escaped(
644 self, client: AsyncClient, db_session: AsyncSession
645 ) -> None:
646 """XSS payload in address must be HTML-escaped in the response."""
647 repo = await create_repo(db_session, owner="asuser", slug="t26-xss")
648 rid = str(repo.repo_id)
649 xss = "<script>alert(1)</script>"
650 await _insert_as_row(db_session, rid, f"src/x.py::{xss[:40]}")
651 await db_session.commit()
652
653 r = await client.get("/asuser/t26-xss/intel/api-surface")
654 assert r.status_code == 200
655 assert "<script>alert" not in r.text, "XSS in address not escaped by Jinja2"
656
657 @pytest.mark.asyncio
658 async def test_T27_xss_in_kind_field_is_escaped(
659 self, client: AsyncClient, db_session: AsyncSession
660 ) -> None:
661 """XSS payload stored in kind must be HTML-escaped in the response."""
662 repo = await create_repo(db_session, owner="asuser", slug="t27-xss-kind")
663 rid = str(repo.repo_id)
664 await _insert_as_row(db_session, rid, "src/x.py::fn",
665 kind='<img src=x onerror=alert(1)>')
666 await db_session.commit()
667
668 r = await client.get("/asuser/t27-xss-kind/intel/api-surface")
669 assert r.status_code == 200
670 assert "<img src=x onerror" not in r.text, "XSS in kind not escaped"
671
672 @pytest.mark.asyncio
673 async def test_T28_sql_injection_in_kind_param_safe(
674 self, client: AsyncClient, as_repo
675 ) -> None:
676 """SQL injection string in ?kind= must be safely coerced, no 500."""
677 r = await client.get(
678 "/asuser/as-e2e/intel/api-surface?kind=function%27%20OR%20%271%27%3D%271"
679 )
680 assert r.status_code == 200, f"Expected 200 after SQL injection attempt, got {r.status_code}"
681
682 @pytest.mark.asyncio
683 async def test_T29_top_zero_coerced_to_default(
684 self, client: AsyncClient, as_repo
685 ) -> None:
686 """?top=0 must not issue an empty-LIMIT query; page returns 200."""
687 r = await client.get("/asuser/as-e2e/intel/api-surface?top=0")
688 # FastAPI will validate int but 0 is a valid int — route must coerce it
689 assert r.status_code in (200, 422), (
690 f"?top=0 returned unexpected status {r.status_code}"
691 )
692
693 @pytest.mark.asyncio
694 async def test_T30_private_repo_returns_403_or_404_unauthenticated(
695 self, client: AsyncClient
696 ) -> None:
697 """A non-existent repo path must not return 200 or 500."""
698 r = await client.get("/nobody/no-such-repo/intel/api-surface")
699 assert r.status_code in (403, 404)
700
701
702 # ---------------------------------------------------------------------------
703 # Internal helpers
704 # ---------------------------------------------------------------------------
705
706 def _mock_process(stdout: str, returncode: int = 0) -> AsyncMock:
707 proc = AsyncMock()
708 proc.returncode = returncode
709 proc.communicate = AsyncMock(return_value=(stdout.encode(), b""))
710 return proc
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago