gabriel / musehub public
test_mpack_index_phase1.py python
246 lines 9.0 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 49 days ago
1 """TDD — Phase 1: mpack index written for every pushed object (issue #63).
2
3 PI-1 After process_mpack_index_job, every object in the mpack has a row
4 in musehub_mpack_index with the correct mpack_id (= mpack_key).
5 PI-2 The mpack_id stored matches the mpack_key exactly.
6 PI-3 A second push of different objects adds new rows; existing rows are
7 untouched (on_conflict_do_nothing).
8 PI-4 Objects from different repos are distinct in the global index
9 (content-addressed objects are globally unique by object_id).
10
11 NOTE: process_mpack_index_job was removed — indexing is now inline during
12 push (musehub_wire_push.py step 7d). These tests will be revisited once
13 the MVP wire protocol and pack index are fully wired in.
14 """
15 from __future__ import annotations
16
17 import datetime
18 import hashlib
19 from collections.abc import Mapping
20
21 import msgpack
22 import pytest
23
24 from sqlalchemy import select
25 from sqlalchemy.ext.asyncio import AsyncSession
26
27 from muse.core.types import blob_id
28 from musehub.db import musehub_repo_models as db
29 from musehub.core.genesis import compute_identity_id
30 from musehub.services.musehub_repository import create_repo
31
32
33 # ---------------------------------------------------------------------------
34 # Helpers
35 # ---------------------------------------------------------------------------
36
37 def _make_mpack(objects: Mapping[str, bytes]) -> tuple[bytes, str]:
38 """Build a minimal MPack and return (wire_bytes, mpack_key)."""
39 mpack = {
40 "commits": [],
41 "snapshots": [],
42 "objects": [
43 {"object_id": oid, "content": data}
44 for oid, data in objects.items()
45 ],
46 "branch_heads": {},
47 }
48 wire_bytes = msgpack.packb(mpack, use_bin_type=True)
49 mpack_key = "sha256:" + hashlib.sha256(wire_bytes).hexdigest()
50 return wire_bytes, mpack_key
51
52
53 async def _store_mpack(wire_bytes: bytes, mpack_key: str) -> None:
54 """Write mpack bytes to storage so process_mpack_index_job can fetch it."""
55 import musehub.storage.backends as _backends_mod
56 backend = _backends_mod.get_backend()
57 await backend.put_mpack(mpack_key, wire_bytes)
58
59
60 async def _enqueue_and_process(
61 session: AsyncSession,
62 repo_id: str,
63 mpack_key: str,
64 n_objects: int,
65 ) -> str:
66 """Enqueue a mpack.index job and run it synchronously. Returns job_id."""
67 from musehub.core.genesis import compute_job_id
68 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
69 from musehub.services.musehub_wire import process_mpack_index_job
70
71 now = datetime.datetime.now(datetime.timezone.utc)
72 job_id = compute_job_id(repo_id, "mpack.index", now.isoformat())
73 session.add(MusehubBackgroundJob(
74 job_id=job_id,
75 repo_id=repo_id,
76 job_type="mpack.index",
77 payload={
78 "mpack_key": mpack_key,
79 "branch": "main",
80 "head": "",
81 "pusher_id": "",
82 "declared_objects_count": n_objects,
83 "declared_commits_count": 0,
84 },
85 status="pending",
86 created_at=now,
87 attempt=0,
88 ))
89 await session.commit()
90 await process_mpack_index_job(session, job_id)
91 await session.commit()
92 return job_id
93
94
95 # ---------------------------------------------------------------------------
96 # PI-1
97 # ---------------------------------------------------------------------------
98
99 @pytest.mark.asyncio
100 async def test_pi1_mpack_index_written_for_every_object(db_session: AsyncSession) -> None:
101 """Every object in the mpack must have a row in musehub_mpack_index."""
102 repo = await create_repo(
103 db_session,
104 name="pi-test-1",
105 owner="gabriel",
106 owner_user_id=compute_identity_id(b"gabriel"),
107 visibility="public",
108 initialize=False,
109 )
110
111 objects = {blob_id(f"pi1-obj-{i}".encode()): f"pi1-obj-{i}".encode() for i in range(5)}
112 wire_bytes, mpack_key = _make_mpack(objects)
113 await _store_mpack(wire_bytes, mpack_key)
114 await _enqueue_and_process(db_session, repo.repo_id, mpack_key, len(objects))
115
116 rows_q = await db_session.execute(
117 select(db.MusehubMPackIndex).where(
118 db.MusehubMPackIndex.entity_id.in_(list(objects.keys()))
119 )
120 )
121 rows = rows_q.scalars().all()
122 indexed_oids = {r.entity_id for r in rows}
123
124 assert indexed_oids == set(objects.keys()), (
125 f"expected {len(objects)} mpack index rows, got {len(rows)}\n"
126 f"missing: {set(objects.keys()) - indexed_oids}"
127 )
128
129
130 # ---------------------------------------------------------------------------
131 # PI-2
132 # ---------------------------------------------------------------------------
133
134 @pytest.mark.asyncio
135 async def test_pi2_mpack_id_matches_mpack_key(db_session: AsyncSession) -> None:
136 """The mpack_id stored in musehub_mpack_index must equal the mpack_key."""
137 repo = await create_repo(
138 db_session,
139 name="pi-test-2",
140 owner="gabriel",
141 owner_user_id=compute_identity_id(b"gabriel"),
142 visibility="public",
143 initialize=False,
144 )
145
146 objects = {blob_id(b"pi2-only-obj"): b"pi2-only-obj"}
147 wire_bytes, mpack_key = _make_mpack(objects)
148 await _store_mpack(wire_bytes, mpack_key)
149 await _enqueue_and_process(db_session, repo.repo_id, mpack_key, len(objects))
150
151 row = await db_session.get(db.MusehubMPackIndex, (list(objects.keys())[0], mpack_key))
152 assert row is not None, "mpack index row not found"
153 assert row.mpack_id == mpack_key, f"mpack_id {row.mpack_id!r} != mpack_key {mpack_key!r}"
154
155
156 # ---------------------------------------------------------------------------
157 # PI-3
158 # ---------------------------------------------------------------------------
159
160 @pytest.mark.asyncio
161 async def test_pi3_second_push_adds_rows_without_overwriting(db_session: AsyncSession) -> None:
162 """A second push of different objects adds new rows; first push rows survive."""
163 repo = await create_repo(
164 db_session,
165 name="pi-test-3",
166 owner="gabriel",
167 owner_user_id=compute_identity_id(b"gabriel"),
168 visibility="public",
169 initialize=False,
170 )
171
172 objs_a = {blob_id(f"pi3-a-{i}".encode()): f"pi3-a-{i}".encode() for i in range(3)}
173 wire_a, key_a = _make_mpack(objs_a)
174 await _store_mpack(wire_a, key_a)
175 await _enqueue_and_process(db_session, repo.repo_id, key_a, len(objs_a))
176
177 objs_b = {blob_id(f"pi3-b-{i}".encode()): f"pi3-b-{i}".encode() for i in range(3)}
178 wire_b, key_b = _make_mpack(objs_b)
179 await _store_mpack(wire_b, key_b)
180 await _enqueue_and_process(db_session, repo.repo_id, key_b, len(objs_b))
181
182 all_oids = set(objs_a.keys()) | set(objs_b.keys())
183 rows_q = await db_session.execute(
184 select(db.MusehubMPackIndex).where(db.MusehubMPackIndex.entity_id.in_(list(all_oids)))
185 )
186 rows = rows_q.scalars().all()
187 indexed_oids = {r.entity_id for r in rows}
188 indexed_mpacks = {r.mpack_id for r in rows}
189
190 assert set(objs_a.keys()) <= indexed_oids, "first push objects missing from index"
191 assert set(objs_b.keys()) <= indexed_oids, "second push objects missing from index"
192 assert key_a in indexed_mpacks, "first mpack_key missing from index"
193 assert key_b in indexed_mpacks, "second mpack_key missing from index"
194
195
196 # ---------------------------------------------------------------------------
197 # PI-4
198 # ---------------------------------------------------------------------------
199
200 @pytest.mark.asyncio
201 async def test_pi4_objects_from_different_repos_are_distinct(db_session: AsyncSession) -> None:
202 """Objects pushed to different repos are distinct in the global index (no collisions)."""
203 repo_a = await create_repo(
204 db_session,
205 name="pi-test-4a",
206 owner="gabriel",
207 owner_user_id=compute_identity_id(b"gabriel"),
208 visibility="public",
209 initialize=False,
210 )
211 repo_b = await create_repo(
212 db_session,
213 name="pi-test-4b",
214 owner="gabriel",
215 owner_user_id=compute_identity_id(b"gabriel"),
216 visibility="public",
217 initialize=False,
218 )
219
220 objs_a = {blob_id(b"pi4-repo-a-obj"): b"pi4-repo-a-obj"}
221 wire_a, key_a = _make_mpack(objs_a)
222 await _store_mpack(wire_a, key_a)
223 await _enqueue_and_process(db_session, repo_a.repo_id, key_a, 1)
224
225 objs_b = {blob_id(b"pi4-repo-b-obj"): b"pi4-repo-b-obj"}
226 wire_b, key_b = _make_mpack(objs_b)
227 await _store_mpack(wire_b, key_b)
228 await _enqueue_and_process(db_session, repo_b.repo_id, key_b, 1)
229
230 rows_a = (await db_session.execute(
231 select(db.MusehubMPackIndex).where(
232 db.MusehubMPackIndex.entity_id.in_(list(objs_a.keys()))
233 )
234 )).scalars().all()
235
236 rows_b = (await db_session.execute(
237 select(db.MusehubMPackIndex).where(
238 db.MusehubMPackIndex.entity_id.in_(list(objs_b.keys()))
239 )
240 )).scalars().all()
241
242 assert {r.entity_id for r in rows_a} == set(objs_a.keys()), "repo_a objects not indexed"
243 assert {r.entity_id for r in rows_b} == set(objs_b.keys()), "repo_b objects not indexed"
244 assert {r.entity_id for r in rows_a}.isdisjoint({r.entity_id for r in rows_b}), (
245 "object collision — two repos pushed the same object_id (seeds must be unique)"
246 )
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 49 days ago