gabriel / musehub public
test_intel_codemap.py python
737 lines 31.3 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """Code Map intel — full 7-tier test suite (issue #21).
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–T05 Layer T1 — DB model (columns, nullable, cascade, meta, index)
9 T06–T11 Layer T2 — Provider (no subprocess, fan_in, fan_out, cycles, edges, empty)
10 T12–T19 Layer T3 — Route (200, empty state, 404, sort, top filter, stat chips, meta)
11 T20–T23 Layer T4 — E2E HTML (stat chips, fan-in bar, cycle panel, dashboard link)
12 T24–T26 Layer T5 — Data integrity (upsert idempotent, meta overwrite, cross-repo)
13 T27–T29 Layer T6 — Performance (provider speed, route speed, index check)
14 T30–T32 Layer T7 — Security (XSS escape, SQL injection, no 500 on bad params)
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 muse.core.types import long_id
30 from musehub.db import musehub_models as dbm
31 from musehub.types.json_types import JSONObject
32 from tests.factories import create_repo
33
34 _REF = long_id("b" * 64)
35
36
37 # ─────────────────────────────────────────────────────────────────────────────
38 # Helpers
39 # ─────────────────────────────────────────────────────────────────────────────
40
41 async def _insert_module_row(
42 session: AsyncSession,
43 repo_id: str,
44 file_path: str,
45 symbol_count: int = 0,
46 fan_in: int = 0,
47 fan_out: int = 0,
48 language: str = "Python",
49 ref: str = _REF,
50 ) -> None:
51 """Upsert one row into musehub_intel_codemap_modules."""
52 await session.execute(
53 pg_insert(dbm.MusehubIntelCodemapModule)
54 .values(
55 repo_id=repo_id,
56 file_path=file_path,
57 symbol_count=symbol_count,
58 fan_in=fan_in,
59 fan_out=fan_out,
60 language=language,
61 ref=ref,
62 )
63 .on_conflict_do_update(
64 index_elements=["repo_id", "file_path"],
65 set_={
66 "symbol_count": symbol_count,
67 "fan_in": fan_in,
68 "fan_out": fan_out,
69 "language": language,
70 "ref": ref,
71 },
72 )
73 )
74
75
76 async def _insert_meta_row(
77 session: AsyncSession,
78 repo_id: str,
79 total_modules: int = 0,
80 total_edges: int = 0,
81 cycle_count: int = 0,
82 cycles_json: list[list[str]] | None = None,
83 ref: str = _REF,
84 ) -> None:
85 """Upsert one row into musehub_intel_codemap_meta."""
86 await session.execute(
87 pg_insert(dbm.MusehubIntelCodemapMeta)
88 .values(
89 repo_id=repo_id,
90 total_modules=total_modules,
91 total_edges=total_edges,
92 cycle_count=cycle_count,
93 cycles_json=cycles_json,
94 ref=ref,
95 )
96 .on_conflict_do_update(
97 index_elements=["repo_id"],
98 set_={
99 "total_modules": total_modules,
100 "total_edges": total_edges,
101 "cycle_count": cycle_count,
102 "cycles_json": cycles_json,
103 "ref": ref,
104 },
105 )
106 )
107
108
109 async def _seed_snapshot(
110 session: AsyncSession,
111 repo_id: str,
112 manifest: dict[str, str],
113 ) -> str:
114 """Insert a MusehubCommit + MusehubSnapshot, return snapshot_id."""
115 import msgpack
116
117 snap_id = long_id("c" * 64)
118 commit_id = long_id("d" * 64)
119
120 await session.execute(
121 pg_insert(dbm.MusehubSnapshot)
122 .values(
123 snapshot_id = snap_id,
124 repo_id = repo_id,
125 directories = [],
126 manifest_blob= msgpack.packb(manifest),
127 entry_count = len(manifest),
128 created_at = datetime(2026, 1, 1, tzinfo=timezone.utc),
129 )
130 .on_conflict_do_nothing()
131 )
132 await session.execute(
133 pg_insert(dbm.MusehubCommit)
134 .values(
135 commit_id = commit_id,
136 repo_id = repo_id,
137 branch = "dev",
138 parent_ids = [],
139 message = "test",
140 author = "cmuser",
141 timestamp = datetime(2026, 1, 1, tzinfo=timezone.utc),
142 snapshot_id = snap_id,
143 )
144 .on_conflict_do_nothing()
145 )
146 await session.commit()
147 return snap_id
148
149
150 def _fake_import_tree(
151 file_path: str,
152 imports: list[str],
153 n_symbols: int = 3,
154 ) -> JSONObject:
155 """Return a SymbolTree with n_symbols functions plus import records."""
156 tree: JSONObject = {}
157 for i in range(n_symbols):
158 tree[f"{file_path}::fn_{i}"] = {
159 "kind": "function",
160 "name": f"fn_{i}",
161 "qualified_name": f"fn_{i}",
162 "content_id": long_id("a" * 64),
163 "body_hash": long_id("b" * 64),
164 "signature_id": long_id("c" * 64),
165 "metadata_id": "",
166 "canonical_key": f"{file_path}##function#fn_{i}#1",
167 "lineno": i + 1,
168 "end_lineno": i + 2,
169 }
170 for j, dotted in enumerate(imports):
171 key = f"{file_path}::_import_{j}"
172 tree[key] = {
173 "kind": "import",
174 "name": dotted.split(".")[-1],
175 "qualified_name": f"import::{dotted}::_sym",
176 "content_id": long_id("e" * 64),
177 "body_hash": "",
178 "signature_id": "",
179 "metadata_id": "",
180 "canonical_key": f"{file_path}##import#{dotted}#0",
181 "lineno": n_symbols + j + 1,
182 "end_lineno": n_symbols + j + 2,
183 }
184 return tree
185
186
187 @pytest_asyncio.fixture
188 async def cm_repo(db_session: AsyncSession):
189 """Repo seeded with 5 module rows and a meta row."""
190 repo = await create_repo(db_session, owner="cmuser", slug="cm-e2e")
191 rid = str(repo.repo_id)
192
193 for i, (fp, fi, fo) in enumerate([
194 ("musehub/api/routes/ui.py", 8, 3),
195 ("musehub/services/svc.py", 5, 4),
196 ("musehub/db/models.py", 4, 1),
197 ("musehub/core/types.py", 3, 0),
198 ("musehub/utils/helpers.py", 1, 2),
199 ]):
200 await _insert_module_row(
201 db_session, rid, fp,
202 symbol_count=10 + i,
203 fan_in=fi, fan_out=fo,
204 language="Python",
205 )
206
207 await _insert_meta_row(
208 db_session, rid,
209 total_modules=5, total_edges=17, cycle_count=0,
210 )
211 await db_session.commit()
212 return repo
213
214
215 # ─────────────────────────────────────────────────────────────────────────────
216 # Layer T1 — DB model
217 # ─────────────────────────────────────────────────────────────────────────────
218
219 class TestDBModel:
220
221 def test_T01_module_model_has_all_required_columns(self) -> None:
222 """MusehubIntelCodemapModule must declare all expected mapped columns."""
223 cols = {
224 c.key
225 for c in sa.inspect(dbm.MusehubIntelCodemapModule).mapper.column_attrs
226 }
227 for required in ("repo_id", "file_path", "symbol_count", "fan_in", "fan_out", "language", "ref"):
228 assert required in cols, f"Column '{required}' missing from MusehubIntelCodemapModule"
229
230 def test_T02_meta_model_has_all_required_columns(self) -> None:
231 """MusehubIntelCodemapMeta must declare all expected mapped columns."""
232 cols = {
233 c.key
234 for c in sa.inspect(dbm.MusehubIntelCodemapMeta).mapper.column_attrs
235 }
236 for required in ("repo_id", "total_modules", "total_edges", "cycle_count", "cycles_json", "ref"):
237 assert required in cols, f"Column '{required}' missing from MusehubIntelCodemapMeta"
238
239 def test_T03_cycles_json_is_nullable(self) -> None:
240 """cycles_json must be nullable — most repos have no cycles."""
241 col = dbm.MusehubIntelCodemapMeta.__table__.c["cycles_json"]
242 assert col.nullable, "cycles_json must be nullable"
243
244 def test_T04_composite_pk_modules(self) -> None:
245 """Primary key of codemap_modules must be (repo_id, file_path)."""
246 pk_cols = {c.name for c in dbm.MusehubIntelCodemapModule.__table__.primary_key.columns}
247 assert pk_cols == {"repo_id", "file_path"}, f"Unexpected PK: {pk_cols}"
248
249 @pytest.mark.asyncio
250 async def test_T05_cascade_delete_removes_module_rows(
251 self, db_session: AsyncSession
252 ) -> None:
253 """Deleting a repo must cascade-delete all codemap module rows."""
254 repo = await create_repo(db_session, owner="cmuser2", slug="cm-cascade")
255 rid = str(repo.repo_id)
256 await _insert_module_row(db_session, rid, "src/a.py", fan_in=1)
257 await db_session.commit()
258
259 await db_session.delete(repo)
260 await db_session.commit()
261
262 result = await db_session.execute(
263 sa.select(dbm.MusehubIntelCodemapModule)
264 .where(dbm.MusehubIntelCodemapModule.repo_id == rid)
265 )
266 assert result.first() is None, "Cascade delete failed — module rows remain"
267
268
269 # ─────────────────────────────────────────────────────────────────────────────
270 # Layer T2 — Provider
271 # ─────────────────────────────────────────────────────────────────────────────
272
273 class TestProvider:
274
275 @pytest.mark.asyncio
276 async def test_T06_provider_returns_no_subprocess(
277 self, db_session: AsyncSession
278 ) -> None:
279 """CodemapProvider.compute must not import subprocess or asyncio.create_subprocess_exec."""
280 import inspect
281 from musehub.services.musehub_intel_providers import CodemapProvider
282 src = inspect.getsource(CodemapProvider.compute)
283 assert "subprocess" not in src, "CodemapProvider.compute spawns a subprocess"
284 assert "create_subprocess" not in src, "CodemapProvider.compute uses create_subprocess"
285
286 @pytest.mark.asyncio
287 async def test_T07_provider_computes_fan_out(
288 self, db_session: AsyncSession
289 ) -> None:
290 """fan_out counts resolved imports from the manifest, not stdlib."""
291 from musehub.services.musehub_intel_providers import CodemapProvider
292
293 repo = await create_repo(db_session, owner="cmuser3", slug="cm-fanout")
294 rid = str(repo.repo_id)
295
296 # a.py imports b.py; c.py is stdlib (unresolved)
297 manifest = {
298 "src/a.py": long_id("a" * 64),
299 "src/b.py": long_id("b" * 64),
300 }
301 await _seed_snapshot(db_session, rid, manifest)
302
303 a_tree = _fake_import_tree("src/a.py", ["src.b"], n_symbols=2)
304 b_tree = _fake_import_tree("src/b.py", ["os.path"], n_symbols=1)
305
306 mock_backend = AsyncMock()
307 mock_backend.get = AsyncMock(side_effect=[b"src_a", b"src_b"])
308
309 with (
310 patch("musehub.services.musehub_intel_providers.get_backend", return_value=mock_backend),
311 patch("musehub.services.musehub_intel_providers.parse_symbols", side_effect=[a_tree, b_tree]),
312 patch("musehub.services.musehub_intel_providers.language_of", return_value="Python"),
313 ):
314 provider = CodemapProvider()
315 results = await provider.compute(db_session, rid, "dev", {"owner": "cmuser3", "slug": "cm-fanout"})
316
317 assert results, "Provider returned empty results"
318 assert results[0][0] == "intel.code.codemap"
319 data = results[0][1]
320 # a.py resolves src.b → 1 edge; b.py resolves os.path → 0 (stdlib)
321 assert data["edges"] == 1, f"Expected 1 edge, got {data['edges']}"
322
323 @pytest.mark.asyncio
324 async def test_T08_provider_computes_fan_in(
325 self, db_session: AsyncSession
326 ) -> None:
327 """fan_in of b.py must equal number of files that import b.py."""
328 from musehub.services.musehub_intel_providers import CodemapProvider
329
330 repo = await create_repo(db_session, owner="cmuser4", slug="cm-fanin")
331 rid = str(repo.repo_id)
332
333 manifest = {
334 "src/a.py": long_id("a" * 64),
335 "src/b.py": long_id("b" * 64),
336 "src/c.py": long_id("c" * 64),
337 }
338 await _seed_snapshot(db_session, rid, manifest)
339
340 a_tree = _fake_import_tree("src/a.py", ["src.b"], n_symbols=1)
341 b_tree = _fake_import_tree("src/b.py", [], n_symbols=1)
342 c_tree = _fake_import_tree("src/c.py", ["src.b"], n_symbols=1)
343
344 mock_backend = AsyncMock()
345 mock_backend.get = AsyncMock(side_effect=[b"src_a", b"src_b", b"src_c"])
346
347 with (
348 patch("musehub.services.musehub_intel_providers.get_backend", return_value=mock_backend),
349 patch("musehub.services.musehub_intel_providers.parse_symbols", side_effect=[a_tree, b_tree, c_tree]),
350 patch("musehub.services.musehub_intel_providers.language_of", return_value="Python"),
351 ):
352 provider = CodemapProvider()
353 await provider.compute(db_session, rid, "dev", {"owner": "cmuser4", "slug": "cm-fanin"})
354 await db_session.commit()
355
356 result = await db_session.execute(
357 sa.select(dbm.MusehubIntelCodemapModule)
358 .where(
359 dbm.MusehubIntelCodemapModule.repo_id == rid,
360 dbm.MusehubIntelCodemapModule.file_path == "src/b.py",
361 )
362 )
363 row = result.scalar_one_or_none()
364 assert row is not None, "src/b.py row not found"
365 assert row.fan_in == 2, f"Expected fan_in=2, got {row.fan_in}"
366
367 @pytest.mark.asyncio
368 async def test_T09_provider_detects_no_cycles_for_dag(
369 self, db_session: AsyncSession
370 ) -> None:
371 """A pure DAG import graph must produce cycle_count=0."""
372 from musehub.services.musehub_intel_providers import CodemapProvider
373
374 repo = await create_repo(db_session, owner="cmuser5", slug="cm-nocycle")
375 rid = str(repo.repo_id)
376
377 manifest = {"src/a.py": long_id("a" * 64), "src/b.py": long_id("b" * 64)}
378 await _seed_snapshot(db_session, rid, manifest)
379
380 a_tree = _fake_import_tree("src/a.py", ["src.b"])
381 b_tree = _fake_import_tree("src/b.py", [])
382
383 mock_backend = AsyncMock()
384 mock_backend.get = AsyncMock(side_effect=[b"a", b"b"])
385
386 with (
387 patch("musehub.services.musehub_intel_providers.get_backend", return_value=mock_backend),
388 patch("musehub.services.musehub_intel_providers.parse_symbols", side_effect=[a_tree, b_tree]),
389 patch("musehub.services.musehub_intel_providers.language_of", return_value="Python"),
390 ):
391 results = await CodemapProvider().compute(db_session, rid, "dev", {"owner": "cmuser5", "slug": "cm-nocycle"})
392
393 assert results[0][1]["cycles"] == 0
394
395 @pytest.mark.asyncio
396 async def test_T10_provider_detects_mutual_import_cycle(
397 self, db_session: AsyncSession
398 ) -> None:
399 """A ↔ B mutual import must be detected as one cycle."""
400 from musehub.services.musehub_intel_providers import CodemapProvider
401
402 repo = await create_repo(db_session, owner="cmuser6", slug="cm-cycle")
403 rid = str(repo.repo_id)
404
405 manifest = {"src/a.py": long_id("a" * 64), "src/b.py": long_id("b" * 64)}
406 await _seed_snapshot(db_session, rid, manifest)
407
408 # a imports b AND b imports a → cycle
409 a_tree = _fake_import_tree("src/a.py", ["src.b"])
410 b_tree = _fake_import_tree("src/b.py", ["src.a"])
411
412 mock_backend = AsyncMock()
413 mock_backend.get = AsyncMock(side_effect=[b"a", b"b"])
414
415 with (
416 patch("musehub.services.musehub_intel_providers.get_backend", return_value=mock_backend),
417 patch("musehub.services.musehub_intel_providers.parse_symbols", side_effect=[a_tree, b_tree]),
418 patch("musehub.services.musehub_intel_providers.language_of", return_value="Python"),
419 ):
420 results = await CodemapProvider().compute(db_session, rid, "dev", {"owner": "cmuser6", "slug": "cm-cycle"})
421
422 assert results[0][1]["cycles"] == 1, f"Expected 1 cycle, got {results[0][1]['cycles']}"
423
424 @pytest.mark.asyncio
425 async def test_T11_provider_returns_empty_for_missing_manifest(
426 self, db_session: AsyncSession
427 ) -> None:
428 """Provider must return [] when no commits exist for the repo."""
429 from musehub.services.musehub_intel_providers import CodemapProvider
430
431 repo = await create_repo(db_session, owner="cmuser7", slug="cm-empty")
432 rid = str(repo.repo_id)
433 await db_session.commit()
434
435 results = await CodemapProvider().compute(db_session, rid, "dev", {"owner": "cmuser7", "slug": "cm-empty"})
436 assert results == [], f"Expected [], got {results}"
437
438
439 # ─────────────────────────────────────────────────────────────────────────────
440 # Layer T3 — Route
441 # ─────────────────────────────────────────────────────────────────────────────
442
443 class TestRoute:
444
445 @pytest.mark.asyncio
446 async def test_T12_codemap_page_returns_200(
447 self, client: AsyncClient, cm_repo
448 ) -> None:
449 """GET /cmuser/cm-e2e/intel/codemap must return HTTP 200."""
450 resp = await client.get("/cmuser/cm-e2e/intel/codemap")
451 assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
452
453 @pytest.mark.asyncio
454 async def test_T13_codemap_page_empty_state(
455 self, client: AsyncClient, db_session: AsyncSession
456 ) -> None:
457 """Route must render empty state when no codemap rows exist."""
458 repo = await create_repo(db_session, owner="cmempty", slug="cm-nodata")
459 await db_session.commit()
460 resp = await client.get("/cmempty/cm-nodata/intel/codemap")
461 assert resp.status_code == 200
462 assert "Push a commit" in resp.text
463
464 @pytest.mark.asyncio
465 async def test_T14_codemap_page_404_for_missing_repo(
466 self, client: AsyncClient
467 ) -> None:
468 """Route must return 404 for a repo that does not exist."""
469 resp = await client.get("/ghost/no-such-repo/intel/codemap")
470 assert resp.status_code == 404
471
472 @pytest.mark.asyncio
473 async def test_T15_sort_by_fan_in(
474 self, client: AsyncClient, cm_repo
475 ) -> None:
476 """sort=fan-in must return modules ordered by fan_in descending."""
477 resp = await client.get("/cmuser/cm-e2e/intel/codemap?sort=fan-in")
478 assert resp.status_code == 200
479 # musehub/api/routes/ui.py has fan_in=8, must appear before others
480 text = resp.text
481 pos_api = text.find("ui.py")
482 pos_types = text.find("types.py")
483 assert pos_api < pos_types, "fan-in sort: ui.py (fi=8) should appear before types.py (fi=3)"
484
485 @pytest.mark.asyncio
486 async def test_T16_sort_by_fan_out(
487 self, client: AsyncClient, cm_repo
488 ) -> None:
489 """sort=fan-out must return modules ordered by fan_out descending."""
490 resp = await client.get("/cmuser/cm-e2e/intel/codemap?sort=fan-out")
491 assert resp.status_code == 200
492 text = resp.text
493 # svc.py has fan_out=4, must appear before models.py (fo=1)
494 pos_svc = text.find("svc.py")
495 pos_models = text.find("models.py")
496 assert pos_svc < pos_models, "fan-out sort: svc.py (fo=4) should appear before models.py (fo=1)"
497
498 @pytest.mark.asyncio
499 async def test_T17_unknown_sort_coerces_to_symbols(
500 self, client: AsyncClient, cm_repo
501 ) -> None:
502 """An unknown sort param must be coerced to 'symbols' (no 400/500)."""
503 resp = await client.get("/cmuser/cm-e2e/intel/codemap?sort=invalid_sort")
504 assert resp.status_code == 200
505
506 @pytest.mark.asyncio
507 async def test_T18_top_filter_limits_rows(
508 self, client: AsyncClient, db_session: AsyncSession
509 ) -> None:
510 """top=20 with 25 total modules must show max 20 rows but stat chip shows 25."""
511 repo = await create_repo(db_session, owner="cmtop", slug="cm-top")
512 rid = str(repo.repo_id)
513 for i in range(25):
514 await _insert_module_row(db_session, rid, f"src/mod_{i}.py", fan_in=i)
515 await _insert_meta_row(db_session, rid, total_modules=25, total_edges=0)
516 await db_session.commit()
517
518 resp = await client.get("/cmtop/cm-top/intel/codemap?top=20")
519 assert resp.status_code == 200
520 # stat chip must show 25, not 20
521 assert "25" in resp.text, "Stat chip must show total module count (25), not page length"
522
523 @pytest.mark.asyncio
524 async def test_T19_stat_chips_use_meta_row(
525 self, client: AsyncClient, cm_repo
526 ) -> None:
527 """Stat chips must display meta row values (5 modules, 17 edges, 0 cycles)."""
528 resp = await client.get("/cmuser/cm-e2e/intel/codemap")
529 assert resp.status_code == 200
530 assert "17" in resp.text, "Edge count from meta row (17) not found in response"
531
532
533 # ─────────────────────────────────────────────────────────────────────────────
534 # Layer T4 — E2E HTML
535 # ─────────────────────────────────────────────────────────────────────────────
536
537 class TestHTML:
538
539 @pytest.mark.asyncio
540 async def test_T20_stat_chips_present(
541 self, client: AsyncClient, cm_repo
542 ) -> None:
543 """HTML must contain stat chip labels: Modules, Edges, Cycles."""
544 resp = await client.get("/cmuser/cm-e2e/intel/codemap")
545 assert resp.status_code == 200
546 for label in ("Modules", "Edges", "Cycles"):
547 assert label in resp.text, f"Stat chip label '{label}' missing"
548
549 @pytest.mark.asyncio
550 async def test_T21_fan_in_bar_present(
551 self, client: AsyncClient, cm_repo
552 ) -> None:
553 """HTML must contain cm-fan-bar element for fan-in visualization."""
554 resp = await client.get("/cmuser/cm-e2e/intel/codemap")
555 assert resp.status_code == 200
556 assert "cm-fan-bar" in resp.text, "cm-fan-bar class missing from HTML"
557
558 @pytest.mark.asyncio
559 async def test_T22_cycle_ok_shown_when_no_cycles(
560 self, client: AsyncClient, cm_repo
561 ) -> None:
562 """cm-cycle-ok element must be present when cycle_count == 0."""
563 resp = await client.get("/cmuser/cm-e2e/intel/codemap")
564 assert resp.status_code == 200
565 assert "cm-cycle-ok" in resp.text, "cm-cycle-ok class missing (cycle_count=0)"
566
567 @pytest.mark.asyncio
568 async def test_T23_dashboard_back_link_present(
569 self, client: AsyncClient, cm_repo
570 ) -> None:
571 """Page must contain a back link to the intel dashboard."""
572 resp = await client.get("/cmuser/cm-e2e/intel/codemap")
573 assert resp.status_code == 200
574 assert "/intel" in resp.text, "Back link to Intel Hub missing"
575
576
577 # ─────────────────────────────────────────────────────────────────────────────
578 # Layer T5 — Data integrity
579 # ─────────────────────────────────────────────────────────────────────────────
580
581 class TestDataIntegrity:
582
583 @pytest.mark.asyncio
584 async def test_T24_upsert_is_idempotent(
585 self, db_session: AsyncSession
586 ) -> None:
587 """Upserting the same module row twice must not create duplicates."""
588 repo = await create_repo(db_session, owner="cmdup", slug="cm-dup")
589 rid = str(repo.repo_id)
590
591 for _ in range(2):
592 await _insert_module_row(db_session, rid, "src/a.py", fan_in=3)
593 await db_session.commit()
594
595 result = await db_session.execute(
596 sa.select(sa.func.count())
597 .select_from(dbm.MusehubIntelCodemapModule)
598 .where(dbm.MusehubIntelCodemapModule.repo_id == rid)
599 )
600 count = result.scalar_one()
601 assert count == 1, f"Upsert created duplicate: expected 1 row, got {count}"
602
603 @pytest.mark.asyncio
604 async def test_T25_meta_upsert_overwrites(
605 self, db_session: AsyncSession
606 ) -> None:
607 """Second meta upsert must overwrite total_edges, not create a second row."""
608 repo = await create_repo(db_session, owner="cmmeta", slug="cm-meta")
609 rid = str(repo.repo_id)
610
611 await _insert_meta_row(db_session, rid, total_modules=5, total_edges=10)
612 await _insert_meta_row(db_session, rid, total_modules=6, total_edges=20)
613 await db_session.commit()
614
615 result = await db_session.execute(
616 sa.select(dbm.MusehubIntelCodemapMeta)
617 .where(dbm.MusehubIntelCodemapMeta.repo_id == rid)
618 )
619 rows = result.scalars().all()
620 assert len(rows) == 1, f"Expected 1 meta row, got {len(rows)}"
621 assert rows[0].total_edges == 20, f"Expected total_edges=20, got {rows[0].total_edges}"
622
623 @pytest.mark.asyncio
624 async def test_T26_cross_repo_isolation(
625 self, db_session: AsyncSession
626 ) -> None:
627 """Module rows from repo A must not appear in repo B queries."""
628 repo_a = await create_repo(db_session, owner="cmisolate", slug="repo-a")
629 repo_b = await create_repo(db_session, owner="cmisolate", slug="repo-b")
630 rid_a = str(repo_a.repo_id)
631 rid_b = str(repo_b.repo_id)
632
633 await _insert_module_row(db_session, rid_a, "src/a.py", fan_in=5)
634 await db_session.commit()
635
636 result = await db_session.execute(
637 sa.select(dbm.MusehubIntelCodemapModule)
638 .where(dbm.MusehubIntelCodemapModule.repo_id == rid_b)
639 )
640 assert result.first() is None, "Cross-repo contamination: repo B sees repo A rows"
641
642
643 # ─────────────────────────────────────────────────────────────────────────────
644 # Layer T6 — Performance
645 # ─────────────────────────────────────────────────────────────────────────────
646
647 class TestPerformance:
648
649 @pytest.mark.asyncio
650 async def test_T27_provider_completes_under_threshold(
651 self, db_session: AsyncSession
652 ) -> None:
653 """CodemapProvider must complete within 10 s for a 50-file manifest."""
654 from musehub.services.musehub_intel_providers import CodemapProvider
655
656 repo = await create_repo(db_session, owner="cmperfp", slug="cm-perf-p")
657 rid = str(repo.repo_id)
658 n = 50
659 manifest = {f"src/mod_{i}.py": long_id(f"{i:064x}") for i in range(n)}
660 await _seed_snapshot(db_session, rid, manifest)
661
662 def _tree_for(path: str) -> JSONObject:
663 return _fake_import_tree(path, [], n_symbols=5)
664
665 trees = [_tree_for(fp) for fp in manifest]
666 src_bytes = [b"src"] * n
667 mock_backend = AsyncMock()
668 mock_backend.get = AsyncMock(side_effect=src_bytes)
669
670 with (
671 patch("musehub.services.musehub_intel_providers.get_backend", return_value=mock_backend),
672 patch("musehub.services.musehub_intel_providers.parse_symbols", side_effect=trees),
673 patch("musehub.services.musehub_intel_providers.language_of", return_value="Python"),
674 ):
675 t0 = time.monotonic()
676 await CodemapProvider().compute(db_session, rid, "dev", {"owner": "cmperfp", "slug": "cm-perf-p"})
677 elapsed = time.monotonic() - t0
678
679 assert elapsed < 10.0, f"Provider took {elapsed:.2f}s — exceeds 10s threshold"
680
681 @pytest.mark.asyncio
682 async def test_T28_route_responds_under_500ms(
683 self, client: AsyncClient, cm_repo
684 ) -> None:
685 """GET /intel/codemap must respond in under 500 ms for a seeded repo."""
686 t0 = time.monotonic()
687 resp = await client.get("/cmuser/cm-e2e/intel/codemap")
688 elapsed = time.monotonic() - t0
689 assert resp.status_code == 200
690 assert elapsed < 0.5, f"Route took {elapsed:.3f}s — exceeds 500ms threshold"
691
692 def test_T29_module_table_has_repo_index(self) -> None:
693 """ix_intel_codemap_modules_repo index must exist on the ORM model."""
694 table = dbm.MusehubIntelCodemapModule.__table__
695 index_names = {idx.name for idx in table.indexes}
696 assert "ix_intel_codemap_modules_repo" in index_names, (
697 f"Index missing. Found: {index_names}"
698 )
699
700
701 # ─────────────────────────────────────────────────────────────────────────────
702 # Layer T7 — Security
703 # ─────────────────────────────────────────────────────────────────────────────
704
705 class TestSecurity:
706
707 @pytest.mark.asyncio
708 async def test_T30_xss_file_path_is_escaped(
709 self, client: AsyncClient, db_session: AsyncSession
710 ) -> None:
711 """file_path containing HTML must be escaped in the rendered page."""
712 repo = await create_repo(db_session, owner="cmxss", slug="cm-xss")
713 rid = str(repo.repo_id)
714 xss_path = "src/<script>alert(1)</script>.py"
715 await _insert_module_row(db_session, rid, xss_path, fan_in=1)
716 await _insert_meta_row(db_session, rid, total_modules=1, total_edges=0)
717 await db_session.commit()
718
719 resp = await client.get("/cmxss/cm-xss/intel/codemap")
720 assert resp.status_code == 200
721 assert "<script>alert(1)</script>" not in resp.text, "Unescaped XSS payload in response"
722
723 @pytest.mark.asyncio
724 async def test_T31_sql_injection_in_sort_param_is_safe(
725 self, client: AsyncClient, cm_repo
726 ) -> None:
727 """SQL injection attempt in sort param must return 200, not 500."""
728 resp = await client.get("/cmuser/cm-e2e/intel/codemap?sort=symbols%3BDROP+TABLE+musehub_intel_codemap_modules")
729 assert resp.status_code == 200
730
731 @pytest.mark.asyncio
732 async def test_T32_invalid_top_param_does_not_500(
733 self, client: AsyncClient, cm_repo
734 ) -> None:
735 """Invalid top param (non-integer string) must not return 500."""
736 resp = await client.get("/cmuser/cm-e2e/intel/codemap?top=INVALID")
737 assert resp.status_code == 200
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago