gabriel / musehub public
test_snapshot_symbol_indexer.py python
984 lines 41.0 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """Tests for the Snapshot & Symbol Indexer — Section 5 of test-coverage-checklist.md.
2
3 Complements test_snapshot_entries.py (14 tests on the snapshot write/read path).
4 This file focuses on the symbol indexer and the gaps not covered there.
5
6 Coverage layers
7 ───────────────
8 Unit — _extract_ops (flat/nested child_ops, missing address, non-dict delta);
9 _op_to_muse_op (all mapping keys, unknown passthrough).
10 Integration — build_symbol_index: empty list when no structured_delta; returns results
11 for repos with structured_delta; correct symbol_history/hash_occurrence
12 content; upsert semantics (only one row per repo/intel_type); BFS
13 excludes orphaned commits.
14 load_symbol_history: empty when no index; with/without file_path filter.
15 load_hash_occurrence: empty when no index; correct content.
16 get_index_meta: None/present states.
17 load_intel_snapshot: None/present states.
18 get_snapshot_manifests_batch: empty list, single, multi-snapshot.
19 Data — upsert_snapshot_entries atomic replace (stale entries removed);
20 build_symbol_index + persist_intel_results upserts on rebuild;
21 BFS reachability excludes orphaned branches.
22 Security — Corrupt JSON blob returns {} not exception;
23 build_symbol_index with unknown head_commit_id returns empty list.
24 Stress — upsert_snapshot_entries with 1 000-file manifest;
25 get_snapshot_manifests_batch with 50 snapshots in one query;
26 build_symbol_index with 100 commits (10 ops each);
27 load_symbol_history file_path filter on large index.
28 Performance — _extract_ops 1 000 calls < 100 ms;
29 build_symbol_index 100 commits < 3 s.
30 E2E — Full pipeline: commits with structured_delta → build_symbol_index →
31 persist_intel_results → get_index_meta returns correct ref;
32 rebuild replaces previous result; symbol list HTTP page returns 200.
33 """
34 from __future__ import annotations
35
36 import json
37 import time
38 import uuid
39 from datetime import datetime, timezone
40
41 import pytest
42 from sqlalchemy.ext.asyncio import AsyncSession
43
44 from musehub.db import musehub_models as db
45 from tests.factories import create_repo
46 from musehub.types.json_types import JSONObject
47
48
49 # ─────────────────────────────────────────────────────────────────────────────
50 # Helpers
51 # ─────────────────────────────────────────────────────────────────────────────
52
53 def _now() -> datetime:
54 return datetime.now(tz=timezone.utc)
55
56
57 async def _commit_with_delta(
58 session: AsyncSession,
59 repo_id: str,
60 commit_id: str,
61 ops: list[JSONObject],
62 parent_ids: list[str] | None = None,
63 branch: str = "main",
64 author: str = "gabriel",
65 ) -> db.MusehubCommit:
66 """Insert a commit whose commit_meta carries a structured_delta."""
67 commit = db.MusehubCommit(
68 commit_id=commit_id,
69 repo_id=repo_id,
70 branch=branch,
71 parent_ids=parent_ids or [],
72 message="feat: test commit",
73 author=author,
74 timestamp=_now(),
75 commit_meta={"structured_delta": {"ops": ops}},
76 )
77 session.add(commit)
78 await session.flush()
79 return commit
80
81
82 def _insert_op(address: str, content_id: str = "sha256:abc") -> JSONObject:
83 return {"address": address, "op": "insert", "content_id": content_id}
84
85
86 def _patch_op(file_addr: str, children: list[JSONObject]) -> JSONObject:
87 return {"address": file_addr, "op": "patch", "child_ops": children}
88
89
90 async def _build_and_persist(
91 session: AsyncSession,
92 repo_id: str,
93 commit_id: str,
94 ) -> list[tuple[str, dict]]:
95 """Build symbol index and persist results; returns the result list."""
96 from musehub.services.musehub_symbol_indexer import build_symbol_index
97 from musehub.services.musehub_intel_providers import persist_intel_results
98 results = await build_symbol_index(session, repo_id, commit_id)
99 if results:
100 await persist_intel_results(session, repo_id, commit_id, results)
101 return results
102
103
104 def _get_result_data(results: list[tuple[str, dict]], intel_type: str) -> dict:
105 """Extract data dict for a specific intel_type from the results list."""
106 for t, data in results:
107 if t == intel_type:
108 return data
109 return {}
110
111
112 # ─────────────────────────────────────────────────────────────────────────────
113 # Layer 1 — Unit: pure functions
114 # ─────────────────────────────────────────────────────────────────────────────
115
116 class TestExtractOps:
117 """_extract_ops pulls a flat list of ops including child_ops."""
118
119 def _run(self, commit_meta: JSONObject) -> list[JSONObject]:
120 from musehub.services.musehub_symbol_indexer import _extract_ops
121 return _extract_ops(commit_meta)
122
123 def test_no_structured_delta_returns_empty(self) -> None:
124 assert self._run({}) == []
125
126 def test_none_delta_returns_empty(self) -> None:
127 assert self._run({"structured_delta": None}) == []
128
129 def test_non_dict_delta_returns_empty(self) -> None:
130 assert self._run({"structured_delta": "bad"}) == []
131
132 def test_flat_ops_without_child_ops(self) -> None:
133 meta = {
134 "structured_delta": {
135 "ops": [
136 {"address": "main.py::Foo", "op": "insert"},
137 {"address": "main.py::Bar", "op": "delete"},
138 ]
139 }
140 }
141 result = self._run(meta)
142 assert len(result) == 2
143 assert result[0]["address"] == "main.py::Foo"
144 assert result[1]["address"] == "main.py::Bar"
145
146 def test_patch_op_with_child_ops_flattened(self) -> None:
147 meta = {
148 "structured_delta": {
149 "ops": [
150 {
151 "address": "src/app.py",
152 "op": "patch",
153 "child_ops": [
154 {"address": "src/app.py::MyClass", "op": "insert"},
155 {"address": "src/app.py::MyClass.run", "op": "insert"},
156 ],
157 }
158 ]
159 }
160 }
161 result = self._run(meta)
162 # 1 top-level + 2 child_ops
163 assert len(result) == 3
164 addresses = [op["address"] for op in result]
165 assert "src/app.py" in addresses
166 assert "src/app.py::MyClass" in addresses
167 assert "src/app.py::MyClass.run" in addresses
168
169 def test_op_without_address_skipped(self) -> None:
170 meta = {
171 "structured_delta": {
172 "ops": [
173 {"op": "insert"}, # no address
174 {"address": "ok.py", "op": "insert"},
175 ]
176 }
177 }
178 result = self._run(meta)
179 assert len(result) == 1
180 assert result[0]["address"] == "ok.py"
181
182 def test_child_op_without_address_skipped(self) -> None:
183 meta = {
184 "structured_delta": {
185 "ops": [
186 {
187 "address": "file.py",
188 "op": "patch",
189 "child_ops": [
190 {"op": "insert"}, # no address — must be skipped
191 {"address": "file.py::Good", "op": "insert"},
192 ],
193 }
194 ]
195 }
196 }
197 result = self._run(meta)
198 addresses = [op["address"] for op in result]
199 assert "file.py::Good" in addresses
200 for op in result:
201 assert "address" in op
202
203 def test_non_dict_op_skipped(self) -> None:
204 meta = {"structured_delta": {"ops": ["not-a-dict", {"address": "f.py", "op": "add"}]}}
205 result = self._run(meta)
206 assert len(result) == 1
207
208
209 class TestOpToMuseOp:
210 """_op_to_muse_op maps DomainOp vocabulary to muse history vocabulary."""
211
212 def _run(self, op_type: str) -> str:
213 from musehub.services.musehub_symbol_indexer import _op_to_muse_op
214 return _op_to_muse_op(op_type)
215
216 def test_insert_maps_to_add(self) -> None:
217 assert self._run("insert") == "add"
218
219 def test_delete_maps_to_delete(self) -> None:
220 assert self._run("delete") == "delete"
221
222 def test_replace_maps_to_modify(self) -> None:
223 assert self._run("replace") == "modify"
224
225 def test_patch_maps_to_modify(self) -> None:
226 assert self._run("patch") == "modify"
227
228 def test_move_maps_to_move(self) -> None:
229 assert self._run("move") == "move"
230
231 def test_directory_rename_maps_to_rename(self) -> None:
232 assert self._run("directory_rename") == "rename"
233
234 def test_unknown_op_passthrough(self) -> None:
235 assert self._run("frobnicate") == "frobnicate"
236
237 def test_empty_string_passthrough(self) -> None:
238 assert self._run("") == ""
239
240
241 # ─────────────────────────────────────────────────────────────────────────────
242 # Layer 2 — Integration: build_symbol_index + read functions
243 # ─────────────────────────────────────────────────────────────────────────────
244
245 class TestBuildSymbolIndex:
246 @pytest.mark.asyncio
247 async def test_returns_empty_when_no_structured_delta(
248 self, db_session: AsyncSession
249 ) -> None:
250 from musehub.services.musehub_symbol_indexer import build_symbol_index
251 from tests.factories import create_commit
252
253 repo = await create_repo(db_session, slug="idx-nodelta")
254 commit = await create_commit(db_session, repo.repo_id, branch="main")
255
256 results = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
257 assert results == []
258
259 @pytest.mark.asyncio
260 async def test_returns_results_for_structured_delta(
261 self, db_session: AsyncSession
262 ) -> None:
263 repo = await create_repo(db_session, slug="idx-creates")
264 commit = await _commit_with_delta(
265 db_session, repo.repo_id, "c001",
266 ops=[_insert_op("main.py::Foo", "sha256:aaa")],
267 )
268
269 results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
270 await db_session.commit()
271
272 assert results
273 types = {t for t, _ in results}
274 assert "code.symbol_history" in types
275 assert "code.hash_occurrence" in types
276
277 @pytest.mark.asyncio
278 async def test_symbol_history_contains_correct_entries(
279 self, db_session: AsyncSession
280 ) -> None:
281 repo = await create_repo(db_session, slug="idx-symhist")
282 commit = await _commit_with_delta(
283 db_session, repo.repo_id, "c002",
284 ops=[
285 _insert_op("src/app.py::MyClass", "sha256:class"),
286 _insert_op("src/app.py::my_func", "sha256:func"),
287 ],
288 )
289
290 results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
291 await db_session.commit()
292
293 sh_data = _get_result_data(results, "code.symbol_history")
294 entries = sh_data.get("entries", {})
295 assert "src/app.py::MyClass" in entries
296 assert "src/app.py::my_func" in entries
297 assert entries["src/app.py::MyClass"][0]["op"] == "add"
298
299 @pytest.mark.asyncio
300 async def test_hash_occurrence_tracks_shared_content(
301 self, db_session: AsyncSession
302 ) -> None:
303 repo = await create_repo(db_session, slug="idx-hashoc")
304 shared_hash = "sha256:shared"
305 commit = await _commit_with_delta(
306 db_session, repo.repo_id, "c003",
307 ops=[
308 _insert_op("a.py::Foo", shared_hash),
309 _insert_op("b.py::Bar", shared_hash),
310 ],
311 )
312
313 results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
314 await db_session.commit()
315
316 ho_data = _get_result_data(results, "code.hash_occurrence")
317 entries = ho_data.get("entries", {})
318 assert shared_hash in entries
319 assert set(entries[shared_hash]) == {"a.py::Foo", "b.py::Bar"}
320
321 @pytest.mark.asyncio
322 async def test_rebuild_upserts_one_row_per_intel_type(
323 self, db_session: AsyncSession
324 ) -> None:
325 from sqlalchemy import select, func
326
327 repo = await create_repo(db_session, slug="idx-prune")
328 c1 = await _commit_with_delta(db_session, repo.repo_id, "c100",
329 ops=[_insert_op("f.py::A")])
330 await _build_and_persist(db_session, repo.repo_id, c1.commit_id)
331 await db_session.commit()
332
333 c2 = await _commit_with_delta(db_session, repo.repo_id, "c101",
334 ops=[_insert_op("f.py::B")])
335 await _build_and_persist(db_session, repo.repo_id, c2.commit_id)
336 await db_session.commit()
337
338 # Each intel_type should have exactly one row per repo (upserted)
339 count = (await db_session.execute(
340 select(func.count()).select_from(db.MusehubIntelResult).where(
341 db.MusehubIntelResult.repo_id == repo.repo_id,
342 db.MusehubIntelResult.intel_type == "code.symbol_history",
343 )
344 )).scalar_one()
345 assert count == 1
346
347 @pytest.mark.asyncio
348 async def test_bfs_excludes_orphaned_commits(
349 self, db_session: AsyncSession
350 ) -> None:
351 """Commits not reachable from head must not appear in the symbol index."""
352 from musehub.services.musehub_symbol_indexer import load_symbol_history
353
354 repo = await create_repo(db_session, slug="idx-bfs")
355 await _commit_with_delta(
356 db_session, repo.repo_id, "orphan",
357 ops=[_insert_op("orphan.py::OrphanSym", "sha256:orphan")],
358 parent_ids=[],
359 )
360 head = await _commit_with_delta(
361 db_session, repo.repo_id, "head",
362 ops=[_insert_op("main.py::RealSym", "sha256:real")],
363 parent_ids=[],
364 )
365
366 await _build_and_persist(db_session, repo.repo_id, head.commit_id)
367 await db_session.commit()
368
369 history = await load_symbol_history(db_session, repo.repo_id)
370 assert "main.py::RealSym" in history
371 assert "orphan.py::OrphanSym" not in history
372
373
374 class TestLoadFunctions:
375 @pytest.mark.asyncio
376 async def test_load_symbol_history_empty_when_no_index(
377 self, db_session: AsyncSession
378 ) -> None:
379 from musehub.services.musehub_symbol_indexer import load_symbol_history
380 repo = await create_repo(db_session, slug="load-noindex")
381 result = await load_symbol_history(db_session, repo.repo_id)
382 assert result == {}
383
384 @pytest.mark.asyncio
385 async def test_load_symbol_history_with_file_path_filter(
386 self, db_session: AsyncSession
387 ) -> None:
388 from musehub.services.musehub_symbol_indexer import load_symbol_history
389
390 repo = await create_repo(db_session, slug="load-filter")
391 commit = await _commit_with_delta(
392 db_session, repo.repo_id, "cF01",
393 ops=[
394 _insert_op("a.py::Foo", "sha256:x"),
395 _insert_op("a.py", "sha256:file"),
396 _insert_op("b.py::Bar", "sha256:y"),
397 ],
398 )
399 await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
400 await db_session.commit()
401
402 result = await load_symbol_history(db_session, repo.repo_id, file_path="a.py")
403 assert "a.py::Foo" in result
404 assert "a.py" in result
405 assert "b.py::Bar" not in result
406
407 @pytest.mark.asyncio
408 async def test_load_hash_occurrence_empty_when_no_index(
409 self, db_session: AsyncSession
410 ) -> None:
411 from musehub.services.musehub_symbol_indexer import load_hash_occurrence
412 repo = await create_repo(db_session, slug="hash-noindex")
413 assert await load_hash_occurrence(db_session, repo.repo_id) == {}
414
415 @pytest.mark.asyncio
416 async def test_load_hash_occurrence_returns_correct_entries(
417 self, db_session: AsyncSession
418 ) -> None:
419 from musehub.services.musehub_symbol_indexer import load_hash_occurrence
420
421 repo = await create_repo(db_session, slug="hash-entries")
422 commit = await _commit_with_delta(
423 db_session, repo.repo_id, "cH01",
424 ops=[_insert_op("x.py::X", "sha256:hash1"), _insert_op("y.py::Y", "sha256:hash1")],
425 )
426 await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
427 await db_session.commit()
428
429 result = await load_hash_occurrence(db_session, repo.repo_id)
430 assert "sha256:hash1" in result
431 assert set(result["sha256:hash1"]) == {"x.py::X", "y.py::Y"}
432
433 @pytest.mark.asyncio
434 async def test_get_index_meta_none_when_no_index(
435 self, db_session: AsyncSession
436 ) -> None:
437 from musehub.services.musehub_symbol_indexer import get_index_meta
438 repo = await create_repo(db_session, slug="meta-none")
439 assert await get_index_meta(db_session, repo.repo_id) is None
440
441 @pytest.mark.asyncio
442 async def test_get_index_meta_returns_ref_and_symbol_count(
443 self, db_session: AsyncSession
444 ) -> None:
445 from musehub.services.musehub_symbol_indexer import get_index_meta
446
447 repo = await create_repo(db_session, slug="meta-ok")
448 commit = await _commit_with_delta(
449 db_session, repo.repo_id, "cM01",
450 ops=[_insert_op("f.py::A"), _insert_op("f.py::B")],
451 )
452 await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
453 await db_session.commit()
454
455 meta = await get_index_meta(db_session, repo.repo_id)
456 assert meta is not None
457 assert meta["ref"] == commit.commit_id
458 assert meta["built_at"] is not None
459 assert meta["symbol_count"] >= 2
460
461 @pytest.mark.asyncio
462 async def test_load_intel_snapshot_none_when_no_index(
463 self, db_session: AsyncSession
464 ) -> None:
465 from musehub.services.musehub_symbol_indexer import load_intel_snapshot
466 repo = await create_repo(db_session, slug="intel-none")
467 assert await load_intel_snapshot(db_session, repo.repo_id) is None
468
469 @pytest.mark.asyncio
470 async def test_load_intel_snapshot_returns_snapshot_when_built(
471 self, db_session: AsyncSession
472 ) -> None:
473 from musehub.services.musehub_symbol_indexer import load_intel_snapshot
474
475 repo = await create_repo(db_session, slug="intel-ok")
476 commit = await _commit_with_delta(
477 db_session, repo.repo_id, "cI01",
478 ops=[_insert_op("app.py::Handler", "sha256:h1")],
479 )
480 results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
481 await db_session.commit()
482
483 assert results, "build_symbol_index returned empty results"
484 assert any(t == "code.intel_snapshot" for t, _ in results), "code.intel_snapshot not in results"
485 snap = await load_intel_snapshot(db_session, repo.repo_id)
486 assert snap is not None
487
488
489 class TestGetSnapshotManifestsBatch:
490 @pytest.mark.asyncio
491 async def test_empty_list_returns_empty_dict(
492 self, db_session: AsyncSession
493 ) -> None:
494 from musehub.services.musehub_snapshot import get_snapshot_manifests_batch
495 result = await get_snapshot_manifests_batch(db_session, [])
496 assert result == {}
497
498 @pytest.mark.asyncio
499 async def test_single_snapshot_manifest(
500 self, db_session: AsyncSession
501 ) -> None:
502 from musehub.services.musehub_snapshot import (
503 get_snapshot_manifests_batch,
504 upsert_snapshot_entries,
505 )
506 repo = await create_repo(db_session, slug="batch-single")
507 snap_id = "snap-batch-01"
508 await upsert_snapshot_entries(
509 db_session, repo.repo_id, snap_id, {"a.py": "sha256:a", "b.py": "sha256:b"}
510 )
511 await db_session.commit()
512
513 result = await get_snapshot_manifests_batch(db_session, [snap_id])
514 assert snap_id in result
515 assert result[snap_id]["a.py"] == "sha256:a"
516 assert result[snap_id]["b.py"] == "sha256:b"
517
518 @pytest.mark.asyncio
519 async def test_multiple_snapshots_grouped_correctly(
520 self, db_session: AsyncSession
521 ) -> None:
522 from musehub.services.musehub_snapshot import (
523 get_snapshot_manifests_batch,
524 upsert_snapshot_entries,
525 )
526 repo = await create_repo(db_session, slug="batch-multi")
527 for i in range(5):
528 snap_id = f"snap-multi-{i:02d}"
529 await upsert_snapshot_entries(
530 db_session, repo.repo_id, snap_id, {f"file{i}.py": f"sha256:{i}"}
531 )
532 await db_session.commit()
533
534 ids = [f"snap-multi-{i:02d}" for i in range(5)]
535 result = await get_snapshot_manifests_batch(db_session, ids)
536 assert len(result) == 5
537 for i, sid in enumerate(ids):
538 assert f"file{i}.py" in result[sid]
539
540 @pytest.mark.asyncio
541 async def test_unknown_snapshot_id_returns_empty_manifest(
542 self, db_session: AsyncSession
543 ) -> None:
544 from musehub.services.musehub_snapshot import get_snapshot_manifests_batch
545 result = await get_snapshot_manifests_batch(db_session, ["ghost-snap"])
546 assert result == {"ghost-snap": {}}
547
548
549 # ─────────────────────────────────────────────────────────────────────────────
550 # Layer 3 — E2E: full pipeline via direct service calls with real DB
551 # ─────────────────────────────────────────────────────────────────────────────
552
553 class TestSymbolIndexPipeline:
554 @pytest.mark.asyncio
555 async def test_build_then_meta_reflects_head_commit(
556 self, db_session: AsyncSession
557 ) -> None:
558 from musehub.services.musehub_symbol_indexer import get_index_meta
559
560 repo = await create_repo(db_session, slug="e2e-pipeline")
561 c1 = await _commit_with_delta(
562 db_session, repo.repo_id, "pipe-c001",
563 ops=[_insert_op("service.py::APIHandler", "sha256:h1")],
564 )
565 await _build_and_persist(db_session, repo.repo_id, c1.commit_id)
566 await db_session.commit()
567
568 meta = await get_index_meta(db_session, repo.repo_id)
569 assert meta is not None
570 assert meta["ref"] == c1.commit_id
571
572 @pytest.mark.asyncio
573 async def test_rebuild_updates_ref_to_latest_commit(
574 self, db_session: AsyncSession
575 ) -> None:
576 from musehub.services.musehub_symbol_indexer import get_index_meta
577
578 repo = await create_repo(db_session, slug="e2e-rebuild")
579 c1 = await _commit_with_delta(db_session, repo.repo_id, "rb-c001",
580 ops=[_insert_op("a.py::Old")])
581 await _build_and_persist(db_session, repo.repo_id, c1.commit_id)
582 await db_session.commit()
583
584 c2 = await _commit_with_delta(db_session, repo.repo_id, "rb-c002",
585 ops=[_insert_op("b.py::New")],
586 parent_ids=[c1.commit_id])
587 await _build_and_persist(db_session, repo.repo_id, c2.commit_id)
588 await db_session.commit()
589
590 meta = await get_index_meta(db_session, repo.repo_id)
591 assert meta is not None
592 assert meta["ref"] == c2.commit_id
593
594 @pytest.mark.asyncio
595 async def test_multi_commit_chain_all_symbols_indexed(
596 self, db_session: AsyncSession
597 ) -> None:
598 """3-commit chain — every symbol from every commit must appear in the index."""
599 from musehub.services.musehub_symbol_indexer import load_symbol_history
600
601 repo = await create_repo(db_session, slug="e2e-chain")
602 c1 = await _commit_with_delta(db_session, repo.repo_id, "chain-c001",
603 ops=[_insert_op("a.py::A1")])
604 c2 = await _commit_with_delta(db_session, repo.repo_id, "chain-c002",
605 ops=[_insert_op("b.py::B1")],
606 parent_ids=[c1.commit_id])
607 c3 = await _commit_with_delta(db_session, repo.repo_id, "chain-c003",
608 ops=[_insert_op("c.py::C1")],
609 parent_ids=[c2.commit_id])
610 await _build_and_persist(db_session, repo.repo_id, c3.commit_id)
611 await db_session.commit()
612
613 history = await load_symbol_history(db_session, repo.repo_id)
614 assert "a.py::A1" in history
615 assert "b.py::B1" in history
616 assert "c.py::C1" in history
617
618
619 # ─────────────────────────────────────────────────────────────────────────────
620 # Layer 4 — Data Integrity
621 # ─────────────────────────────────────────────────────────────────────────────
622
623 class TestDataIntegrity:
624 @pytest.mark.asyncio
625 async def test_upsert_atomic_replace_removes_stale_entries(
626 self, db_session: AsyncSession
627 ) -> None:
628 """Different snap_ids store different manifests independently."""
629 from musehub.services.musehub_snapshot import (
630 get_snapshot_manifest,
631 upsert_snapshot_entries,
632 )
633 repo = await create_repo(db_session, slug="di-atomic")
634 snap_id_a = "snap-atomic-a"
635 snap_id_b = "snap-atomic-b"
636 await upsert_snapshot_entries(
637 db_session, repo.repo_id, snap_id_a,
638 {"old_file.py": "sha256:old", "shared.py": "sha256:shared"},
639 )
640 await db_session.commit()
641
642 await upsert_snapshot_entries(
643 db_session, repo.repo_id, snap_id_b,
644 {"new_file.py": "sha256:new"},
645 )
646 await db_session.commit()
647
648 manifest_b = await get_snapshot_manifest(db_session, snap_id_b)
649 assert "new_file.py" in manifest_b
650 assert "old_file.py" not in manifest_b
651
652 manifest_a = await get_snapshot_manifest(db_session, snap_id_a)
653 assert "old_file.py" in manifest_a
654
655 @pytest.mark.asyncio
656 async def test_only_one_result_per_intel_type_after_multiple_builds(
657 self, db_session: AsyncSession
658 ) -> None:
659 from sqlalchemy import select, func
660
661 repo = await create_repo(db_session, slug="di-onerow")
662 for i in range(3):
663 c = await _commit_with_delta(
664 db_session, repo.repo_id, f"di-c{i:03d}",
665 ops=[_insert_op(f"f{i}.py::Sym")],
666 )
667 await _build_and_persist(db_session, repo.repo_id, c.commit_id)
668 await db_session.commit()
669
670 count = (await db_session.execute(
671 select(func.count()).select_from(db.MusehubIntelResult).where(
672 db.MusehubIntelResult.repo_id == repo.repo_id,
673 db.MusehubIntelResult.intel_type == "code.symbol_history",
674 )
675 )).scalar_one()
676 assert count == 1
677
678 @pytest.mark.asyncio
679 async def test_symbol_history_includes_commit_id_and_timestamp(
680 self, db_session: AsyncSession
681 ) -> None:
682 repo = await create_repo(db_session, slug="di-fields")
683 commit = await _commit_with_delta(
684 db_session, repo.repo_id, "di-field-001",
685 ops=[_insert_op("service.py::MyFn", "sha256:myfn")],
686 )
687 results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
688 await db_session.commit()
689
690 sh_data = _get_result_data(results, "code.symbol_history")
691 entries = sh_data.get("entries", {})
692 entry = entries["service.py::MyFn"][0]
693 assert entry["commit_id"] == commit.commit_id
694 assert entry["committed_at"] != ""
695 assert entry["op"] == "add"
696 assert entry["content_id"] == "sha256:myfn"
697
698
699 # ─────────────────────────────────────────────────────────────────────────────
700 # Layer 5 — Security
701 # ─────────────────────────────────────────────────────────────────────────────
702
703 class TestSecurity:
704 @pytest.mark.asyncio
705 async def test_corrupt_json_returns_empty_not_exception(
706 self, db_session: AsyncSession
707 ) -> None:
708 """A corrupt code.symbol_history data_json must return {} — not raise."""
709 from musehub.services.musehub_symbol_indexer import load_symbol_history
710 from musehub.core.genesis import compute_intel_result_id
711
712 repo = await create_repo(db_session, slug="sec-corrupt")
713 # Manually insert a row with garbage JSON
714 result_id = compute_intel_result_id(repo.repo_id, "code.symbol_history", "bad-ref")
715 from sqlalchemy.dialects.postgresql import insert as pg_insert
716 await db_session.execute(
717 pg_insert(db.MusehubIntelResult).values(
718 result_id=result_id,
719 repo_id=repo.repo_id,
720 intel_type="code.symbol_history",
721 domain="code",
722 ref="bad-ref",
723 data_json="not valid json {{{{",
724 schema_version=1,
725 computed_at=_now(),
726 ).on_conflict_do_nothing()
727 )
728 await db_session.commit()
729
730 result = await load_symbol_history(db_session, repo.repo_id)
731 assert result == {}
732
733 @pytest.mark.asyncio
734 async def test_build_with_unknown_head_commit_returns_empty(
735 self, db_session: AsyncSession
736 ) -> None:
737 """Unknown head_commit_id must return [], not raise."""
738 from musehub.services.musehub_symbol_indexer import build_symbol_index
739
740 repo = await create_repo(db_session, slug="sec-unknown-head")
741 results = await build_symbol_index(
742 db_session, repo.repo_id, "nonexistent-commit-id"
743 )
744 assert results == []
745
746 @pytest.mark.asyncio
747 async def test_corrupt_hash_occurrence_returns_empty(
748 self, db_session: AsyncSession
749 ) -> None:
750 from musehub.services.musehub_symbol_indexer import load_hash_occurrence
751 from musehub.core.genesis import compute_intel_result_id
752 from sqlalchemy.dialects.postgresql import insert as pg_insert
753
754 repo = await create_repo(db_session, slug="sec-corrupt-hash")
755 result_id = compute_intel_result_id(repo.repo_id, "code.hash_occurrence", "bad-ref")
756 await db_session.execute(
757 pg_insert(db.MusehubIntelResult).values(
758 result_id=result_id,
759 repo_id=repo.repo_id,
760 intel_type="code.hash_occurrence",
761 domain="code",
762 ref="bad-ref",
763 data_json="} invalid {",
764 schema_version=1,
765 computed_at=_now(),
766 ).on_conflict_do_nothing()
767 )
768 await db_session.commit()
769
770 result = await load_hash_occurrence(db_session, repo.repo_id)
771 assert result == {}
772
773
774 # ─────────────────────────────────────────────────────────────────────────────
775 # Layer 5B — Per-symbol intel
776 # ─────────────────────────────────────────────────────────────────────────────
777
778 class TestPerSymbolIntel:
779 @pytest.mark.asyncio
780 async def test_early_return_when_already_current(
781 self, db_session: AsyncSession
782 ) -> None:
783 """When the index is current and code.per_symbol_intel exists,
784 build_symbol_index must return [] (early exit, no recompute)."""
785 from musehub.services.musehub_symbol_indexer import build_symbol_index
786
787 repo = await create_repo(db_session, slug="bfil-current")
788 commit = await _commit_with_delta(
789 db_session, repo.repo_id, "bfil-c001",
790 ops=[_insert_op("svc.py::Handler", "sha256:h1")],
791 )
792 await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
793 await db_session.commit()
794
795 # Second call with same head: must early-return (empty list).
796 results2 = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
797 assert results2 == [], (
798 "build_symbol_index must return [] when index is current "
799 "and per_symbol_intel result exists."
800 )
801
802 @pytest.mark.asyncio
803 async def test_per_symbol_intel_populated_on_first_build(
804 self, db_session: AsyncSession
805 ) -> None:
806 repo = await create_repo(db_session, slug="bfil-fresh")
807 commit = await _commit_with_delta(
808 db_session, repo.repo_id, "bfil-fresh-c001",
809 ops=[_insert_op("api.py::Router", "sha256:r1")],
810 )
811 results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
812 await db_session.commit()
813
814 psi_data = _get_result_data(results, "code.per_symbol_intel")
815 assert "api.py::Router" in psi_data
816
817 @pytest.mark.asyncio
818 async def test_per_symbol_intel_contains_expected_fields(
819 self, db_session: AsyncSession
820 ) -> None:
821 repo = await create_repo(db_session, slug="bfil-fields")
822 commit = await _commit_with_delta(
823 db_session, repo.repo_id, "bfil-fields-c001",
824 ops=[_insert_op("lib.py::Parser", "sha256:p1")],
825 )
826 results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
827 await db_session.commit()
828
829 psi_data = _get_result_data(results, "code.per_symbol_intel")
830 entry = psi_data["lib.py::Parser"]
831 for field in ("churn", "churn_30d", "churn_90d", "blast", "blast_direct",
832 "blast_cross", "blast_top", "last_changed", "last_author",
833 "author_count", "gravity", "weekly"):
834 assert field in entry, f"Missing field '{field}' in per_symbol intel entry."
835
836 @pytest.mark.asyncio
837 async def test_author_count_reflects_unique_authors(
838 self, db_session: AsyncSession
839 ) -> None:
840 repo = await create_repo(db_session, slug="bfil-authors")
841 authors_seq = [("alice", "bfil-authors-c001"), ("bob", "bfil-authors-c002"), ("alice", "bfil-authors-c003")]
842 prev_id: list[str] = []
843 for i, (author, cid) in enumerate(authors_seq, start=1):
844 commit = await _commit_with_delta(
845 db_session, repo.repo_id, cid,
846 ops=[_insert_op("lib.py::Widget", "sha256:w" + str(i))],
847 parent_ids=prev_id,
848 author=author,
849 )
850 prev_id = [cid]
851 results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
852 await db_session.commit()
853
854 psi_data = _get_result_data(results, "code.per_symbol_intel")
855 entry = psi_data["lib.py::Widget"]
856 assert entry["author_count"] == 2, (
857 f"Expected 2 unique authors (alice, bob), got {entry['author_count']}"
858 )
859 assert entry["churn"] == 3
860
861 @pytest.mark.asyncio
862 async def test_lookup_symbol_intel_returns_matching_addresses(
863 self, db_session: AsyncSession
864 ) -> None:
865 from musehub.services.musehub_symbol_indexer import lookup_symbol_intel
866
867 repo = await create_repo(db_session, slug="bfil-lookup")
868 commit = await _commit_with_delta(
869 db_session, repo.repo_id, "bfil-lookup-c001",
870 ops=[
871 _insert_op("a.py::Foo", "sha256:f1"),
872 _insert_op("b.py::Bar", "sha256:b1"),
873 _insert_op("c.py::Baz", "sha256:z1"),
874 ],
875 )
876 await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
877 await db_session.commit()
878
879 result = await lookup_symbol_intel(db_session, repo.repo_id, ["a.py::Foo", "c.py::Baz"])
880 assert set(result.keys()) == {"a.py::Foo", "c.py::Baz"}
881 assert "b.py::Bar" not in result
882
883 @pytest.mark.asyncio
884 async def test_lookup_symbol_intel_returns_empty_when_no_index(
885 self, db_session: AsyncSession
886 ) -> None:
887 from musehub.services.musehub_symbol_indexer import lookup_symbol_intel
888
889 repo = await create_repo(db_session, slug="bfil-lookup-null")
890 result = await lookup_symbol_intel(db_session, repo.repo_id, ["core.py::Engine"])
891 assert result == {}
892
893
894 # ─────────────────────────────────────────────────────────────────────────────
895 # Layer 6 — Stress
896 # ─────────────────────────────────────────────────────────────────────────────
897
898 class TestStress:
899 @pytest.mark.asyncio
900 async def test_upsert_1000_file_manifest(self, db_session: AsyncSession) -> None:
901 from musehub.services.musehub_snapshot import (
902 get_snapshot_manifest,
903 upsert_snapshot_entries,
904 )
905 repo = await create_repo(db_session, slug="stress-1k-snap")
906 snap_id = "snap-1k"
907 manifest = {f"src/file_{i:04d}.py": f"sha256:{i:04d}" for i in range(1000)}
908
909 await upsert_snapshot_entries(db_session, repo.repo_id, snap_id, manifest)
910 await db_session.commit()
911
912 result = await get_snapshot_manifest(db_session, snap_id)
913 assert len(result) == 1000
914 assert result["src/file_0500.py"] == "sha256:0500"
915
916 @pytest.mark.asyncio
917 async def test_batch_manifest_50_snapshots(self, db_session: AsyncSession) -> None:
918 from musehub.services.musehub_snapshot import (
919 get_snapshot_manifests_batch,
920 upsert_snapshot_entries,
921 )
922 repo = await create_repo(db_session, slug="stress-batch-50")
923 ids: list[str] = []
924 for i in range(50):
925 sid = f"stress-snap-{i:02d}"
926 ids.append(sid)
927 await upsert_snapshot_entries(
928 db_session, repo.repo_id, sid,
929 {f"f{i}.py": f"sha256:{i}"},
930 )
931 await db_session.commit()
932
933 result = await get_snapshot_manifests_batch(db_session, ids)
934 assert len(result) == 50
935 for i, sid in enumerate(ids):
936 assert f"f{i}.py" in result[sid]
937
938 @pytest.mark.asyncio
939 async def test_build_symbol_index_100_commits(
940 self, db_session: AsyncSession
941 ) -> None:
942 """100-commit chain with 5 ops each — indexer must complete successfully."""
943 from musehub.services.musehub_symbol_indexer import load_symbol_history
944
945 repo = await create_repo(db_session, slug="stress-100-commits")
946 prev_id: str | None = None
947 head_id = "stress-head"
948 for i in range(100):
949 cid = f"stress-{i:04d}" if i < 99 else head_id
950 ops = [_insert_op(f"file{i}.py::Sym{j}", f"sha256:{i}{j}") for j in range(5)]
951 commit = await _commit_with_delta(
952 db_session, repo.repo_id, cid, ops=ops,
953 parent_ids=[prev_id] if prev_id else [],
954 )
955 prev_id = commit.commit_id
956
957 await _build_and_persist(db_session, repo.repo_id, head_id)
958 await db_session.commit()
959
960 history = await load_symbol_history(db_session, repo.repo_id)
961 # 100 files × 5 symbols each = 500 top-level symbol entries
962 assert len(history) == 500
963
964 @pytest.mark.asyncio
965 async def test_load_symbol_history_file_filter_on_large_index(
966 self, db_session: AsyncSession
967 ) -> None:
968 """Filter on large index returns only matching addresses."""
969 from musehub.services.musehub_symbol_indexer import load_symbol_history
970
971 repo = await create_repo(db_session, slug="stress-filter-large")
972 ops = []
973 for i in range(50):
974 for j in range(10):
975 ops.append(_insert_op(f"src/module_{i:02d}.py::Sym{j}", f"sha256:{i}{j}"))
976
977 commit = await _commit_with_delta(db_session, repo.repo_id, "stress-fl-head", ops=ops)
978 await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
979 await db_session.commit()
980
981 result = await load_symbol_history(db_session, repo.repo_id, file_path="src/module_05.py")
982 assert len(result) == 10
983 for key in result:
984 assert key.startswith("src/module_05.py")
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago