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