gabriel / musehub public
test_on_disk_refs.py python
623 lines 21.3 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 122 days ago
1 """Phase 3: On-disk refs as canonical branch pointers — TDD (RED → GREEN).
2
3 Seven tiers:
4 Tier 1 — write_ref writes refs/heads/<branch> via atomic rename
5 Tier 2 — read_ref round-trips what write_ref wrote
6 Tier 3 — read_ref returns None for missing branches
7 Tier 4 — write_ref creates parent dirs on first write
8 Tier 5 — write_ref is truly atomic (tmp file disappears, final file appears)
9 Tier 6 — wire_push_stream writes disk ref after DB commit
10 Tier 7 — GET /repos/{repo_id}/branches/{name}/repair heals DB from disk ref
11 """
12 from __future__ import annotations
13
14 import secrets
15 from collections.abc import AsyncGenerator
16 from pathlib import Path
17 from unittest.mock import patch
18
19 import pytest
20 import pytest_asyncio
21 from httpx import AsyncClient, ASGITransport
22 from sqlalchemy.ext.asyncio import AsyncSession
23
24 from musehub.auth.request_signing import MSignContext, optional_signed_request, require_signed_request
25 from muse.core.types import long_id, now_utc_iso
26 from musehub.core.genesis import compute_identity_id
27 from musehub.db.musehub_models import MusehubIdentity, MusehubRepo
28 from musehub.main import app
29 from musehub.types.json_types import JSONObject, StrDict
30
31
32 # ── helpers ───────────────────────────────────────────────────────────────────
33
34 def _oid() -> str:
35 """Return a valid sha256-prefixed object ID (128 hex chars)."""
36 return long_id(secrets.token_hex(32))
37
38
39 def _repo_root(tmp_path: Path, owner: str = "gabriel", slug: str = "test-repo") -> Path:
40 """Create a minimal server-side repo tree and return its root."""
41 root = tmp_path / owner / slug
42 (root / "refs" / "heads").mkdir(parents=True, exist_ok=True)
43 (root / "objects").mkdir(parents=True, exist_ok=True)
44 return root
45
46
47 # ── fixtures for Tier 6/7 integration tests ───────────────────────────────────
48
49 _OWNER = "ref-test-user"
50 _SLUG = "ref-test-repo"
51 _IDENTITY_ID = compute_identity_id(b"ref-test-user")
52
53 _TEST_CONTEXT = MSignContext(
54 handle=_OWNER,
55 identity_id=_IDENTITY_ID,
56 is_agent=False,
57 is_admin=False,
58 )
59
60
61 @pytest_asyncio.fixture
62 async def async_session(db_session: AsyncSession) -> AsyncSession:
63 """Alias: expose the conftest db_session as async_session."""
64 return db_session
65
66
67 @pytest_asyncio.fixture
68 async def async_client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
69 """Async HTTP client wired to the test app + DB."""
70 from tests.conftest import _Asgi24Wrapper
71 transport = ASGITransport(app=_Asgi24Wrapper(app))
72 async with AsyncClient(transport=transport, base_url="http://test") as ac:
73 yield ac
74
75
76 @pytest_asyncio.fixture
77 async def owner(db_session: AsyncSession) -> str:
78 """Create a test identity and return its handle."""
79 identity = MusehubIdentity(
80 identity_id=_IDENTITY_ID,
81 handle=_OWNER,
82 display_name="Ref Test User",
83 identity_type="human",
84 )
85 db_session.add(identity)
86 await db_session.commit()
87 return _OWNER
88
89
90 @pytest_asyncio.fixture
91 async def slug() -> str:
92 return _SLUG
93
94
95 @pytest_asyncio.fixture
96 async def repo_id(owner: str, db_session: AsyncSession) -> str:
97 """Create a minimal test repo row and return its repo_id."""
98 from datetime import datetime, timezone
99 from musehub.core.genesis import compute_repo_id, compute_branch_id
100 from musehub.db.musehub_models import MusehubBranch
101
102 created_at = datetime.now(tz=timezone.utc)
103 rid = compute_repo_id(_IDENTITY_ID, _SLUG, "code", created_at.isoformat())
104 repo = MusehubRepo(
105 repo_id=rid,
106 owner=owner,
107 slug=_SLUG,
108 name="Ref Test Repo",
109 owner_user_id=_IDENTITY_ID,
110 visibility="public",
111 default_branch="main",
112 created_at=created_at,
113 )
114 db_session.add(repo)
115 await db_session.commit()
116 branch = MusehubBranch(
117 branch_id=compute_branch_id(rid, "main"),
118 repo_id=rid,
119 name="main",
120 )
121 db_session.add(branch)
122 await db_session.commit()
123 return rid
124
125
126 @pytest.fixture
127 def authed_headers(owner: str) -> StrDict:
128 """Inject auth context and return minimal JSON headers."""
129 app.dependency_overrides[require_signed_request] = lambda: _TEST_CONTEXT
130 app.dependency_overrides[optional_signed_request] = lambda: _TEST_CONTEXT
131 yield {"Content-Type": "application/json"}
132 app.dependency_overrides.pop(require_signed_request, None)
133 app.dependency_overrides.pop(optional_signed_request, None)
134
135
136 # ── Tier 1: write_ref writes refs/heads/<branch> ─────────────────────────────
137
138 class TestWriteRefCreatesFile:
139 def test_write_ref_creates_ref_file(self, tmp_path: Path) -> None:
140 from musehub.storage.refs import write_ref
141 from muse.core.paths import server_ref_path
142
143 repo_root = _repo_root(tmp_path)
144 commit_id = _oid()
145
146 write_ref(repo_root, "main", commit_id)
147
148 ref_file = server_ref_path(repo_root, "main")
149 assert ref_file.exists(), "ref file must exist after write_ref"
150
151 def test_write_ref_content_is_commit_id_with_newline(self, tmp_path: Path) -> None:
152 from musehub.storage.refs import write_ref
153 from muse.core.paths import server_ref_path
154
155 repo_root = _repo_root(tmp_path)
156 commit_id = _oid()
157
158 write_ref(repo_root, "main", commit_id)
159
160 ref_file = server_ref_path(repo_root, "main")
161 assert ref_file.read_text() == f"{commit_id}\n"
162
163 def test_write_ref_overwrites_existing(self, tmp_path: Path) -> None:
164 from musehub.storage.refs import write_ref
165 from muse.core.paths import server_ref_path
166
167 repo_root = _repo_root(tmp_path)
168 old_id = _oid()
169 new_id = _oid()
170
171 write_ref(repo_root, "main", old_id)
172 write_ref(repo_root, "main", new_id)
173
174 ref_file = server_ref_path(repo_root, "main")
175 assert ref_file.read_text() == f"{new_id}\n"
176
177 def test_write_ref_supports_feature_branches(self, tmp_path: Path) -> None:
178 from musehub.storage.refs import write_ref
179 from muse.core.paths import server_ref_path
180
181 repo_root = _repo_root(tmp_path)
182 commit_id = _oid()
183
184 write_ref(repo_root, "feat/new-melody", commit_id)
185
186 ref_file = server_ref_path(repo_root, "feat/new-melody")
187 assert ref_file.exists()
188 assert ref_file.read_text() == f"{commit_id}\n"
189
190
191 # ── Tier 2: read_ref round-trips ─────────────────────────────────────────────
192
193 class TestReadRefRoundTrips:
194 def test_read_ref_returns_commit_id(self, tmp_path: Path) -> None:
195 from musehub.storage.refs import write_ref, read_ref
196
197 repo_root = _repo_root(tmp_path)
198 commit_id = _oid()
199
200 write_ref(repo_root, "main", commit_id)
201 result = read_ref(repo_root, "main")
202
203 assert result == commit_id
204
205 def test_read_ref_strips_trailing_newline(self, tmp_path: Path) -> None:
206 from musehub.storage.refs import read_ref
207 from muse.core.paths import server_ref_path
208
209 repo_root = _repo_root(tmp_path)
210 commit_id = _oid()
211 ref_file = server_ref_path(repo_root, "main")
212 ref_file.write_text(f"{commit_id}\n")
213
214 result = read_ref(repo_root, "main")
215 assert result == commit_id
216
217 def test_read_ref_round_trips_feature_branch(self, tmp_path: Path) -> None:
218 from musehub.storage.refs import write_ref, read_ref
219
220 repo_root = _repo_root(tmp_path)
221 commit_id = _oid()
222
223 write_ref(repo_root, "feat/jazz-changes", commit_id)
224 result = read_ref(repo_root, "feat/jazz-changes")
225
226 assert result == commit_id
227
228 def test_read_ref_multiple_branches_independent(self, tmp_path: Path) -> None:
229 from musehub.storage.refs import write_ref, read_ref
230
231 repo_root = _repo_root(tmp_path)
232 main_id = _oid()
233 dev_id = _oid()
234
235 write_ref(repo_root, "main", main_id)
236 write_ref(repo_root, "dev", dev_id)
237
238 assert read_ref(repo_root, "main") == main_id
239 assert read_ref(repo_root, "dev") == dev_id
240
241
242 # ── Tier 3: read_ref returns None for missing ─────────────────────────────────
243
244 class TestReadRefMissing:
245 def test_read_ref_returns_none_for_unknown_branch(self, tmp_path: Path) -> None:
246 from musehub.storage.refs import read_ref
247
248 repo_root = _repo_root(tmp_path)
249
250 assert read_ref(repo_root, "nonexistent") is None
251
252 def test_read_ref_returns_none_empty_repo(self, tmp_path: Path) -> None:
253 from musehub.storage.refs import read_ref
254
255 repo_root = _repo_root(tmp_path)
256
257 assert read_ref(repo_root, "main") is None
258
259 def test_read_ref_returns_none_after_branch_deleted_from_disk(
260 self, tmp_path: Path
261 ) -> None:
262 from musehub.storage.refs import write_ref, read_ref
263 from muse.core.paths import server_ref_path
264
265 repo_root = _repo_root(tmp_path)
266 commit_id = _oid()
267 write_ref(repo_root, "temp-branch", commit_id)
268
269 server_ref_path(repo_root, "temp-branch").unlink()
270
271 assert read_ref(repo_root, "temp-branch") is None
272
273
274 # ── Tier 4: write_ref creates parent dirs ────────────────────────────────────
275
276 class TestWriteRefCreatesParentDirs:
277 def test_write_ref_creates_refs_heads_dir(self, tmp_path: Path) -> None:
278 from musehub.storage.refs import write_ref
279
280 # Start with only the objects dir — no refs/heads/ yet
281 repo_root = tmp_path / "owner" / "repo"
282 (repo_root / "objects").mkdir(parents=True)
283
284 commit_id = _oid()
285 write_ref(repo_root, "main", commit_id)
286
287 assert (repo_root / "refs" / "heads" / "main").exists()
288
289 def test_write_ref_creates_nested_branch_dirs(self, tmp_path: Path) -> None:
290 from musehub.storage.refs import write_ref
291
292 repo_root = _repo_root(tmp_path)
293 commit_id = _oid()
294
295 write_ref(repo_root, "feat/new/nested", commit_id)
296
297 assert (repo_root / "refs" / "heads" / "feat" / "new" / "nested").exists()
298
299
300 # ── Tier 5: atomic write (tmp → rename) ──────────────────────────────────────
301
302 class TestWriteRefAtomic:
303 def test_no_tmp_file_after_write(self, tmp_path: Path) -> None:
304 from musehub.storage.refs import write_ref
305 from muse.core.paths import server_ref_path
306
307 repo_root = _repo_root(tmp_path)
308 commit_id = _oid()
309
310 write_ref(repo_root, "main", commit_id)
311
312 ref_file = server_ref_path(repo_root, "main")
313 tmp_file = ref_file.with_suffix(".tmp")
314 assert not tmp_file.exists(), ".tmp sentinel must be gone after rename"
315
316 def test_partial_write_does_not_corrupt_existing_ref(
317 self, tmp_path: Path
318 ) -> None:
319 """A crash mid-write (simulated by leaving a .tmp) must not corrupt the ref."""
320 from musehub.storage.refs import read_ref
321 from muse.core.paths import server_ref_path
322
323 repo_root = _repo_root(tmp_path)
324 good_id = _oid()
325
326 # Write a good ref first
327 ref_file = server_ref_path(repo_root, "main")
328 ref_file.write_text(f"{good_id}\n")
329
330 # Simulate a crash: write a .tmp but never rename
331 tmp_file = ref_file.with_suffix(".tmp")
332 tmp_file.write_text(long_id("ff" * 32) + "\n")
333
334 # The ref still reads the good value
335 assert read_ref(repo_root, "main") == good_id
336
337
338 # ── Tier 6: wire_push_stream writes disk ref ─────────────────────────────────
339
340 import msgpack as _msgpack
341 from muse.core.types import blob_id
342 from muse.core.mpack import MuseWireFrameWriter as _MuseWireFrameWriter
343 from musehub.models.wire import (
344 SFRAME_COMMIT_PACK,
345 SFRAME_END,
346 SFRAME_HEADER,
347 SFRAME_RESULT,
348 )
349
350 _fw = _MuseWireFrameWriter()
351
352
353 def _mwp_wrap(ft: str, data: JSONObject) -> bytes:
354 payload = _msgpack.packb(data, use_bin_type=True)
355 return _fw.wrap(frame_type=ft, payload=payload)
356
357
358 def _mwp_header(branch: str = "main", n_commits: int = 1) -> bytes:
359 snap_id = blob_id(b"default-snap")
360 return _mwp_wrap(SFRAME_HEADER, {
361 "t": SFRAME_HEADER, "branch": branch, "force": False,
362 "have": [], "head": snap_id, "n_objects": 0, "n_commits": n_commits,
363 })
364
365
366 def _mwp_commit_pack(commits: list[dict], snapshots: list[dict]) -> bytes:
367 return _mwp_wrap(SFRAME_COMMIT_PACK, {
368 "t": SFRAME_COMMIT_PACK, "commits": commits, "snapshots": snapshots,
369 })
370
371
372 def _mwp_end(n_commits: int = 1) -> bytes:
373 return _mwp_wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": 0, "n_commits": n_commits})
374
375
376 def _make_commit_wire(commit_id: str, snap_id: str, author: str = _OWNER) -> JSONObject:
377 from datetime import datetime, timezone
378 return {
379 "commit_id": commit_id,
380 "parent_ids": [],
381 "parent_commit_id": None,
382 "parent2_commit_id": None,
383 "snapshot_id": snap_id,
384 "branch": "main",
385 "message": "disk ref test commit",
386 "author": author,
387 "committed_at": now_utc_iso(),
388 "signature": "",
389 "signer_key_id": "",
390 "agent_id": "",
391 "model_id": "",
392 "metadata": {},
393 }
394
395
396 def _make_snap_wire(snap_id: str) -> JSONObject:
397 from datetime import datetime, timezone
398 return {
399 "snapshot_id": snap_id,
400 "manifest": {},
401 "committed_at": now_utc_iso(),
402 }
403
404
405 async def _run_push(db_session: AsyncSession, repo_id: str, owner: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> str:
406 """Run a minimal push via wire_push_stream and return the pushed commit_id."""
407 from unittest.mock import AsyncMock
408 from musehub.services.musehub_wire import wire_push_stream
409 from musehub.config import settings
410
411 # Stub the R2 backend
412 store: dict[str, bytes] = {}
413 backend = AsyncMock()
414 backend.exists = AsyncMock(side_effect=lambda oid, **kw: oid in store)
415 backend.put = AsyncMock(side_effect=lambda oid, data, **kw: store.update({oid: data}) or f"local://{oid}")
416 backend.get = AsyncMock(side_effect=lambda oid, **kw: store.get(oid))
417 monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend)
418
419 # Redirect repos_dir to tmp_path so disk refs land there
420 monkeypatch.setattr(settings, "musehub_repos_dir", str(tmp_path))
421
422 commit_id = blob_id(f"ref-test-commit-{secrets.token_hex(8)}".encode())
423 snap_id = blob_id(b"ref-test-snap")
424 commit = _make_commit_wire(commit_id, snap_id, owner)
425 snap = _make_snap_wire(snap_id)
426
427 body_frames = _mwp_header() + _mwp_commit_pack([commit], [snap]) + _mwp_end()
428
429 async def body_iter():
430 yield body_frames
431
432 results = []
433 unpacker = _msgpack.Unpacker(raw=False)
434 async for chunk in wire_push_stream(db_session, repo_id, body_iter(), owner):
435 unpacker.feed(chunk)
436 for frame in unpacker:
437 results.append(frame)
438
439 result_frames = [f for f in results if f.get("t") == SFRAME_RESULT]
440 assert result_frames and result_frames[0]["ok"] is True, f"push failed: {results}"
441 return commit_id
442
443
444 class TestWirePushWritesDiskRef:
445 """Integration: after a push the ref file must exist on disk."""
446
447 @pytest.mark.asyncio
448 async def test_push_writes_disk_ref(
449 self,
450 tmp_path: Path,
451 async_session: AsyncSession,
452 repo_id: str,
453 owner: str,
454 slug: str,
455 monkeypatch: pytest.MonkeyPatch,
456 ) -> None:
457 """After a successful push, refs/heads/main must exist on disk."""
458 from musehub.storage.refs import read_ref
459
460 commit_id = await _run_push(async_session, repo_id, owner, monkeypatch, tmp_path)
461
462 repo_root = tmp_path / owner / slug
463 result = read_ref(repo_root, "main")
464 assert result == commit_id
465
466 @pytest.mark.asyncio
467 async def test_push_disk_ref_matches_db_head(
468 self,
469 tmp_path: Path,
470 async_session: AsyncSession,
471 repo_id: str,
472 owner: str,
473 slug: str,
474 monkeypatch: pytest.MonkeyPatch,
475 ) -> None:
476 """The disk ref commit_id must equal the DB branch head commit_id."""
477 from musehub.storage.refs import read_ref
478 from musehub.services.musehub_repository import get_branch_head_commit_id
479
480 commit_id = await _run_push(async_session, repo_id, owner, monkeypatch, tmp_path)
481
482 repo_root = tmp_path / owner / slug
483 disk_head = read_ref(repo_root, "main")
484 db_head = await get_branch_head_commit_id(async_session, repo_id, "main")
485 assert disk_head == commit_id
486 assert db_head == commit_id
487
488
489 # ── Tier 7: repair endpoint ───────────────────────────────────────────────────
490
491 class TestRepairEndpoint:
492 """GET /repos/{repo_id}/branches/{name}/repair must heal DB from disk ref."""
493
494 @pytest.mark.asyncio
495 async def test_repair_heals_db_from_disk(
496 self,
497 tmp_path: Path,
498 async_client: AsyncClient,
499 async_session: AsyncSession,
500 authed_headers: StrDict,
501 repo_id: str,
502 owner: str,
503 slug: str,
504 ) -> None:
505 """When disk ref != DB head, repair updates DB to match disk."""
506 from musehub.storage.refs import write_ref
507 from musehub.services.musehub_repository import get_branch_head_commit_id
508 import musehub.db.musehub_models as models
509 from sqlalchemy import select, update
510
511 canonical_id = _oid()
512 stale_id = _oid()
513
514 with patch(
515 "musehub.storage.backends.settings.musehub_repos_dir",
516 str(tmp_path),
517 ):
518 repo_root = tmp_path / owner / slug
519 # Write the canonical commit to disk
520 write_ref(repo_root, "main", canonical_id)
521
522 # Manually set a stale value in DB
523 await async_session.execute(
524 update(models.MusehubBranch)
525 .where(
526 models.MusehubBranch.repo_id == repo_id,
527 models.MusehubBranch.name == "main",
528 )
529 .values(head_commit_id=stale_id)
530 )
531 await async_session.commit()
532
533 response = await async_client.post(
534 f"/api/repos/{repo_id}/branches/main/repair",
535 headers=authed_headers,
536 )
537 assert response.status_code == 200
538 body = response.json()
539 assert body["healed"] is True
540 assert body["commit_id"] == canonical_id
541
542 # DB must now match disk
543 db_head = await get_branch_head_commit_id(async_session, repo_id, "main")
544 assert db_head == canonical_id
545
546 @pytest.mark.asyncio
547 async def test_repair_noop_when_already_consistent(
548 self,
549 tmp_path: Path,
550 async_client: AsyncClient,
551 async_session: AsyncSession,
552 authed_headers: StrDict,
553 repo_id: str,
554 owner: str,
555 slug: str,
556 ) -> None:
557 """When disk and DB agree, repair returns healed=False."""
558 from musehub.storage.refs import write_ref
559 import musehub.db.musehub_models as models
560 from sqlalchemy import update
561
562 commit_id = _oid()
563
564 with patch(
565 "musehub.storage.backends.settings.musehub_repos_dir",
566 str(tmp_path),
567 ):
568 repo_root = tmp_path / owner / slug
569 write_ref(repo_root, "main", commit_id)
570
571 await async_session.execute(
572 update(models.MusehubBranch)
573 .where(
574 models.MusehubBranch.repo_id == repo_id,
575 models.MusehubBranch.name == "main",
576 )
577 .values(head_commit_id=commit_id)
578 )
579 await async_session.commit()
580
581 response = await async_client.post(
582 f"/api/repos/{repo_id}/branches/main/repair",
583 headers=authed_headers,
584 )
585 assert response.status_code == 200
586 body = response.json()
587 assert body["healed"] is False
588
589 @pytest.mark.asyncio
590 async def test_repair_404_when_no_disk_ref(
591 self,
592 tmp_path: Path,
593 async_client: AsyncClient,
594 authed_headers: StrDict,
595 repo_id: str,
596 ) -> None:
597 """Repair returns 404 when no disk ref exists to repair from."""
598 with patch(
599 "musehub.storage.backends.settings.musehub_repos_dir",
600 str(tmp_path),
601 ):
602 response = await async_client.post(
603 f"/api/repos/{repo_id}/branches/ghost/repair",
604 headers=authed_headers,
605 )
606 assert response.status_code == 404
607
608 @pytest.mark.asyncio
609 async def test_repair_requires_auth(
610 self,
611 tmp_path: Path,
612 async_client: AsyncClient,
613 repo_id: str,
614 ) -> None:
615 """Repair endpoint must reject unauthenticated requests."""
616 with patch(
617 "musehub.storage.backends.settings.musehub_repos_dir",
618 str(tmp_path),
619 ):
620 response = await async_client.post(
621 f"/api/repos/{repo_id}/branches/main/repair",
622 )
623 assert response.status_code in (401, 403)
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 122 days ago