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