gabriel / musehub public
test_intel_languages.py python
749 lines 30.1 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago
1 """Languages intel — full 7-tier test suite (issue #20).
2
3 Tests are written TDD-first: all tests must be RED before Phase 4–7
4 implementation begins, then GREEN after.
5
6 Tiers
7 -----
8 T01–T04 Layer T1 — DB model (columns, nullable, kinds_json, cascade)
9 T05–T09 Layer T2 — Provider (no subprocess, file counts, kinds, pct, empty)
10 T10–T17 Layer T3 — Route (200, empty state, 404, sort, filter, pagination)
11 T18–T21 Layer T4 — E2E HTML (stat chips, bar width, kind chips, dashboard link)
12 T22–T24 Layer T5 — Data integrity (no duplicates, upsert overwrite, cross-repo)
13 T25–T27 Layer T6 — Performance (provider speed, route speed, index check)
14 T28–T30 Layer T7 — Security (XSS escape, SQL injection, no 500 on bad input)
15 """
16 from __future__ import annotations
17
18 import time
19 from datetime import datetime, timezone
20 from unittest.mock import AsyncMock, patch
21
22 import pytest
23 import pytest_asyncio
24 import sqlalchemy as sa
25 from httpx import AsyncClient
26 from sqlalchemy.dialects.postgresql import insert as pg_insert
27 from sqlalchemy.ext.asyncio import AsyncSession
28
29 from musehub.db import musehub_models as dbm
30 from musehub.types.json_types import JSONObject
31 from tests.factories import create_repo
32 from muse.core.types import long_id
33
34 _REF = long_id("b" * 64)
35
36
37 # ─────────────────────────────────────────────────────────────────────────────
38 # Helpers
39 # ─────────────────────────────────────────────────────────────────────────────
40
41 async def _insert_lang_row(
42 session: AsyncSession,
43 repo_id: str,
44 language: str,
45 file_count: int = 1,
46 symbol_count: int = 0,
47 pct: float = 0.0,
48 kinds_json: JSONObject | None = None,
49 ref: str = _REF,
50 ) -> None:
51 """Upsert one row into musehub_intel_languages."""
52 await session.execute(
53 pg_insert(dbm.MusehubIntelLanguages)
54 .values(
55 repo_id=repo_id,
56 language=language,
57 file_count=file_count,
58 symbol_count=symbol_count,
59 pct=pct,
60 kinds_json=kinds_json,
61 ref=ref,
62 )
63 .on_conflict_do_update(
64 index_elements=["repo_id", "language"],
65 set_={
66 "file_count": file_count,
67 "symbol_count": symbol_count,
68 "pct": pct,
69 "kinds_json": kinds_json,
70 "ref": ref,
71 },
72 )
73 )
74
75
76 async def _seed_snapshot(
77 session: AsyncSession,
78 repo_id: str,
79 manifest: dict[str, str],
80 ) -> str:
81 """Insert a MusehubCommit + MusehubSnapshot, return snapshot_id."""
82 import msgpack
83
84 snap_id = long_id("c" * 64)
85 commit_id = long_id("d" * 64)
86
87 await session.execute(
88 pg_insert(dbm.MusehubSnapshot)
89 .values(
90 snapshot_id = snap_id,
91 repo_id = repo_id,
92 directories = [],
93 manifest_blob= msgpack.packb(manifest),
94 entry_count = len(manifest),
95 created_at = datetime(2026, 1, 1, tzinfo=timezone.utc),
96 )
97 .on_conflict_do_nothing()
98 )
99 await session.execute(
100 pg_insert(dbm.MusehubCommit)
101 .values(
102 commit_id = commit_id,
103 repo_id = repo_id,
104 branch = "dev",
105 parent_ids = [],
106 message = "test",
107 author = "lnuser",
108 timestamp = datetime(2026, 1, 1, tzinfo=timezone.utc),
109 snapshot_id = snap_id,
110 )
111 .on_conflict_do_nothing()
112 )
113 await session.commit()
114 return snap_id
115
116
117 def _fake_tree(n: int, kinds: list[str] | None = None) -> JSONObject:
118 """Return a SymbolTree dict with n public symbols spread across kinds."""
119 _kinds = kinds or ["function", "class", "method"]
120 return {
121 f"src/mod.py::sym_{i}": {
122 "kind": _kinds[i % len(_kinds)],
123 "name": f"sym_{i}",
124 "qualified_name": f"sym_{i}",
125 "content_id": long_id("a" * 64),
126 "body_hash": long_id("b" * 64),
127 "signature_id": long_id("c" * 64),
128 "metadata_id": "",
129 "canonical_key": f"src/mod.py##function#sym_{i}#1",
130 "lineno": i + 1,
131 "end_lineno": i + 2,
132 }
133 for i in range(n)
134 }
135
136
137 @pytest_asyncio.fixture
138 async def ln_repo(db_session: AsyncSession):
139 """Repo seeded with Python, TypeScript, and CSS language rows."""
140 repo = await create_repo(db_session, owner="lnuser", slug="ln-e2e")
141 rid = str(repo.repo_id)
142
143 await _insert_lang_row(
144 db_session, rid, "Python",
145 file_count=30, symbol_count=1500, pct=75.0,
146 kinds_json={"function": 800, "class": 400, "method": 300},
147 )
148 await _insert_lang_row(
149 db_session, rid, "TypeScript",
150 file_count=10, symbol_count=400, pct=20.0,
151 kinds_json={"function": 300, "class": 100},
152 )
153 await _insert_lang_row(
154 db_session, rid, "CSS",
155 file_count=5, symbol_count=0, pct=0.0,
156 kinds_json=None,
157 )
158
159 await db_session.commit()
160 return repo
161
162
163 # ─────────────────────────────────────────────────────────────────────────────
164 # Layer T1 — DB model
165 # ─────────────────────────────────────────────────────────────────────────────
166
167 class TestDBModel:
168
169 def test_T01_model_has_all_required_columns(self) -> None:
170 """MusehubIntelLanguages must declare all expected mapped columns."""
171 cols = {
172 c.key
173 for c in sa.inspect(dbm.MusehubIntelLanguages).mapper.column_attrs
174 }
175 for required in (
176 "repo_id", "language", "file_count", "symbol_count", "pct",
177 "kinds_json", "ref",
178 ):
179 assert required in cols, (
180 f"Column '{required}' missing from MusehubIntelLanguages"
181 )
182
183 def test_T02_kinds_json_is_nullable(self) -> None:
184 """kinds_json must be nullable — non-code languages have no symbol breakdown."""
185 col = dbm.MusehubIntelLanguages.__table__.c["kinds_json"]
186 assert col.nullable, "kinds_json must be nullable"
187
188 def test_T03_composite_pk_is_repo_id_plus_language(self) -> None:
189 """Primary key must be (repo_id, language) — no single-column PK."""
190 pk_cols = {
191 c.name
192 for c in dbm.MusehubIntelLanguages.__table__.primary_key.columns
193 }
194 assert pk_cols == {"repo_id", "language"}, (
195 f"Expected PK {{repo_id, language}}, got {pk_cols}"
196 )
197
198 @pytest.mark.asyncio
199 async def test_T04_cascade_delete_removes_lang_rows(
200 self, db_session: AsyncSession
201 ) -> None:
202 """Deleting a repo must cascade-delete all its language rows."""
203 repo = await create_repo(db_session, owner="lnuser", slug="t04-cascade")
204 rid = str(repo.repo_id)
205 await _insert_lang_row(db_session, rid, "Python", file_count=3)
206 await db_session.commit()
207
208 row = await db_session.scalar(
209 sa.select(dbm.MusehubIntelLanguages).where(
210 dbm.MusehubIntelLanguages.repo_id == rid,
211 dbm.MusehubIntelLanguages.language == "Python",
212 )
213 )
214 assert row is not None, "Row not found after insert"
215
216 await db_session.delete(repo)
217 await db_session.commit()
218
219 remaining = (await db_session.execute(
220 sa.select(dbm.MusehubIntelLanguages).where(
221 dbm.MusehubIntelLanguages.repo_id == rid
222 )
223 )).scalars().all()
224 assert not remaining, (
225 "Cascade delete failed — languages rows remain after repo delete"
226 )
227
228
229 # ─────────────────────────────────────────────────────────────────────────────
230 # Layer T2 — Provider
231 # ─────────────────────────────────────────────────────────────────────────────
232
233 class TestProvider:
234
235 @pytest.mark.asyncio
236 async def test_T05_provider_does_not_use_subprocess(
237 self, db_session: AsyncSession
238 ) -> None:
239 """LanguagesProvider must never call asyncio.create_subprocess_exec or _run_muse."""
240 import inspect
241 from musehub.services import musehub_intel_providers as _mod
242
243 src = inspect.getsource(_mod.LanguagesProvider.compute)
244 assert "create_subprocess" not in src, (
245 "LanguagesProvider.compute calls create_subprocess — forbidden"
246 )
247 assert "_run_muse" not in src, (
248 "LanguagesProvider.compute calls _run_muse — forbidden"
249 )
250
251 @pytest.mark.asyncio
252 async def test_T06_provider_counts_files_per_language(
253 self, db_session: AsyncSession
254 ) -> None:
255 """Provider must count files per language via language_of(), not subprocess."""
256 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
257
258 repo = await create_repo(db_session, owner="lnuser", slug="t06-files")
259 rid = str(repo.repo_id)
260 manifest = {
261 "src/a.py": long_id("e" * 64),
262 "src/b.py": long_id("f" * 64),
263 "src/app.ts": long_id("1" * 64),
264 "static/main.css": long_id("2" * 64),
265 }
266 await _seed_snapshot(db_session, rid, manifest)
267
268 mock_backend = AsyncMock()
269 mock_backend.get = AsyncMock(return_value=b"# placeholder")
270
271 with (
272 patch("musehub.services.musehub_intel_providers.get_backend",
273 return_value=mock_backend),
274 patch("musehub.services.musehub_intel_providers.parse_symbols",
275 return_value={}),
276 patch("musehub.services.musehub_intel_providers.language_of",
277 side_effect=lambda p: (
278 "Python" if p.endswith(".py") else
279 "TypeScript" if p.endswith(".ts") else
280 "CSS"
281 )),
282 ):
283 result = await _PROVIDER_REGISTRY["intel.code.languages"].compute(
284 db_session, rid, _REF,
285 {"owner": repo.owner, "slug": repo.slug},
286 )
287
288 assert result == [("intel.code.languages", {"count": 3})], (
289 f"Expected 3 language rows, got: {result}"
290 )
291
292 rows = (await db_session.execute(
293 sa.select(dbm.MusehubIntelLanguages).where(
294 dbm.MusehubIntelLanguages.repo_id == rid
295 )
296 )).scalars().all()
297 by_lang = {r.language: r for r in rows}
298 assert by_lang["Python"].file_count == 2
299 assert by_lang["TypeScript"].file_count == 1
300 assert by_lang["CSS"].file_count == 1
301
302 @pytest.mark.asyncio
303 async def test_T07_provider_records_kinds_json(
304 self, db_session: AsyncSession
305 ) -> None:
306 """kinds_json must contain kind → count breakdown, imports excluded."""
307 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
308
309 repo = await create_repo(db_session, owner="lnuser", slug="t07-kinds")
310 rid = str(repo.repo_id)
311 await _seed_snapshot(db_session, rid, {"src/x.py": long_id("3" * 64)})
312
313 tree = _fake_tree(6, kinds=["function", "class", "import"])
314
315 mock_backend = AsyncMock()
316 mock_backend.get = AsyncMock(return_value=b"# placeholder")
317
318 with (
319 patch("musehub.services.musehub_intel_providers.get_backend",
320 return_value=mock_backend),
321 patch("musehub.services.musehub_intel_providers.parse_symbols",
322 return_value=tree),
323 patch("musehub.services.musehub_intel_providers.language_of",
324 return_value="Python"),
325 ):
326 await _PROVIDER_REGISTRY["intel.code.languages"].compute(
327 db_session, rid, _REF,
328 {"owner": repo.owner, "slug": repo.slug},
329 )
330
331 row = await db_session.scalar(
332 sa.select(dbm.MusehubIntelLanguages).where(
333 dbm.MusehubIntelLanguages.repo_id == rid,
334 dbm.MusehubIntelLanguages.language == "Python",
335 )
336 )
337 assert row is not None
338 assert row.kinds_json is not None, "kinds_json must not be None when symbols exist"
339 assert "import" not in row.kinds_json, (
340 "import pseudo-symbols must be excluded from kinds_json"
341 )
342 assert set(row.kinds_json.keys()) <= {"function", "class", "method",
343 "async_function", "async_method"}, (
344 f"Unexpected kinds in kinds_json: {set(row.kinds_json.keys())}"
345 )
346
347 @pytest.mark.asyncio
348 async def test_T08_provider_pct_sums_correctly(
349 self, db_session: AsyncSession
350 ) -> None:
351 """Sum of pct across all languages must be ≈ 100 when all files have symbols."""
352 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
353
354 repo = await create_repo(db_session, owner="lnuser", slug="t08-pct")
355 rid = str(repo.repo_id)
356 manifest = {
357 "src/a.py": long_id("4" * 64),
358 "src/b.ts": long_id("5" * 64),
359 }
360 await _seed_snapshot(db_session, rid, manifest)
361
362 py_tree = _fake_tree(3, kinds=["function"])
363 ts_tree = _fake_tree(1, kinds=["function"])
364
365 mock_backend = AsyncMock()
366 mock_backend.get = AsyncMock(return_value=b"# placeholder")
367
368 def _fake_parse(src: bytes, path: str) -> JSONObject:
369 return py_tree if path.endswith(".py") else ts_tree
370
371 with (
372 patch("musehub.services.musehub_intel_providers.get_backend",
373 return_value=mock_backend),
374 patch("musehub.services.musehub_intel_providers.parse_symbols",
375 side_effect=_fake_parse),
376 patch("musehub.services.musehub_intel_providers.language_of",
377 side_effect=lambda p: "Python" if p.endswith(".py") else "TypeScript"),
378 ):
379 await _PROVIDER_REGISTRY["intel.code.languages"].compute(
380 db_session, rid, _REF,
381 {"owner": repo.owner, "slug": repo.slug},
382 )
383
384 rows = (await db_session.execute(
385 sa.select(dbm.MusehubIntelLanguages).where(
386 dbm.MusehubIntelLanguages.repo_id == rid
387 )
388 )).scalars().all()
389 total_pct = sum(r.pct for r in rows)
390 assert abs(total_pct - 100.0) < 0.01, (
391 f"pct values do not sum to 100 (sum={total_pct:.2f})"
392 )
393
394 @pytest.mark.asyncio
395 async def test_T09_provider_returns_empty_when_no_snapshot(
396 self, db_session: AsyncSession
397 ) -> None:
398 """Provider must return [] without crashing when the repo has no snapshot."""
399 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
400
401 repo = await create_repo(db_session, owner="lnuser", slug="t09-nosnap")
402 rid = str(repo.repo_id)
403 await db_session.commit()
404
405 result = await _PROVIDER_REGISTRY["intel.code.languages"].compute(
406 db_session, rid, _REF,
407 {"owner": repo.owner, "slug": repo.slug},
408 )
409 assert result == [], f"Expected [] when no snapshot exists, got {result}"
410
411
412 # ─────────────────────────────────────────────────────────────────────────────
413 # Layer T3 — Route
414 # ─────────────────────────────────────────────────────────────────────────────
415
416 class TestRoute:
417
418 @pytest.mark.asyncio
419 async def test_T10_returns_200_with_language_data(
420 self, client: AsyncClient, ln_repo
421 ) -> None:
422 """Route must return 200 when language rows exist."""
423 r = await client.get("/lnuser/ln-e2e/intel/languages")
424 assert r.status_code == 200
425
426 @pytest.mark.asyncio
427 async def test_T11_returns_200_with_empty_repo(
428 self, client: AsyncClient, db_session: AsyncSession
429 ) -> None:
430 """Route must return 200 even when musehub_intel_languages has no rows."""
431 await create_repo(db_session, owner="lnuser", slug="t11-empty")
432 await db_session.commit()
433 r = await client.get("/lnuser/t11-empty/intel/languages")
434 assert r.status_code == 200
435
436 @pytest.mark.asyncio
437 async def test_T12_unknown_repo_returns_404(
438 self, client: AsyncClient
439 ) -> None:
440 """Non-existent repo path must return 403 or 404, not 200 or 500."""
441 r = await client.get("/nobody/no-such-repo/intel/languages")
442 assert r.status_code in (403, 404)
443
444 @pytest.mark.asyncio
445 async def test_T13_sort_by_files_param_accepted(
446 self, client: AsyncClient, ln_repo
447 ) -> None:
448 """?sort=files must return 200 and not raise an error."""
449 r = await client.get("/lnuser/ln-e2e/intel/languages?sort=files")
450 assert r.status_code == 200
451
452 @pytest.mark.asyncio
453 async def test_T14_sort_by_symbols_param_accepted(
454 self, client: AsyncClient, ln_repo
455 ) -> None:
456 """?sort=symbols must return 200."""
457 r = await client.get("/lnuser/ln-e2e/intel/languages?sort=symbols")
458 assert r.status_code == 200
459
460 @pytest.mark.asyncio
461 async def test_T15_unknown_sort_coerced_to_default(
462 self, client: AsyncClient, ln_repo
463 ) -> None:
464 """?sort=garbage must return 200, coerced to default sort (pct desc)."""
465 r = await client.get("/lnuser/ln-e2e/intel/languages?sort=garbage")
466 assert r.status_code == 200
467
468 @pytest.mark.asyncio
469 async def test_T16_top_param_limits_rows(
470 self, client: AsyncClient, db_session: AsyncSession
471 ) -> None:
472 """?top=20 must return at most 20 language rows when 25 exist."""
473 repo = await create_repo(db_session, owner="lnuser", slug="t16-top")
474 rid = str(repo.repo_id)
475 langs = [f"Lang{i:02d}" for i in range(25)]
476 for i, lang in enumerate(langs):
477 await _insert_lang_row(db_session, rid, lang, file_count=i + 1)
478 await db_session.commit()
479
480 r = await client.get("/lnuser/t16-top/intel/languages?top=20")
481 assert r.status_code == 200
482 count = sum(1 for lang in langs if lang in r.text)
483 assert count <= 20, f"Expected ≤20 languages for ?top=20, found {count}"
484
485 @pytest.mark.asyncio
486 async def test_T17_top_invalid_string_returns_422(
487 self, client: AsyncClient, ln_repo
488 ) -> None:
489 """?top=abc must be rejected with 422 (FastAPI type validation)."""
490 r = await client.get("/lnuser/ln-e2e/intel/languages?top=abc")
491 assert r.status_code == 422
492
493
494 # ─────────────────────────────────────────────────────────────────────────────
495 # Layer T4 — E2E HTML
496 # ─────────────────────────────────────────────────────────────────────────────
497
498 class TestE2E:
499
500 @pytest.mark.asyncio
501 async def test_T18_language_names_appear_in_page(
502 self, client: AsyncClient, ln_repo
503 ) -> None:
504 """All seeded language names must appear in the rendered HTML."""
505 r = await client.get("/lnuser/ln-e2e/intel/languages")
506 assert r.status_code == 200
507 for lang in ("Python", "TypeScript", "CSS"):
508 assert lang in r.text, f"Language '{lang}' missing from page"
509
510 @pytest.mark.asyncio
511 async def test_T19_pct_bar_width_rendered(
512 self, client: AsyncClient, ln_repo
513 ) -> None:
514 """A width style attribute must appear in the HTML (for bar rendering)."""
515 r = await client.get("/lnuser/ln-e2e/intel/languages")
516 assert r.status_code == 200
517 assert "width:" in r.text, "No width style found — pct bars not rendered"
518
519 @pytest.mark.asyncio
520 async def test_T20_kind_chips_rendered_for_python(
521 self, client: AsyncClient, ln_repo
522 ) -> None:
523 """Kind chips for Python (function, class, method) must appear in HTML."""
524 r = await client.get("/lnuser/ln-e2e/intel/languages")
525 assert r.status_code == 200
526 body = r.text.lower()
527 for kind in ("function", "class", "method"):
528 assert kind in body, f"Kind chip '{kind}' missing from languages page"
529
530 @pytest.mark.asyncio
531 async def test_T21_dashboard_card_links_to_languages_page(
532 self, client: AsyncClient, ln_repo
533 ) -> None:
534 """Intel dashboard must include a link to /intel/languages."""
535 r = await client.get("/lnuser/ln-e2e/intel")
536 assert r.status_code == 200
537 assert b"/intel/languages" in r.content
538
539
540 # ─────────────────────────────────────────────────────────────────────────────
541 # Layer T5 — Data integrity
542 # ─────────────────────────────────────────────────────────────────────────────
543
544 class TestDataIntegrity:
545
546 @pytest.mark.asyncio
547 async def test_T22_double_upsert_produces_one_row(
548 self, db_session: AsyncSession
549 ) -> None:
550 """Upserting the same (repo_id, language) twice must not create duplicates."""
551 repo = await create_repo(db_session, owner="lnuser", slug="t22-dup")
552 rid = str(repo.repo_id)
553
554 for _ in range(2):
555 await _insert_lang_row(db_session, rid, "Python", file_count=5)
556 await db_session.commit()
557
558 rows = (await db_session.execute(
559 sa.select(dbm.MusehubIntelLanguages).where(
560 dbm.MusehubIntelLanguages.repo_id == rid
561 )
562 )).scalars().all()
563 assert len(rows) == 1, (
564 f"Expected 1 row after double upsert, got {len(rows)}"
565 )
566
567 @pytest.mark.asyncio
568 async def test_T23_second_upsert_overwrites_file_count(
569 self, db_session: AsyncSession
570 ) -> None:
571 """A second upsert must overwrite file_count with the latest value."""
572 repo = await create_repo(db_session, owner="lnuser", slug="t23-overwrite")
573 rid = str(repo.repo_id)
574
575 await _insert_lang_row(db_session, rid, "Python", file_count=5)
576 await _insert_lang_row(db_session, rid, "Python", file_count=12)
577 await db_session.commit()
578
579 row = await db_session.scalar(
580 sa.select(dbm.MusehubIntelLanguages).where(
581 dbm.MusehubIntelLanguages.repo_id == rid,
582 dbm.MusehubIntelLanguages.language == "Python",
583 )
584 )
585 assert row is not None
586 assert row.file_count == 12, (
587 f"Expected file_count=12 after overwrite upsert, got {row.file_count}"
588 )
589
590 @pytest.mark.asyncio
591 async def test_T24_cross_repo_isolation(
592 self, db_session: AsyncSession
593 ) -> None:
594 """Languages from repo A must not appear in repo B's DB rows."""
595 repo_a = await create_repo(db_session, owner="lnuser", slug="t24-repo-a")
596 repo_b = await create_repo(db_session, owner="lnuser", slug="t24-repo-b")
597
598 await _insert_lang_row(
599 db_session, str(repo_a.repo_id), "SecretLang", file_count=99
600 )
601 await db_session.commit()
602
603 rows_b = (await db_session.execute(
604 sa.select(dbm.MusehubIntelLanguages).where(
605 dbm.MusehubIntelLanguages.repo_id == str(repo_b.repo_id)
606 )
607 )).scalars().all()
608 assert not rows_b, "Repo B must not see Repo A's language rows"
609
610
611 # ─────────────────────────────────────────────────────────────────────────────
612 # Layer T6 — Performance
613 # ─────────────────────────────────────────────────────────────────────────────
614
615 class TestPerformance:
616
617 @pytest.mark.asyncio
618 async def test_T25_provider_completes_100_files_under_2s(
619 self, db_session: AsyncSession
620 ) -> None:
621 """Provider must process a 100-file manifest in < 2 s wall time."""
622 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
623
624 repo = await create_repo(db_session, owner="lnuser", slug="t25-speed")
625 rid = str(repo.repo_id)
626 manifest = {f"src/file_{i}.py": long_id(f"{'0' * 63}{i % 10}") for i in range(100)}
627 await _seed_snapshot(db_session, rid, manifest)
628
629 mock_backend = AsyncMock()
630 mock_backend.get = AsyncMock(return_value=b"# py")
631
632 with (
633 patch("musehub.services.musehub_intel_providers.get_backend",
634 return_value=mock_backend),
635 patch("musehub.services.musehub_intel_providers.parse_symbols",
636 return_value=_fake_tree(10)),
637 patch("musehub.services.musehub_intel_providers.language_of",
638 return_value="Python"),
639 ):
640 t0 = time.monotonic()
641 await _PROVIDER_REGISTRY["intel.code.languages"].compute(
642 db_session, rid, _REF,
643 {"owner": repo.owner, "slug": repo.slug},
644 )
645 elapsed = time.monotonic() - t0
646
647 assert elapsed < 2.0, (
648 f"Provider took {elapsed:.2f}s for 100 files (limit: 2s)"
649 )
650
651 @pytest.mark.asyncio
652 async def test_T26_route_responds_under_200ms_for_50_languages(
653 self, client: AsyncClient, db_session: AsyncSession
654 ) -> None:
655 """Route must respond in < 200 ms when 50 language rows exist."""
656 repo = await create_repo(db_session, owner="lnuser", slug="t26-perf")
657 rid = str(repo.repo_id)
658 rows = [
659 {
660 "repo_id": rid,
661 "language": f"Lang{i:02d}",
662 "file_count": i + 1,
663 "symbol_count": (i + 1) * 100,
664 "pct": 2.0,
665 "kinds_json": {"function": (i + 1) * 50},
666 "ref": _REF,
667 }
668 for i in range(50)
669 ]
670 await db_session.execute(
671 pg_insert(dbm.MusehubIntelLanguages)
672 .values(rows)
673 .on_conflict_do_nothing()
674 )
675 await db_session.commit()
676
677 t0 = time.monotonic()
678 r = await client.get("/lnuser/t26-perf/intel/languages")
679 elapsed = time.monotonic() - t0
680
681 assert r.status_code == 200
682 assert elapsed < 0.2, (
683 f"Route took {elapsed:.3f}s for 50 language rows (limit: 0.2s)"
684 )
685
686 @pytest.mark.asyncio
687 async def test_T27_db_query_uses_lang_index(
688 self, db_session: AsyncSession
689 ) -> None:
690 """SELECT on musehub_intel_languages must use ix_intel_languages_repo index."""
691 explain = await db_session.execute(
692 sa.text(
693 "EXPLAIN SELECT * FROM musehub_intel_languages WHERE repo_id = 'x'"
694 )
695 )
696 plan = " ".join(row[0] for row in explain.all())
697 assert "ix_intel_languages_repo" in plan or "Index" in plan, (
698 f"Query plan does not use ix_intel_languages_repo:\n{plan}"
699 )
700
701
702 # ─────────────────────────────────────────────────────────────────────────────
703 # Layer T7 — Security
704 # ─────────────────────────────────────────────────────────────────────────────
705
706 class TestSecurity:
707
708 @pytest.mark.asyncio
709 async def test_T28_xss_in_language_name_is_escaped(
710 self, client: AsyncClient, db_session: AsyncSession
711 ) -> None:
712 """XSS payload stored in language name must be HTML-escaped in response."""
713 repo = await create_repo(db_session, owner="lnuser", slug="t28-xss")
714 rid = str(repo.repo_id)
715 await _insert_lang_row(
716 db_session, rid,
717 language="<script>alert(1)</script>",
718 file_count=1,
719 )
720 await db_session.commit()
721
722 r = await client.get("/lnuser/t28-xss/intel/languages")
723 assert r.status_code == 200
724 assert "<script>alert" not in r.text, (
725 "XSS in language name not escaped by Jinja2"
726 )
727
728 @pytest.mark.asyncio
729 async def test_T29_sql_injection_in_sort_param_safe(
730 self, client: AsyncClient, ln_repo
731 ) -> None:
732 """SQL injection string in ?sort= must not cause a 500."""
733 r = await client.get(
734 "/lnuser/ln-e2e/intel/languages"
735 "?sort=pct%3B%20DROP%20TABLE%20musehub_intel_languages%3B--"
736 )
737 assert r.status_code in (200, 422), (
738 f"SQL injection in ?sort= caused unexpected status {r.status_code}"
739 )
740
741 @pytest.mark.asyncio
742 async def test_T30_unauthenticated_nonexistent_repo_no_500(
743 self, client: AsyncClient
744 ) -> None:
745 """Accessing a non-existent repo URL must never return 500."""
746 r = await client.get("/attacker/does-not-exist/intel/languages")
747 assert r.status_code != 500, (
748 "Non-existent repo path returned 500 — should be 403 or 404"
749 )
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago