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