gabriel / musehub public
test_symbol_intelligence.py python
1,170 lines 45.3 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """Section 6 — Symbol Intelligence (Intel): 7-layer test suite.
2
3 Covers:
4 - musehub/services/musehub_intel.py (compute_intel, _parse_ts, _health_label,
5 _health_color_class, IntelSnapshot, as_dict/from_dict)
6 - musehub/api/routes/musehub/blame.py (_build_real_symbol_blame, GET /repos/{repo_id}/blame/{ref})
7 - musehub/services/musehub_cross_repo.py (search_symbol_across_repos, cross_repo_impact,
8 workspace_blast_risk_top_n, build_deps_graph,
9 _module_prefix, _short_label)
10
11 Layers:
12 1. Unit — pure function tests, no DB, no I/O
13 2. Integration — real DB (PostgreSQL), service calls, no HTTP layer
14 3. End-to-End — full HTTP via AsyncClient, real DB
15 4. Stress — large data sets, volume correctness
16 5. Data Integrity — stored data correctness, field validation, round-trip
17 6. Security — auth guards, private repo access, injection safety
18 7. Performance — latency budgets for critical paths
19 """
20 from __future__ import annotations
21
22 import json
23 import secrets
24 import time
25 from datetime import datetime, timedelta, timezone
26
27 import msgpack
28
29 type SymbolHistory = dict[str, list[JSONObject]]
30 import pytest
31 import pytest_asyncio
32 from httpx import AsyncClient
33 from sqlalchemy.ext.asyncio import AsyncSession
34
35 from musehub.services.musehub_intel import (
36 IntelSnapshot,
37 BlastRiskEntry,
38 CouplingPair,
39 DeadEntry,
40 HotspotEntry,
41 VelocityWindow,
42 _health_color_class,
43 _health_label,
44 _parse_ts,
45 compute_intel,
46 )
47 from musehub.types.json_types import JSONObject, StrDict
48 from tests.factories import create_repo
49
50 # ---------------------------------------------------------------------------
51 # Local helpers
52 # ---------------------------------------------------------------------------
53
54 def _now() -> datetime:
55 return datetime.now(tz=timezone.utc)
56
57
58 def _ago(days: int = 0, **kwargs: int) -> datetime:
59 return _now() - timedelta(days=days, **kwargs)
60
61
62 def _ts(dt: datetime) -> str:
63 return dt.isoformat()
64
65
66 def _entry(commit_id: str, op: str = "add", ts: datetime | None = None,
67 content_id: str = "sha256:abc") -> JSONObject:
68 return {
69 "commit_id": commit_id,
70 "op": op,
71 "timestamp": _ts(ts or _now()),
72 "committed_at": _ts(ts or _now()),
73 "content_id": content_id,
74 }
75
76
77 def _history(**kwargs: list[JSONObject]) -> SymbolHistory:
78 """Build a symbol_history dict from keyword args: addr=entries."""
79 return dict(kwargs)
80
81
82 async def _build_index(session: AsyncSession, repo_id: str, head_id: str,
83 ops: list[JSONObject]) -> "types.SimpleNamespace":
84 """Insert one commit, build the symbol index, persist results, and return
85 a namespace with intel_full_json and intel_summary attributes."""
86 import types as _types
87 from musehub.db import musehub_models as db
88 from musehub.services.musehub_symbol_indexer import build_symbol_index
89 from musehub.services.musehub_intel_providers import persist_intel_results
90
91 commit = db.MusehubCommit(
92 commit_id=head_id,
93 repo_id=repo_id,
94 branch="main",
95 parent_ids=[],
96 message="test commit",
97 author="gabriel",
98 timestamp=_now(),
99 structured_delta={"ops": ops},
100 )
101 session.add(commit)
102 await session.flush()
103 results = await build_symbol_index(session, repo_id, head_id)
104 await persist_intel_results(session, repo_id, head_id, results)
105 await session.commit()
106 data_by_type = {t: json.dumps(d) for t, d in results}
107 return _types.SimpleNamespace(
108 intel_full_json=data_by_type.get("code.intel_snapshot"),
109 intel_summary=data_by_type.get("code.intel_summary"),
110 )
111
112
113 def _insert_op(address: str, content_id: str = "sha256:abc") -> JSONObject:
114 return {"address": address, "op": "insert", "content_id": content_id}
115
116
117 # ===========================================================================
118 # Layer 1 — Unit tests (pure functions, no DB, no I/O)
119 # ===========================================================================
120
121 class TestParseTs:
122 def test_iso_string_utc(self) -> None:
123 dt = _parse_ts("2025-01-15T10:30:00+00:00")
124 assert dt.year == 2025
125 assert dt.month == 1
126 assert dt.tzinfo is not None
127
128 def test_iso_string_z_suffix(self) -> None:
129 dt = _parse_ts("2025-06-01T00:00:00Z")
130 assert dt.tzinfo is not None
131 assert dt.year == 2025
132
133 def test_unix_int(self) -> None:
134 dt = _parse_ts(0)
135 assert dt.year == 1970
136 assert dt.tzinfo is not None
137
138 def test_unix_float(self) -> None:
139 dt = _parse_ts(1_700_000_000.5)
140 assert dt.year == 2023
141
142 def test_invalid_string_raises(self) -> None:
143 with pytest.raises(Exception):
144 _parse_ts("not-a-date")
145
146
147 class TestHealthLabel:
148 def test_excellent(self) -> None:
149 assert _health_label(100) == "Excellent"
150 assert _health_label(90) == "Excellent"
151
152 def test_good(self) -> None:
153 assert _health_label(89) == "Good"
154 assert _health_label(75) == "Good"
155
156 def test_fair(self) -> None:
157 assert _health_label(74) == "Fair"
158 assert _health_label(55) == "Fair"
159
160 def test_poor(self) -> None:
161 assert _health_label(54) == "Poor"
162 assert _health_label(35) == "Poor"
163
164 def test_critical(self) -> None:
165 assert _health_label(34) == "Critical"
166 assert _health_label(0) == "Critical"
167
168
169 class TestHealthColorClass:
170 def test_excellent(self) -> None:
171 assert _health_color_class(90) == "intel-health--excellent"
172
173 def test_good(self) -> None:
174 assert _health_color_class(75) == "intel-health--good"
175
176 def test_fair(self) -> None:
177 assert _health_color_class(55) == "intel-health--fair"
178
179 def test_poor(self) -> None:
180 assert _health_color_class(35) == "intel-health--poor"
181
182 def test_critical(self) -> None:
183 assert _health_color_class(0) == "intel-health--critical"
184
185
186 class TestComputeIntelUnit:
187 def test_empty_history_returns_zero_score(self) -> None:
188 snap = compute_intel({}, [], now_utc=_now())
189 assert snap.total_symbols == 0
190 assert snap.total_commits_indexed == 0
191 assert snap.health_score == 100 # no penalties = 100
192 assert snap.health_label == "Excellent"
193
194 def test_single_symbol_no_ts(self) -> None:
195 history = {"file.py::Foo": [{"commit_id": "c1", "op": "add"}]}
196 snap = compute_intel(history, [], now_utc=_now())
197 assert snap.total_symbols == 1
198 assert snap.total_commits_indexed == 1
199
200 def test_hotspot_detection(self) -> None:
201 # 12 changes on one symbol — exceeds _HOTSPOT_THRESHOLD (10)
202 entries = [_entry(f"c{i}") for i in range(12)]
203 history = {"file.py::HotFn": entries}
204 snap = compute_intel(history, [], now_utc=_now())
205 assert snap.alert_hotspot_count >= 1
206 assert any(h.address == "file.py::HotFn" for h in snap.hotspots)
207
208 def test_dead_code_detection(self) -> None:
209 # One old entry, last touched 100 days ago
210 old_ts = _ago(100)
211 history = {"file.py::Stale": [_entry("c1", ts=old_ts)]}
212 snap = compute_intel(history, [], now_utc=_now())
213 assert snap.alert_dead_count >= 1
214 assert any(d.address == "file.py::Stale" for d in snap.dead_candidates)
215
216 def test_recent_symbol_not_dead(self) -> None:
217 recent_ts = _ago(5)
218 history = {"file.py::Fresh": [_entry("c1", ts=recent_ts)]}
219 snap = compute_intel(history, [], now_utc=_now())
220 assert snap.alert_dead_count == 0
221
222 def test_blast_risk_co_change(self) -> None:
223 # Two symbols always change together → blast risk for both
224 entries_a = [_entry("c1"), _entry("c2")]
225 entries_b = [_entry("c1"), _entry("c2")]
226 history = {
227 "file.py::Alpha": entries_a,
228 "file.py::Beta": entries_b,
229 }
230 snap = compute_intel(history, [], now_utc=_now())
231 # Both are co-changed — blast risk entries should include at least one
232 assert len(snap.blast_risk) >= 1
233
234 def test_coupling_pairs_detected(self) -> None:
235 # Symbols sharing same commit → coupling pair
236 entries_a = [_entry("shared-commit")]
237 entries_b = [_entry("shared-commit")]
238 history = {
239 "file.py::A": entries_a,
240 "file.py::B": entries_b,
241 }
242 snap = compute_intel(history, [], now_utc=_now())
243 assert len(snap.coupling_pairs) >= 1
244 pair = snap.coupling_pairs[0]
245 assert pair.shared_commits >= 1
246
247 def test_breaking_changes_reduce_score(self) -> None:
248 snap_no_breaks = compute_intel({}, [], now_utc=_now())
249 snap_with_breaks = compute_intel({}, ["break1", "break2", "break3"], now_utc=_now())
250 assert snap_with_breaks.health_score < snap_no_breaks.health_score
251 assert snap_with_breaks.alert_breaking_count == 3
252
253 def test_velocity_buckets_populated(self) -> None:
254 recent = _ago(days=1)
255 history = {"file.py::Fn": [_entry("c1", ts=recent)]}
256 snap = compute_intel(history, [], now_utc=_now())
257 assert len(snap.velocity.weeks) == 12
258 assert snap.velocity.weeks[0] >= 1 # most recent week bucket
259
260 def test_health_score_capped_at_100(self) -> None:
261 snap = compute_intel({}, [], now_utc=_now())
262 assert 0 <= snap.health_score <= 100
263
264 def test_top_n_hotspots_limit(self) -> None:
265 # 20 symbols each changed 15 times → _TOP_N=10 returned
266 history: SymbolHistory = {}
267 for i in range(20):
268 history[f"file.py::Fn{i}"] = [_entry(f"c{i}_{j}") for j in range(15)]
269 snap = compute_intel(history, [], now_utc=_now())
270 assert len(snap.hotspots) <= 10
271
272 def test_dead_candidates_sorted_by_coldest_first(self) -> None:
273 h = {
274 "file.py::Old": [_entry("c1", ts=_ago(200))],
275 "file.py::Older": [_entry("c2", ts=_ago(300))],
276 }
277 snap = compute_intel(h, [], now_utc=_now())
278 if len(snap.dead_candidates) >= 2:
279 assert snap.dead_candidates[0].days_cold >= snap.dead_candidates[1].days_cold
280
281 def test_timestamp_invalid_gracefully_ignored(self) -> None:
282 history = {
283 "file.py::BadTs": [{"commit_id": "c1", "op": "add", "timestamp": "NOT_A_DATE"}]
284 }
285 snap = compute_intel(history, [], now_utc=_now())
286 # Should not raise; symbol counted but ts ignored
287 assert snap.total_symbols == 1
288
289
290 class TestIntelSnapshotSerialisation:
291 def _make_snap(self) -> IntelSnapshot:
292 return IntelSnapshot(
293 health_score=80,
294 health_label="Good",
295 alert_hotspot_count=2,
296 alert_dead_count=1,
297 alert_blast_risk_count=3,
298 alert_breaking_count=0,
299 hotspots=[HotspotEntry(address="a.py::Fn", change_count=15, last_changed=None)],
300 dead_candidates=[DeadEntry(address="b.py::Old", days_cold=120, blast_radius=0, added_at=None)],
301 blast_risk=[BlastRiskEntry(address="c.py::Risk", co_change_count=25, top_co_symbols=["d.py::X"])],
302 coupling_pairs=[CouplingPair(address_a="a.py::F", address_b="b.py::G", shared_commits=5)],
303 velocity=VelocityWindow(weeks=[1, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0]),
304 total_symbols=50,
305 total_commits_indexed=10,
306 )
307
308 def test_as_dict_round_trip(self) -> None:
309 snap = self._make_snap()
310 d = snap.as_dict()
311 reconstructed = IntelSnapshot.from_dict(d)
312 assert reconstructed.health_score == 80
313 assert reconstructed.health_label == "Good"
314 assert reconstructed.total_symbols == 50
315 assert reconstructed.hotspots[0].address == "a.py::Fn"
316 assert reconstructed.dead_candidates[0].days_cold == 120
317 assert reconstructed.blast_risk[0].co_change_count == 25
318 assert reconstructed.coupling_pairs[0].shared_commits == 5
319 assert reconstructed.velocity.weeks[0] == 1
320
321 def test_as_dict_json_serialisable(self) -> None:
322 snap = self._make_snap()
323 d = snap.as_dict()
324 # Must be JSON-serialisable (no datetimes, no custom objects)
325 json_str = json.dumps(d)
326 assert "health_score" in json_str
327
328 def test_from_dict_missing_optional_fields(self) -> None:
329 minimal = {
330 "health_score": 70,
331 "health_label": "Fair",
332 "alert_hotspot_count": 0,
333 "alert_dead_count": 0,
334 "alert_blast_risk_count": 0,
335 "alert_breaking_count": 0,
336 "total_symbols": 0,
337 "total_commits_indexed": 0,
338 }
339 snap = IntelSnapshot.from_dict(minimal)
340 assert snap.hotspots == []
341 assert snap.dead_candidates == []
342 assert snap.coupling_pairs == []
343 assert snap.velocity.weeks == []
344
345
346 class TestModulePrefix:
347 def test_three_segments(self) -> None:
348 from musehub.services.musehub_cross_repo import _module_prefix
349 result = _module_prefix("musehub.services.musehub_ci.enqueue_run")
350 assert result == "musehub.services.musehub_ci"
351
352 def test_fewer_than_depth(self) -> None:
353 from musehub.services.musehub_cross_repo import _module_prefix
354 result = _module_prefix("a.b")
355 assert result == "a.b" # shorter than depth=3, returns as-is
356
357 def test_exactly_depth(self) -> None:
358 from musehub.services.musehub_cross_repo import _module_prefix
359 result = _module_prefix("a.b.c")
360 assert result == "a.b.c"
361
362 def test_custom_depth(self) -> None:
363 from musehub.services.musehub_cross_repo import _module_prefix
364 result = _module_prefix("a.b.c.d.e", depth=2)
365 assert result == "a.b"
366
367
368 class TestShortLabel:
369 def test_two_segments(self) -> None:
370 from musehub.services.musehub_cross_repo import _short_label
371 assert _short_label("a.b.c") == "b.c"
372
373 def test_single_segment(self) -> None:
374 from musehub.services.musehub_cross_repo import _short_label
375 assert _short_label("single") == "single"
376
377
378 class TestBuildRealSymbolBlame:
379 def test_filters_to_path(self) -> None:
380 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
381
382 history = {
383 "musehub/api.py::Foo": [_entry("c1")],
384 "other/file.py::Bar": [_entry("c2")],
385 }
386 commit_map = {
387 "c1": {"message": "add Foo", "author": "gabriel", "timestamp": _now()},
388 }
389 results = _build_real_symbol_blame(history, "musehub/api.py", commit_map)
390 assert len(results) == 1
391 assert results[0].symbol_name == "Foo"
392
393 def test_excludes_import_declarations(self) -> None:
394 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
395
396 history = {
397 "file.py::import::os": [_entry("c1")],
398 "file.py::MyFn": [_entry("c1")],
399 }
400 commit_map = {"c1": {"message": "m", "author": "g", "timestamp": _now()}}
401 results = _build_real_symbol_blame(history, "file.py", commit_map)
402 names = [r.symbol_name for r in results]
403 assert "MyFn" in names
404 assert "import::os" not in names
405
406 def test_excludes_deleted_symbols(self) -> None:
407 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
408
409 history = {
410 "file.py::Gone": [_entry("c1", op="delete")],
411 "file.py::Here": [_entry("c2", op="add")],
412 }
413 commit_map = {
414 "c1": {"message": "del", "author": "g", "timestamp": _now()},
415 "c2": {"message": "add", "author": "g", "timestamp": _now()},
416 }
417 results = _build_real_symbol_blame(history, "file.py", commit_map)
418 names = [r.symbol_name for r in results]
419 assert "Gone" not in names
420 assert "Here" in names
421
422 def test_intel_signals_populated(self) -> None:
423 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
424
425 history = {
426 "file.py::HotFn": [_entry("c1")],
427 }
428 commit_map = {"c1": {"message": "m", "author": "g", "timestamp": _now()}}
429 intel = compute_intel(
430 {"file.py::HotFn": [_entry(f"c{i}") for i in range(15)]},
431 [],
432 now_utc=_now(),
433 )
434 results = _build_real_symbol_blame(history, "file.py", commit_map, intel=intel)
435 assert len(results) == 1
436 assert results[0].is_hotspot is True
437
438 def test_change_count_reflects_history_length(self) -> None:
439 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
440
441 history = {
442 "file.py::Changed": [_entry("c1"), _entry("c2"), _entry("c3")],
443 }
444 commit_map = {
445 "c1": {"message": "m", "author": "g", "timestamp": _now()},
446 "c2": {"message": "m", "author": "g", "timestamp": _now()},
447 "c3": {"message": "m", "author": "g", "timestamp": _now()},
448 }
449 results = _build_real_symbol_blame(history, "file.py", commit_map)
450 assert results[0].change_count == 3
451
452 def test_empty_history_returns_empty_list(self) -> None:
453 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
454
455 results = _build_real_symbol_blame({}, "file.py", {})
456 assert results == []
457
458 def test_unknown_commit_id_falls_back_gracefully(self) -> None:
459 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
460
461 history = {"file.py::Fn": [_entry("unknown-commit")]}
462 results = _build_real_symbol_blame(history, "file.py", {})
463 assert len(results) == 1
464 assert results[0].author == ""
465 assert results[0].commit_message == ""
466
467
468 # ===========================================================================
469 # Layer 2 — Integration tests (real DB, service layer, no HTTP)
470 # ===========================================================================
471
472 class TestComputeIntelIntegration:
473 @pytest.mark.asyncio
474 async def test_load_intel_snapshot_none_when_no_index(
475 self, db_session: AsyncSession
476 ) -> None:
477 from musehub.services.musehub_symbol_indexer import load_intel_snapshot
478
479 repo = await create_repo(db_session, slug="intel-no-index")
480 result = await load_intel_snapshot(db_session, repo.repo_id)
481 assert result is None
482
483 @pytest.mark.asyncio
484 async def test_build_index_populates_intel_full_json(
485 self, db_session: AsyncSession
486 ) -> None:
487 from musehub.services.musehub_symbol_indexer import load_intel_snapshot
488
489 repo = await create_repo(db_session, slug="intel-populated")
490 ops = [_insert_op("src/main.py::run"), _insert_op("src/main.py::setup")]
491 row = await _build_index(db_session, repo.repo_id, "head-intel-1", ops)
492 assert row is not None
493 assert row.intel_full_json is not None
494
495 snap = await load_intel_snapshot(db_session, repo.repo_id)
496 assert snap is not None
497 assert snap.total_symbols == 2
498
499 @pytest.mark.asyncio
500 async def test_intel_health_score_range(
501 self, db_session: AsyncSession
502 ) -> None:
503 from musehub.services.musehub_symbol_indexer import load_intel_snapshot
504
505 repo = await create_repo(db_session, slug="intel-health-range")
506 ops = [_insert_op(f"src/f.py::Fn{i}") for i in range(5)]
507 await _build_index(db_session, repo.repo_id, "head-hr", ops)
508
509 snap = await load_intel_snapshot(db_session, repo.repo_id)
510 assert snap is not None
511 assert 0 <= snap.health_score <= 100
512
513 @pytest.mark.asyncio
514 async def test_intel_summary_json_fields(
515 self, db_session: AsyncSession
516 ) -> None:
517 from musehub.db import musehub_models as db
518
519 repo = await create_repo(db_session, slug="intel-summary-fields")
520 ops = [_insert_op("api.py::endpoint")]
521 row = await _build_index(db_session, repo.repo_id, "head-summ", ops)
522 assert row is not None
523 assert row.intel_summary is not None
524 summary = json.loads(row.intel_summary)
525 assert "health_score" in summary
526 assert "symbol_count" in summary
527 assert "hotspot_count" in summary
528 assert "dead_symbol_count" in summary
529
530
531 class TestBlameIntegration:
532 @pytest.mark.asyncio
533 async def test_blame_returns_empty_when_no_index(
534 self, db_session: AsyncSession
535 ) -> None:
536 from musehub.services.musehub_symbol_indexer import load_symbol_history
537 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
538
539 repo = await create_repo(db_session, slug="blame-no-idx")
540 history = await load_symbol_history(db_session, repo.repo_id, file_path="file.py")
541 results = _build_real_symbol_blame(history, "file.py", {})
542 assert results == []
543
544 @pytest.mark.asyncio
545 async def test_blame_entries_after_index_build(
546 self, db_session: AsyncSession
547 ) -> None:
548 from musehub.services.musehub_symbol_indexer import load_symbol_history
549 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
550
551 repo = await create_repo(db_session, slug="blame-with-idx")
552 ops = [
553 _insert_op("src/api.py::handle_request"),
554 _insert_op("src/api.py::parse_args"),
555 ]
556 await _build_index(db_session, repo.repo_id, "head-blame", ops)
557
558 history = await load_symbol_history(
559 db_session, repo.repo_id, file_path="src/api.py"
560 )
561 results = _build_real_symbol_blame(history, "src/api.py", {})
562 names = [r.symbol_name for r in results]
563 assert "handle_request" in names
564 assert "parse_args" in names
565
566
567 class TestCrossRepoIntegration:
568 @pytest.mark.asyncio
569 async def test_search_symbol_no_repos(
570 self, db_session: AsyncSession
571 ) -> None:
572 from musehub.services.musehub_cross_repo import search_symbol_across_repos
573
574 result = await search_symbol_across_repos(
575 db_session, "ghost-owner", "Fn", visible_to_user="ghost-owner"
576 )
577 assert result == []
578
579 @pytest.mark.asyncio
580 async def test_search_symbol_finds_match(
581 self, db_session: AsyncSession
582 ) -> None:
583 from musehub.services.musehub_cross_repo import search_symbol_across_repos
584
585 owner = f"owner-{secrets.token_hex(4)}"
586 repo = await create_repo(db_session, slug="search-sym-repo", owner=owner,
587 visibility="public")
588 ops = [_insert_op("api.py::compute_intel")]
589 await _build_index(db_session, repo.repo_id, "head-search", ops)
590
591 results = await search_symbol_across_repos(
592 db_session, owner, "compute_intel", visible_to_user=owner
593 )
594 assert len(results) >= 1
595 assert any("compute_intel" in r.address for r in results)
596
597 @pytest.mark.asyncio
598 async def test_search_symbol_case_insensitive(
599 self, db_session: AsyncSession
600 ) -> None:
601 from musehub.services.musehub_cross_repo import search_symbol_across_repos
602
603 owner = f"owner-{secrets.token_hex(4)}"
604 repo = await create_repo(db_session, slug="search-case-repo", owner=owner,
605 visibility="public")
606 ops = [_insert_op("api.py::MyFunction")]
607 await _build_index(db_session, repo.repo_id, "head-case", ops)
608
609 results = await search_symbol_across_repos(
610 db_session, owner, "myfunction", visible_to_user=owner
611 )
612 assert any("MyFunction" in r.address for r in results)
613
614 @pytest.mark.asyncio
615 async def test_search_symbol_private_repo_excluded_without_auth(
616 self, db_session: AsyncSession
617 ) -> None:
618 from musehub.services.musehub_cross_repo import search_symbol_across_repos
619
620 owner = f"owner-{secrets.token_hex(4)}"
621 repo = await create_repo(db_session, slug="search-private-repo", owner=owner,
622 visibility="private")
623 ops = [_insert_op("api.py::SecretFn")]
624 await _build_index(db_session, repo.repo_id, "head-priv", ops)
625
626 # visible_to_user=None → only public repos
627 results = await search_symbol_across_repos(
628 db_session, owner, "SecretFn", visible_to_user=None
629 )
630 assert not any("SecretFn" in r.address for r in results)
631
632 @pytest.mark.asyncio
633 async def test_workspace_blast_risk_empty(
634 self, db_session: AsyncSession
635 ) -> None:
636 from musehub.services.musehub_cross_repo import workspace_blast_risk_top_n
637
638 result = await workspace_blast_risk_top_n(
639 db_session, "nonexistent-owner", visible_to_user="nonexistent-owner"
640 )
641 assert result == []
642
643 @pytest.mark.asyncio
644 async def test_workspace_blast_risk_populated(
645 self, db_session: AsyncSession
646 ) -> None:
647 from musehub.services.musehub_cross_repo import workspace_blast_risk_top_n
648
649 owner = f"owner-{secrets.token_hex(4)}"
650 repo = await create_repo(db_session, slug="wbr-repo", owner=owner,
651 visibility="public")
652 ops = [_insert_op("a.py::Fn"), _insert_op("b.py::Gn")]
653 await _build_index(db_session, repo.repo_id, "head-wbr", ops)
654
655 results = await workspace_blast_risk_top_n(
656 db_session, owner, visible_to_user=owner
657 )
658 assert len(results) >= 2
659 # Sorted by co_change_count descending
660 for i in range(len(results) - 1):
661 assert results[i].co_change_count >= results[i + 1].co_change_count
662
663 @pytest.mark.asyncio
664 async def test_cross_repo_impact_no_source_repo(
665 self, db_session: AsyncSession
666 ) -> None:
667 from musehub.services.musehub_cross_repo import cross_repo_impact
668
669 result = await cross_repo_impact(
670 db_session, "ghost-owner", secrets.token_hex(16), "file.py::Fn",
671 visible_to_user="ghost-owner",
672 )
673 assert result is None
674
675 @pytest.mark.asyncio
676 async def test_cross_repo_impact_unknown_address(
677 self, db_session: AsyncSession
678 ) -> None:
679 from musehub.services.musehub_cross_repo import cross_repo_impact
680
681 owner = f"owner-{secrets.token_hex(4)}"
682 repo = await create_repo(db_session, slug="cri-unknown", owner=owner,
683 visibility="public")
684 ops = [_insert_op("a.py::KnownFn")]
685 await _build_index(db_session, repo.repo_id, "head-cri", ops)
686
687 result = await cross_repo_impact(
688 db_session, owner, repo.repo_id, "a.py::NonExistent",
689 visible_to_user=owner,
690 )
691 assert result is None
692
693 @pytest.mark.asyncio
694 async def test_build_deps_graph_single_repo(
695 self, db_session: AsyncSession
696 ) -> None:
697 from musehub.services.musehub_cross_repo import build_deps_graph
698
699 owner = f"owner-{secrets.token_hex(4)}"
700 repo = await create_repo(db_session, slug="deps-single", owner=owner,
701 visibility="public")
702 ops = [
703 _insert_op("a.b.c.Fn"),
704 _insert_op("a.b.d.Gn"),
705 ]
706 await _build_index(db_session, repo.repo_id, "head-deps", ops)
707
708 graph = await build_deps_graph(
709 db_session, owner, repo.repo_id, visible_to_user=owner
710 )
711 assert hasattr(graph, "nodes")
712 assert hasattr(graph, "edges")
713
714 @pytest.mark.asyncio
715 async def test_build_deps_graph_no_source_repo_returns_empty(
716 self, db_session: AsyncSession
717 ) -> None:
718 from musehub.services.musehub_cross_repo import build_deps_graph, DepsGraph
719
720 owner = f"owner-{secrets.token_hex(4)}"
721 graph = await build_deps_graph(
722 db_session, owner, secrets.token_hex(16), visible_to_user=owner
723 )
724 assert isinstance(graph, DepsGraph)
725
726
727 # ===========================================================================
728 # Layer 3 — End-to-End tests (full HTTP via AsyncClient, real DB)
729 # ===========================================================================
730
731 class TestBlameEndToEnd:
732 @pytest.mark.asyncio
733 async def test_blame_404_unknown_repo(
734 self, client: AsyncClient, db_session: AsyncSession
735 ) -> None:
736 resp = await client.get(
737 f"/api/repos/{secrets.token_hex(16)}/blame/HEAD",
738 params={"path": "file.py"},
739 )
740 assert resp.status_code == 404
741
742 @pytest.mark.asyncio
743 async def test_blame_public_repo_no_auth(
744 self, client: AsyncClient, db_session: AsyncSession
745 ) -> None:
746 repo = await create_repo(db_session, slug="blame-e2e-pub", visibility="public")
747 await db_session.commit()
748 resp = await client.get(
749 f"/api/repos/{repo.repo_id}/blame/HEAD",
750 params={"path": "file.py"},
751 )
752 assert resp.status_code == 200
753 data = resp.json()
754 assert "entries" in data
755 assert "totalEntries" in data
756 assert "path" in data
757
758 @pytest.mark.asyncio
759 async def test_blame_private_repo_requires_auth(
760 self, client: AsyncClient, db_session: AsyncSession
761 ) -> None:
762 repo = await create_repo(db_session, slug="blame-e2e-priv", visibility="private")
763 await db_session.commit()
764 resp = await client.get(
765 f"/api/repos/{repo.repo_id}/blame/HEAD",
766 params={"path": "file.py"},
767 )
768 assert resp.status_code == 401
769
770 @pytest.mark.asyncio
771 async def test_blame_returns_entries_after_index_build(
772 self, client: AsyncClient, db_session: AsyncSession
773 ) -> None:
774 repo = await create_repo(db_session, slug="blame-e2e-entries", visibility="public")
775 ops = [_insert_op("api/routes.py::dispatch"), _insert_op("api/routes.py::validate")]
776 await _build_index(db_session, repo.repo_id, "head-blame-e2e", ops)
777
778 resp = await client.get(
779 f"/api/repos/{repo.repo_id}/blame/HEAD",
780 params={"path": "api/routes.py"},
781 )
782 assert resp.status_code == 200
783 data = resp.json()
784 names = [e["symbolName"] for e in data["entries"]]
785 assert "dispatch" in names
786 assert "validate" in names
787
788 @pytest.mark.asyncio
789 async def test_blame_path_filter_respected(
790 self, client: AsyncClient, db_session: AsyncSession
791 ) -> None:
792 repo = await create_repo(db_session, slug="blame-e2e-filter", visibility="public")
793 ops = [
794 _insert_op("path/a.py::FnA"),
795 _insert_op("path/b.py::FnB"),
796 ]
797 await _build_index(db_session, repo.repo_id, "head-filter", ops)
798
799 resp = await client.get(
800 f"/api/repos/{repo.repo_id}/blame/HEAD",
801 params={"path": "path/a.py"},
802 )
803 assert resp.status_code == 200
804 data = resp.json()
805 names = [e["symbolName"] for e in data["entries"]]
806 assert "FnA" in names
807 assert "FnB" not in names
808
809 @pytest.mark.asyncio
810 async def test_symbol_index_rebuild_endpoint(
811 self, client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict
812 ) -> None:
813 from musehub.db import musehub_models as dbm
814
815 repo = await create_repo(db_session, slug="rebuild-e2e")
816 # Create a head commit on "main"
817 commit = dbm.MusehubCommit(
818 commit_id="rebuild-head",
819 repo_id=repo.repo_id,
820 branch="main",
821 parent_ids=[],
822 message="initial",
823 author="gabriel",
824 timestamp=_now(),
825 structured_delta={"ops": [_insert_op("x.py::Fn")]},
826 )
827 db_branch = dbm.MusehubBranch(
828 branch_id=secrets.token_hex(16),
829 repo_id=repo.repo_id,
830 name="main",
831 head_commit_id="rebuild-head",
832 )
833 db_session.add(commit)
834 db_session.add(db_branch)
835 await db_session.commit()
836
837 resp = await client.post(
838 f"/api/repos/{repo.repo_id}/symbol-index/rebuild",
839 headers=auth_headers,
840 )
841 assert resp.status_code in (200, 202)
842
843 @pytest.mark.asyncio
844 async def test_symbol_index_rebuild_requires_auth(
845 self, client: AsyncClient, db_session: AsyncSession
846 ) -> None:
847 repo = await create_repo(db_session, slug="rebuild-noauth")
848 await db_session.commit()
849 resp = await client.post(f"/api/repos/{repo.repo_id}/symbol-index/rebuild")
850 assert resp.status_code == 401
851
852
853 # ===========================================================================
854 # Layer 4 — Stress tests
855 # ===========================================================================
856
857 class TestStress:
858 def test_compute_intel_1000_symbols(self) -> None:
859 """compute_intel on 1000 symbols completes without error."""
860 history: SymbolHistory = {}
861 for i in range(1000):
862 ts = _ago(days=i % 200)
863 history[f"module/file_{i % 20}.py::Fn{i}"] = [
864 _entry(f"c{i}", ts=ts)
865 ]
866 snap = compute_intel(history, [], now_utc=_now())
867 assert snap.total_symbols == 1000
868 assert 0 <= snap.health_score <= 100
869
870 def test_compute_intel_many_co_changing_symbols(self) -> None:
871 """50 symbols all sharing the same commit — coupling matrix stays bounded."""
872 commit_id = "shared"
873 history = {
874 f"file.py::Fn{i}": [_entry(commit_id)] for i in range(50)
875 }
876 snap = compute_intel(history, [], now_utc=_now())
877 # _TOP_COUPLING=5 cap must be respected
878 assert len(snap.coupling_pairs) <= 5
879
880 @pytest.mark.asyncio
881 async def test_search_symbol_across_10_repos(
882 self, db_session: AsyncSession
883 ) -> None:
884 """Search across 10 repos each with 20 symbols."""
885 from musehub.services.musehub_cross_repo import search_symbol_across_repos
886
887 owner = f"stress-owner-{secrets.token_hex(3)}"
888 for i in range(10):
889 repo = await create_repo(
890 db_session, slug=f"stress-repo-{i}", owner=owner, visibility="public"
891 )
892 ops = [_insert_op(f"mod{j}.py::TargetFn{j}") for j in range(20)]
893 await _build_index(db_session, repo.repo_id, f"head-stress-{i}", ops)
894
895 results = await search_symbol_across_repos(
896 db_session, owner, "TargetFn", visible_to_user=owner, limit=50
897 )
898 assert len(results) >= 1
899
900 @pytest.mark.asyncio
901 async def test_workspace_blast_risk_across_5_repos(
902 self, db_session: AsyncSession
903 ) -> None:
904 from musehub.services.musehub_cross_repo import workspace_blast_risk_top_n
905
906 owner = f"wbr-owner-{secrets.token_hex(3)}"
907 for i in range(5):
908 repo = await create_repo(
909 db_session, slug=f"wbr-sr-{i}", owner=owner, visibility="public"
910 )
911 ops = [_insert_op(f"f{j}.py::Fn{j}") for j in range(10)]
912 await _build_index(db_session, repo.repo_id, f"head-wbr-{i}", ops)
913
914 results = await workspace_blast_risk_top_n(
915 db_session, owner, top_n=20, visible_to_user=owner
916 )
917 # 5 repos × 10 symbols each = 50 entries, capped at top_n=20
918 assert len(results) <= 20
919 assert len(results) >= 1
920
921 def test_blame_build_500_symbols(self) -> None:
922 """_build_real_symbol_blame with 500 symbols in one file stays fast."""
923 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
924
925 history = {f"big/file.py::Fn{i}": [_entry(f"c{i}")] for i in range(500)}
926 commit_map = {f"c{i}": {"message": "m", "author": "g", "timestamp": _now()}
927 for i in range(500)}
928 results = _build_real_symbol_blame(history, "big/file.py", commit_map)
929 assert len(results) == 500
930
931
932 # ===========================================================================
933 # Layer 5 — Data Integrity tests
934 # ===========================================================================
935
936 class TestDataIntegrity:
937 def test_intel_snapshot_as_dict_from_dict_identity(self) -> None:
938 """Round-trip through as_dict/from_dict is lossless for all fields."""
939 snap = compute_intel(
940 {
941 "file.py::Fn": [_entry(f"c{i}") for i in range(15)],
942 "file.py::Old": [_entry("co", ts=_ago(150))],
943 },
944 ["breaking1"],
945 now_utc=_now(),
946 )
947 d = snap.as_dict()
948 reconstructed = IntelSnapshot.from_dict(d)
949 assert reconstructed.health_score == snap.health_score
950 assert reconstructed.alert_hotspot_count == snap.alert_hotspot_count
951 assert reconstructed.alert_dead_count == snap.alert_dead_count
952 assert reconstructed.alert_breaking_count == snap.alert_breaking_count
953 assert len(reconstructed.hotspots) == len(snap.hotspots)
954
955 @pytest.mark.asyncio
956 async def test_intel_full_json_stored_and_retrievable(
957 self, db_session: AsyncSession
958 ) -> None:
959 from musehub.services.musehub_symbol_indexer import load_intel_snapshot
960
961 repo = await create_repo(db_session, slug="di-intel-json")
962 ops = [_insert_op("svc.py::do_work", "sha256:beef")]
963 row = await _build_index(db_session, repo.repo_id, "head-di", ops)
964
965 assert row.intel_full_json is not None
966 snap = await load_intel_snapshot(db_session, repo.repo_id)
967 assert snap is not None
968 assert snap.total_symbols == 1
969 hotspot_addrs = [h.address for h in snap.hotspots]
970 # Address must be present in symbol set
971 all_in_dict = json.loads(row.intel_full_json)
972 assert all_in_dict["total_symbols"] == 1
973
974 def test_velocity_week_buckets_count(self) -> None:
975 """Velocity must always have exactly 12 buckets."""
976 history = {
977 "f.py::Fn": [_entry("c1", ts=_ago(days=1))],
978 }
979 snap = compute_intel(history, [], now_utc=_now())
980 assert len(snap.velocity.weeks) == 12
981
982 def test_hotspot_entries_have_required_fields(self) -> None:
983 history = {
984 "f.py::Fn": [_entry(f"c{i}") for i in range(12)],
985 }
986 snap = compute_intel(history, [], now_utc=_now())
987 for h in snap.hotspots:
988 assert isinstance(h.address, str)
989 assert isinstance(h.change_count, int)
990 assert h.change_count > 0
991
992 def test_dead_entry_days_cold_matches_expected(self) -> None:
993 old_ts = _ago(120)
994 history = {"f.py::Old": [_entry("c1", ts=old_ts)]}
995 snap = compute_intel(history, [], now_utc=_now())
996 if snap.dead_candidates:
997 entry = snap.dead_candidates[0]
998 assert 110 <= entry.days_cold <= 130 # allow ±10 days rounding
999
1000 @pytest.mark.asyncio
1001 async def test_blame_entry_fields_complete(
1002 self, db_session: AsyncSession
1003 ) -> None:
1004 from musehub.services.musehub_symbol_indexer import load_symbol_history
1005 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
1006
1007 repo = await create_repo(db_session, slug="di-blame-fields")
1008 ops = [_insert_op("f.py::Fn", "sha256:data1")]
1009 await _build_index(db_session, repo.repo_id, "head-di-blame", ops)
1010
1011 history = await load_symbol_history(db_session, repo.repo_id, file_path="f.py")
1012 commit_map = {"head-di-blame": {"message": "feat: add fn", "author": "gabriel",
1013 "timestamp": _now()}}
1014 results = _build_real_symbol_blame(history, "f.py", commit_map)
1015 assert len(results) == 1
1016 entry = results[0]
1017 assert entry.symbol_name == "Fn"
1018 assert entry.symbol_address == "f.py::Fn"
1019 assert entry.op in ("add", "modify", "delete", "insert", "replace", "patch", "rename")
1020
1021
1022 # ===========================================================================
1023 # Layer 6 — Security tests
1024 # ===========================================================================
1025
1026 class TestSecurity:
1027 @pytest.mark.asyncio
1028 async def test_blame_private_repo_401_no_token(
1029 self, client: AsyncClient, db_session: AsyncSession
1030 ) -> None:
1031 repo = await create_repo(db_session, slug="sec-blame-priv", visibility="private")
1032 await db_session.commit()
1033 resp = await client.get(
1034 f"/api/repos/{repo.repo_id}/blame/HEAD",
1035 params={"path": "file.py"},
1036 )
1037 assert resp.status_code == 401
1038
1039 @pytest.mark.asyncio
1040 async def test_blame_404_for_deleted_repo(
1041 self, client: AsyncClient, db_session: AsyncSession
1042 ) -> None:
1043 repo = await create_repo(db_session, slug="sec-blame-deleted", visibility="public")
1044 await db_session.delete(repo)
1045 await db_session.commit()
1046
1047 resp = await client.get(
1048 f"/api/repos/{repo.repo_id}/blame/HEAD",
1049 params={"path": "file.py"},
1050 )
1051 assert resp.status_code == 404
1052
1053 @pytest.mark.asyncio
1054 async def test_search_private_repo_not_visible_to_other_user(
1055 self, db_session: AsyncSession
1056 ) -> None:
1057 from musehub.services.musehub_cross_repo import search_symbol_across_repos
1058
1059 owner = f"sec-owner-{secrets.token_hex(3)}"
1060 repo = await create_repo(db_session, slug="sec-priv-search", owner=owner,
1061 visibility="private")
1062 ops = [_insert_op("secret.py::TopSecretFn")]
1063 await _build_index(db_session, repo.repo_id, "head-sec-priv", ops)
1064
1065 # Different user can't see private repo
1066 results = await search_symbol_across_repos(
1067 db_session, owner, "TopSecretFn", visible_to_user="other-user"
1068 )
1069 assert not any("TopSecretFn" in r.address for r in results)
1070
1071 @pytest.mark.asyncio
1072 async def test_blame_path_with_traversal_chars_no_crash(
1073 self, client: AsyncClient, db_session: AsyncSession
1074 ) -> None:
1075 repo = await create_repo(db_session, slug="sec-traversal", visibility="public")
1076 await db_session.commit()
1077 # Path with traversal attempt — server should return 200 with empty entries
1078 resp = await client.get(
1079 f"/api/repos/{repo.repo_id}/blame/HEAD",
1080 params={"path": "../../../etc/passwd"},
1081 )
1082 assert resp.status_code == 200
1083 data = resp.json()
1084 assert data["entries"] == []
1085
1086 def test_compute_intel_with_injected_commit_ids(self) -> None:
1087 """Malformed commit IDs in history do not cause exceptions."""
1088 history = {
1089 "f.py::Fn": [
1090 {"commit_id": "'; DROP TABLE commits; --", "op": "add"},
1091 {"commit_id": "", "op": "modify"},
1092 {"commit_id": None, "op": "add"},
1093 ]
1094 }
1095 snap = compute_intel(history, [], now_utc=_now())
1096 assert snap.total_symbols == 1
1097
1098 def test_blame_build_with_xss_in_commit_message(self) -> None:
1099 """XSS in commit messages is returned verbatim, not executed."""
1100 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
1101
1102 history = {"f.py::Fn": [_entry("c1")]}
1103 xss_msg = "<script>alert('xss')</script>"
1104 commit_map = {"c1": {"message": xss_msg, "author": "<img onerror=alert()>",
1105 "timestamp": _now()}}
1106 results = _build_real_symbol_blame(history, "f.py", commit_map)
1107 assert results[0].commit_message == xss_msg # stored as-is (escaping is UI's job)
1108
1109
1110 # ===========================================================================
1111 # Layer 7 — Performance tests
1112 # ===========================================================================
1113
1114 class TestPerformance:
1115 def test_compute_intel_500_symbols_under_200ms(self) -> None:
1116 history = {
1117 f"pkg/mod_{i}.py::Symbol{i}": [
1118 _entry(f"c{i}_{j}", ts=_ago(j % 300))
1119 for j in range(5)
1120 ]
1121 for i in range(100)
1122 }
1123 t0 = time.perf_counter()
1124 snap = compute_intel(history, [], now_utc=_now())
1125 elapsed_ms = (time.perf_counter() - t0) * 1000
1126 assert elapsed_ms < 200, f"compute_intel took {elapsed_ms:.1f}ms"
1127 assert snap.total_symbols == 100
1128
1129 def test_intel_as_dict_from_dict_1000_entries_under_50ms(self) -> None:
1130 history = {f"f.py::Fn{i}": [_entry(f"c{i}")] for i in range(1000)}
1131 snap = compute_intel(history, [], now_utc=_now())
1132 t0 = time.perf_counter()
1133 d = snap.as_dict()
1134 IntelSnapshot.from_dict(d)
1135 elapsed_ms = (time.perf_counter() - t0) * 1000
1136 assert elapsed_ms < 50, f"as_dict/from_dict took {elapsed_ms:.1f}ms"
1137
1138 def test_blame_build_1000_symbols_under_200ms(self) -> None:
1139 from musehub.api.routes.musehub.blame import _build_real_symbol_blame
1140
1141 history = {f"big/file.py::Fn{i}": [_entry(f"c{i}")] for i in range(1000)}
1142 commit_map = {f"c{i}": {"message": "m", "author": "g", "timestamp": _now()}
1143 for i in range(1000)}
1144 t0 = time.perf_counter()
1145 results = _build_real_symbol_blame(history, "big/file.py", commit_map)
1146 elapsed_ms = (time.perf_counter() - t0) * 1000
1147 assert elapsed_ms < 200, f"_build_real_symbol_blame took {elapsed_ms:.1f}ms"
1148 assert len(results) == 1000
1149
1150 @pytest.mark.asyncio
1151 async def test_search_across_5_repos_under_1s(
1152 self, db_session: AsyncSession
1153 ) -> None:
1154 from musehub.services.musehub_cross_repo import search_symbol_across_repos
1155
1156 owner = f"perf-owner-{secrets.token_hex(3)}"
1157 for i in range(5):
1158 repo = await create_repo(
1159 db_session, slug=f"perf-repo-{i}", owner=owner, visibility="public"
1160 )
1161 ops = [_insert_op(f"m{j}.py::Fn{j}") for j in range(30)]
1162 await _build_index(db_session, repo.repo_id, f"head-perf-{i}", ops)
1163
1164 t0 = time.perf_counter()
1165 results = await search_symbol_across_repos(
1166 db_session, owner, "Fn", visible_to_user=owner
1167 )
1168 elapsed_ms = (time.perf_counter() - t0) * 1000
1169 assert elapsed_ms < 1000, f"search_symbol_across_repos took {elapsed_ms:.1f}ms"
1170 assert len(results) >= 1
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago