gabriel / musehub public
test_gc_background_tasks.py python
546 lines 24.0 KB
Raw
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ breaking 143 days ago
1 """Section 41 — GC & Background Tasks: 7-layer test suite.
2
3 Covers:
4 - musehub/services/musehub_gc.py::GCResult, run_gc
5
6 Background jobs (symbol indexing, GC) are now dispatched via enqueue_job
7 and executed by the job runner — there are no standalone _build_symbol_index_async
8 or _run_gc_async coroutines in wire.py.
9 """
10 from __future__ import annotations
11
12 import asyncio
13 import secrets
14 import time
15 import uuid
16 from datetime import UTC, datetime
17
18 import msgpack
19 import pytest
20 from sqlalchemy import select
21 from sqlalchemy.ext.asyncio import AsyncSession
22
23 from musehub.core.genesis import compute_identity_id, compute_repo_id
24 from musehub.db.musehub_models import (
25 MusehubBranch,
26 MusehubCommit,
27 MusehubRepo,
28 MusehubSnapshot,
29 )
30 from musehub.services.musehub_gc import GCResult, run_gc
31
32
33 # ─────────────────────────────────────────────────────────────────────────────
34 # Helpers
35 # ─────────────────────────────────────────────────────────────────────────────
36
37 def _uid() -> str:
38 return str(uuid.uuid4())
39
40
41 def _hex(n: int = 32) -> str:
42 return secrets.token_hex(n)
43
44
45 async def _mk_repo(session: AsyncSession, suffix: str = "") -> MusehubRepo:
46 slug = f"gc-test{suffix}-{_uid()[:8]}"
47 created_at = datetime.now(tz=UTC)
48 owner_id = compute_identity_id(b"testuser")
49 repo = MusehubRepo(
50 repo_id=compute_repo_id(owner_id, slug, "code", created_at.isoformat()),
51 name=f"gc-test{suffix}",
52 owner="testuser",
53 slug=slug,
54 owner_user_id=owner_id,
55 created_at=created_at,
56 updated_at=created_at,
57 )
58 session.add(repo)
59 await session.flush()
60 return repo
61
62
63 async def _mk_branch(
64 session: AsyncSession,
65 repo_id: str,
66 head_commit_id: str | None = None,
67 name: str = "main",
68 ) -> MusehubBranch:
69 branch = MusehubBranch(
70 branch_id=_uid(),
71 repo_id=repo_id,
72 name=name,
73 head_commit_id=head_commit_id,
74 )
75 session.add(branch)
76 await session.flush()
77 return branch
78
79
80 async def _mk_commit(
81 session: AsyncSession,
82 repo_id: str,
83 commit_id: str | None = None,
84 parent_ids: list[str] | None = None,
85 snapshot_id: str | None = None,
86 ) -> MusehubCommit:
87 commit = MusehubCommit(
88 commit_id=commit_id or _hex(),
89 repo_id=repo_id,
90 branch="main",
91 parent_ids=parent_ids or [],
92 message="test commit",
93 author="testuser",
94 timestamp=datetime.now(UTC),
95 snapshot_id=snapshot_id,
96 )
97 session.add(commit)
98 await session.flush()
99 return commit
100
101
102 async def _mk_snapshot(
103 session: AsyncSession, repo_id: str, snapshot_id: str | None = None
104 ) -> MusehubSnapshot:
105 snap = MusehubSnapshot(
106 snapshot_id=snapshot_id or _hex(),
107 repo_id=repo_id,
108 manifest_blob=msgpack.packb({}, use_bin_type=True),
109 entry_count=0,
110 )
111 session.add(snap)
112 await session.flush()
113 return snap
114
115
116 async def _count_commits(session: AsyncSession, repo_id: str) -> int:
117 result = await session.execute(
118 select(MusehubCommit).where(MusehubCommit.repo_id == repo_id)
119 )
120 return len(result.scalars().all())
121
122
123 # ─────────────────────────────────────────────────────────────────────────────
124 # LAYER 1 — UNIT
125 # ─────────────────────────────────────────────────────────────────────────────
126
127
128 class TestGCResultUnit:
129 """Unit: GCResult dataclass shape and defaults."""
130
131 def test_default_counts_are_zero(self) -> None:
132 r = GCResult(repo_id="abc")
133 assert r.commits_deleted == 0
134 assert r.snapshots_deleted == 0
135 assert r.reachable_commit_count == 0
136
137 def test_default_errors_is_empty_list(self) -> None:
138 r = GCResult(repo_id="abc")
139 assert r.errors == []
140
141 def test_errors_is_independent_per_instance(self) -> None:
142 a = GCResult(repo_id="a")
143 b = GCResult(repo_id="b")
144 a.errors.append("x")
145 assert b.errors == []
146
147 def test_fields_set_correctly(self) -> None:
148 r = GCResult(repo_id="x", commits_deleted=3, snapshots_deleted=1, reachable_commit_count=5)
149 assert r.repo_id == "x"
150 assert r.commits_deleted == 3
151 assert r.snapshots_deleted == 1
152 assert r.reachable_commit_count == 5
153
154
155 class TestRunGcNoBranches:
156 """Unit: run_gc early-exit when repo has no branches."""
157
158 async def test_no_branches_returns_zero_deletions(self, db_session: AsyncSession) -> None:
159 repo = await _mk_repo(db_session)
160 result = await run_gc(db_session, repo.repo_id)
161 assert result.commits_deleted == 0
162 assert result.reachable_commit_count == 0
163
164 async def test_branch_with_null_head_is_ignored(self, db_session: AsyncSession) -> None:
165 repo = await _mk_repo(db_session)
166 await _mk_branch(db_session, repo.repo_id, head_commit_id=None)
167 result = await run_gc(db_session, repo.repo_id)
168 assert result.commits_deleted == 0
169
170 async def test_unknown_repo_id_returns_empty_result(self, db_session: AsyncSession) -> None:
171 result = await run_gc(db_session, "nonexistent-repo-id")
172 assert result.commits_deleted == 0
173 assert result.reachable_commit_count == 0
174
175
176
177
178 # ─────────────────────────────────────────────────────────────────────────────
179 # LAYER 2 — INTEGRATION
180 # ─────────────────────────────────────────────────────────────────────────────
181
182
183 class TestRunGcIntegration:
184 """Integration: run_gc with real in-memory DB."""
185
186 async def test_clean_repo_deletes_nothing(self, db_session: AsyncSession) -> None:
187 repo = await _mk_repo(db_session)
188 commit = await _mk_commit(db_session, repo.repo_id)
189 await _mk_branch(db_session, repo.repo_id, head_commit_id=commit.commit_id)
190 result = await run_gc(db_session, repo.repo_id)
191 assert result.commits_deleted == 0
192 assert result.reachable_commit_count == 1
193
194 async def test_orphaned_commit_is_deleted(self, db_session: AsyncSession) -> None:
195 repo = await _mk_repo(db_session)
196 # reachable commit
197 head = await _mk_commit(db_session, repo.repo_id)
198 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
199 # orphaned commit — not on any branch
200 await _mk_commit(db_session, repo.repo_id)
201 await db_session.commit()
202
203 result = await run_gc(db_session, repo.repo_id)
204 assert result.commits_deleted == 1
205 assert result.reachable_commit_count == 1
206
207 async def test_orphaned_chain_all_deleted(self, db_session: AsyncSession) -> None:
208 repo = await _mk_repo(db_session)
209 head = await _mk_commit(db_session, repo.repo_id)
210 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
211 # chain of 3 orphaned commits
212 c1 = await _mk_commit(db_session, repo.repo_id)
213 c2 = await _mk_commit(db_session, repo.repo_id, parent_ids=[c1.commit_id])
214 c3 = await _mk_commit(db_session, repo.repo_id, parent_ids=[c2.commit_id])
215 await db_session.commit()
216
217 result = await run_gc(db_session, repo.repo_id)
218 assert result.commits_deleted == 3
219 assert result.reachable_commit_count == 1
220
221 async def test_reachable_commit_chain_untouched(self, db_session: AsyncSession) -> None:
222 repo = await _mk_repo(db_session)
223 c1 = await _mk_commit(db_session, repo.repo_id)
224 c2 = await _mk_commit(db_session, repo.repo_id, parent_ids=[c1.commit_id])
225 c3 = await _mk_commit(db_session, repo.repo_id, parent_ids=[c2.commit_id])
226 await _mk_branch(db_session, repo.repo_id, head_commit_id=c3.commit_id)
227 await db_session.commit()
228
229 result = await run_gc(db_session, repo.repo_id)
230 assert result.commits_deleted == 0
231 assert result.reachable_commit_count == 3
232
233 async def test_gc_scoped_to_repo(self, db_session: AsyncSession) -> None:
234 """Orphaned commits in repo_a are not touched when GC runs on repo_b."""
235 repo_a = await _mk_repo(db_session, "-a")
236 repo_b = await _mk_repo(db_session, "-b")
237 # repo_b: clean
238 head_b = await _mk_commit(db_session, repo_b.repo_id)
239 await _mk_branch(db_session, repo_b.repo_id, head_commit_id=head_b.commit_id)
240 # repo_a: orphaned commit
241 await _mk_commit(db_session, repo_a.repo_id)
242 await db_session.commit()
243
244 result = await run_gc(db_session, repo_b.repo_id)
245 assert result.commits_deleted == 0
246 # repo_a's orphan still exists
247 assert await _count_commits(db_session, repo_a.repo_id) == 1
248
249 async def test_orphaned_snapshot_deleted_with_commit(self, db_session: AsyncSession) -> None:
250 repo = await _mk_repo(db_session)
251 snap = await _mk_snapshot(db_session, repo.repo_id)
252 head = await _mk_commit(db_session, repo.repo_id)
253 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
254 # orphaned commit references a snapshot
255 await _mk_commit(db_session, repo.repo_id, snapshot_id=snap.snapshot_id)
256 await db_session.commit()
257
258 result = await run_gc(db_session, repo.repo_id)
259 assert result.commits_deleted == 1
260 assert result.snapshots_deleted == 1
261
262
263 # ─────────────────────────────────────────────────────────────────────────────
264 # LAYER 3 — E2E
265 # ─────────────────────────────────────────────────────────────────────────────
266
267
268 class TestBackgroundTaskE2E:
269 """E2E: run_gc runs to completion against the test DB."""
270
271 async def test_run_gc_completes_on_real_db(self, db_session: AsyncSession) -> None:
272 repo = await _mk_repo(db_session)
273 head = await _mk_commit(db_session, repo.repo_id)
274 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
275 await db_session.commit()
276 result = await run_gc(db_session, repo.repo_id)
277 assert result.commits_deleted == 0
278
279 async def test_run_gc_with_orphan_completes(self, db_session: AsyncSession) -> None:
280 repo = await _mk_repo(db_session)
281 head = await _mk_commit(db_session, repo.repo_id)
282 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
283 await _mk_commit(db_session, repo.repo_id) # orphan
284 await db_session.commit()
285 result = await run_gc(db_session, repo.repo_id)
286 assert result.commits_deleted == 1
287
288
289 # ─────────────────────────────────────────────────────────────────────────────
290 # LAYER 4 — STRESS
291 # ─────────────────────────────────────────────────────────────────────────────
292
293
294 class TestGCStress:
295 """Stress: GC under large commit volumes."""
296
297 async def test_gc_100_commit_linear_chain_all_reachable(self, db_session: AsyncSession) -> None:
298 repo = await _mk_repo(db_session)
299 parent_id: str | None = None
300 commits = []
301 for _ in range(100):
302 c = await _mk_commit(db_session, repo.repo_id, parent_ids=[parent_id] if parent_id else [])
303 commits.append(c)
304 parent_id = c.commit_id
305 await _mk_branch(db_session, repo.repo_id, head_commit_id=commits[-1].commit_id)
306 await db_session.commit()
307
308 result = await run_gc(db_session, repo.repo_id)
309 assert result.commits_deleted == 0
310 assert result.reachable_commit_count == 100
311
312 async def test_gc_50_orphaned_commits_all_deleted(self, db_session: AsyncSession) -> None:
313 repo = await _mk_repo(db_session)
314 head = await _mk_commit(db_session, repo.repo_id)
315 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
316 for _ in range(50):
317 await _mk_commit(db_session, repo.repo_id)
318 await db_session.commit()
319
320 result = await run_gc(db_session, repo.repo_id)
321 assert result.commits_deleted == 50
322
323 async def test_gc_diamond_merge_topology(self, db_session: AsyncSession) -> None:
324 """Diamond: base → left + right → merge; all 4 commits are reachable."""
325 repo = await _mk_repo(db_session)
326 base = await _mk_commit(db_session, repo.repo_id)
327 left = await _mk_commit(db_session, repo.repo_id, parent_ids=[base.commit_id])
328 right = await _mk_commit(db_session, repo.repo_id, parent_ids=[base.commit_id])
329 merge = await _mk_commit(
330 db_session, repo.repo_id, parent_ids=[left.commit_id, right.commit_id]
331 )
332 await _mk_branch(db_session, repo.repo_id, head_commit_id=merge.commit_id)
333 await db_session.commit()
334
335 result = await run_gc(db_session, repo.repo_id)
336 assert result.commits_deleted == 0
337 assert result.reachable_commit_count == 4
338
339 async def test_sequential_gc_calls_idempotent(self, db_session: AsyncSession) -> None:
340 """Two sequential GC calls on the same repo both complete without error."""
341 repo = await _mk_repo(db_session)
342 head = await _mk_commit(db_session, repo.repo_id)
343 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
344 await db_session.commit()
345
346 first = await run_gc(db_session, repo.repo_id)
347 second = await run_gc(db_session, repo.repo_id)
348 assert first.commits_deleted == 0
349 assert second.commits_deleted == 0
350
351
352 # ─────────────────────────────────────────────────────────────────────────────
353 # LAYER 5 — DATA INTEGRITY
354 # ─────────────────────────────────────────────────────────────────────────────
355
356
357 class TestGCDataIntegrity:
358 """Data Integrity: GC preserves reachable objects and correctly tracks counts."""
359
360 async def test_reachable_commits_still_in_db_after_gc(self, db_session: AsyncSession) -> None:
361 repo = await _mk_repo(db_session)
362 c1 = await _mk_commit(db_session, repo.repo_id)
363 c2 = await _mk_commit(db_session, repo.repo_id, parent_ids=[c1.commit_id])
364 await _mk_branch(db_session, repo.repo_id, head_commit_id=c2.commit_id)
365 await _mk_commit(db_session, repo.repo_id) # orphan
366 await db_session.commit()
367
368 await run_gc(db_session, repo.repo_id)
369 remaining = await _count_commits(db_session, repo.repo_id)
370 assert remaining == 2
371
372 async def test_orphaned_commits_gone_after_gc(self, db_session: AsyncSession) -> None:
373 repo = await _mk_repo(db_session)
374 head = await _mk_commit(db_session, repo.repo_id)
375 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
376 orphan = await _mk_commit(db_session, repo.repo_id)
377 orphan_id = orphan.commit_id
378 await db_session.commit()
379
380 await run_gc(db_session, repo.repo_id)
381 row = await db_session.get(MusehubCommit, orphan_id)
382 assert row is None
383
384 async def test_reachable_count_plus_deleted_equals_total(self, db_session: AsyncSession) -> None:
385 repo = await _mk_repo(db_session)
386 head = await _mk_commit(db_session, repo.repo_id)
387 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
388 for _ in range(5):
389 await _mk_commit(db_session, repo.repo_id)
390 await db_session.commit()
391
392 result = await run_gc(db_session, repo.repo_id)
393 assert result.reachable_commit_count + result.commits_deleted == 6
394
395 async def test_shared_snapshot_not_deleted_when_reachable(self, db_session: AsyncSession) -> None:
396 """Snapshot referenced by both a reachable and orphaned commit must not be deleted."""
397 repo = await _mk_repo(db_session)
398 snap = await _mk_snapshot(db_session, repo.repo_id)
399 head = await _mk_commit(db_session, repo.repo_id, snapshot_id=snap.snapshot_id)
400 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
401 await _mk_commit(db_session, repo.repo_id, snapshot_id=snap.snapshot_id) # orphan
402 await db_session.commit()
403
404 result = await run_gc(db_session, repo.repo_id)
405 assert result.commits_deleted == 1
406 assert result.snapshots_deleted == 0
407 # snapshot still present
408 snap_row = await db_session.get(MusehubSnapshot, snap.snapshot_id)
409 assert snap_row is not None
410
411 async def test_gc_idempotent_second_run_deletes_nothing(self, db_session: AsyncSession) -> None:
412 repo = await _mk_repo(db_session)
413 head = await _mk_commit(db_session, repo.repo_id)
414 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
415 await _mk_commit(db_session, repo.repo_id) # orphan
416 await db_session.commit()
417
418 first = await run_gc(db_session, repo.repo_id)
419 assert first.commits_deleted == 1
420
421 second = await run_gc(db_session, repo.repo_id)
422 assert second.commits_deleted == 0
423
424 async def test_multi_branch_all_heads_reachable(self, db_session: AsyncSession) -> None:
425 """Commits pointed to by multiple branches are all preserved."""
426 repo = await _mk_repo(db_session)
427 c_main = await _mk_commit(db_session, repo.repo_id)
428 c_dev = await _mk_commit(db_session, repo.repo_id)
429 c_feat = await _mk_commit(db_session, repo.repo_id)
430 await _mk_branch(db_session, repo.repo_id, head_commit_id=c_main.commit_id, name="main")
431 await _mk_branch(db_session, repo.repo_id, head_commit_id=c_dev.commit_id, name="dev")
432 await _mk_branch(db_session, repo.repo_id, head_commit_id=c_feat.commit_id, name="feat")
433 await db_session.commit()
434
435 result = await run_gc(db_session, repo.repo_id)
436 assert result.commits_deleted == 0
437 assert result.reachable_commit_count == 3
438
439
440 # ─────────────────────────────────────────────────────────────────────────────
441 # LAYER 6 — SECURITY
442 # ─────────────────────────────────────────────────────────────────────────────
443
444
445 class TestGCBackgroundSecurity:
446 """Security: isolation, error suppression, no cross-repo contamination."""
447
448 async def test_gc_cannot_delete_commits_in_other_repo(self, db_session: AsyncSession) -> None:
449 repo_a = await _mk_repo(db_session, "-sec-a")
450 repo_b = await _mk_repo(db_session, "-sec-b")
451 # repo_b: clean
452 head_b = await _mk_commit(db_session, repo_b.repo_id)
453 await _mk_branch(db_session, repo_b.repo_id, head_commit_id=head_b.commit_id)
454 # repo_a: orphaned commit
455 orphan = await _mk_commit(db_session, repo_a.repo_id)
456 orphan_id = orphan.commit_id
457 await db_session.commit()
458
459 await run_gc(db_session, repo_b.repo_id)
460
461 # repo_a's commit untouched
462 still_there = await db_session.get(MusehubCommit, orphan_id)
463 assert still_there is not None
464
465 async def test_gc_invalid_repo_id_no_exception(self, db_session: AsyncSession) -> None:
466 result = await run_gc(db_session, "totally-invalid-uuid")
467 assert result.commits_deleted == 0
468
469 async def test_gc_large_repo_id_no_exception(self, db_session: AsyncSession) -> None:
470 result = await run_gc(db_session, "x" * 1000)
471 assert result.commits_deleted == 0
472
473 async def test_gc_does_not_expose_internal_state_via_result(self, db_session: AsyncSession) -> None:
474 """GCResult contains no raw SQL or stack traces."""
475 repo = await _mk_repo(db_session)
476 await db_session.commit()
477 result = await run_gc(db_session, repo.repo_id)
478 result_str = str(result)
479 assert "SELECT" not in result_str
480 assert "Traceback" not in result_str
481
482
483 # ─────────────────────────────────────────────────────────────────────────────
484 # LAYER 7 — PERFORMANCE
485 # ─────────────────────────────────────────────────────────────────────────────
486
487
488 class TestGCPerformance:
489 """Performance: GC latency budgets."""
490
491 async def test_gc_clean_repo_under_100ms(self, db_session: AsyncSession) -> None:
492 repo = await _mk_repo(db_session)
493 head = await _mk_commit(db_session, repo.repo_id)
494 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
495 await db_session.commit()
496
497 t0 = time.perf_counter()
498 await run_gc(db_session, repo.repo_id)
499 elapsed = time.perf_counter() - t0
500 assert elapsed < 0.1, f"GC on clean repo took {elapsed:.3f}s"
501
502 async def test_gc_100_commit_chain_under_2s(self, db_session: AsyncSession) -> None:
503 repo = await _mk_repo(db_session)
504 parent_id: str | None = None
505 commits = []
506 for _ in range(100):
507 c = await _mk_commit(
508 db_session, repo.repo_id, parent_ids=[parent_id] if parent_id else []
509 )
510 commits.append(c)
511 parent_id = c.commit_id
512 await _mk_branch(db_session, repo.repo_id, head_commit_id=commits[-1].commit_id)
513 await db_session.commit()
514
515 t0 = time.perf_counter()
516 await run_gc(db_session, repo.repo_id)
517 elapsed = time.perf_counter() - t0
518 assert elapsed < 2.0, f"GC on 100-commit chain took {elapsed:.3f}s"
519
520 async def test_gc_50_orphans_under_1s(self, db_session: AsyncSession) -> None:
521 repo = await _mk_repo(db_session)
522 head = await _mk_commit(db_session, repo.repo_id)
523 await _mk_branch(db_session, repo.repo_id, head_commit_id=head.commit_id)
524 for _ in range(50):
525 await _mk_commit(db_session, repo.repo_id)
526 await db_session.commit()
527
528 t0 = time.perf_counter()
529 await run_gc(db_session, repo.repo_id)
530 elapsed = time.perf_counter() - t0
531 assert elapsed < 1.0, f"GC on 50 orphans took {elapsed:.3f}s"
532
533 async def test_gc_result_construction_is_negligible(self) -> None:
534 t0 = time.perf_counter()
535 for _ in range(10_000):
536 GCResult(repo_id="x")
537 elapsed = time.perf_counter() - t0
538 assert elapsed < 0.1, f"10K GCResult() took {elapsed:.3f}s"
539
540 async def test_gc_result_construction_fast(self) -> None:
541 """GCResult construction overhead is negligible."""
542 t0 = time.perf_counter()
543 for _ in range(10_000):
544 GCResult(repo_id="x", reachable_commit_count=5)
545 elapsed = time.perf_counter() - t0
546 assert elapsed < 0.1, f"10K GCResult() took {elapsed:.3f}s"
File History 1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ 143 days ago