gabriel / musehub public
test_wire_shallow_fetch.py python
387 lines 14.5 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago
1 """TDD — shallow fetch: depth-limited BFS in wire_fetch_stream.
2
3 Test plan
4 ---------
5 S1 WireFetchRequest accepts an optional depth field (schema contract).
6 S2 depth=None (default) returns full commit history — N commits in, N out.
7 S3 depth=1 returns only the tip commit, not its ancestors.
8 S4 depth=2 returns tip + one ancestor, stops before the root.
9 S5 H frame carries shallow_commits list when depth is active.
10 S6 Objects in the response are only those referenced by depth-limited commits.
11 S7 depth=1 on a single-commit repo behaves identically to depth=None.
12 S8 depth cap works correctly on a branching (multi-parent) DAG.
13 """
14 from __future__ import annotations
15 from collections.abc import AsyncIterator
16
17 import asyncio
18 import zlib
19 from datetime import datetime, timezone
20
21 import msgpack
22 import pytest
23 from sqlalchemy.ext.asyncio import AsyncSession
24
25 from muse.core.types import blob_id, fake_id
26 from musehub.db import musehub_models as db
27 from musehub.models.wire import (
28 SFRAME_COMMIT_PACK,
29 SFRAME_END,
30 SFRAME_HEADER,
31 SFRAME_OBJECT,
32 WireFetchRequest,
33 )
34 from musehub.types.json_types import JSONObject
35 from tests.factories import create_repo
36
37
38 # ---------------------------------------------------------------------------
39 # Helpers
40 # ---------------------------------------------------------------------------
41
42 def _uid(seed: str) -> str:
43 return fake_id(seed)
44
45
46 def _now() -> datetime:
47 return datetime.now(tz=timezone.utc)
48
49
50 async def _make_snapshot(
51 session: AsyncSession,
52 repo_id: str,
53 manifest: dict[str, str],
54 ) -> db.MusehubSnapshot:
55 sid = _uid(str(manifest))
56 snap = db.MusehubSnapshot(
57 snapshot_id=sid,
58 repo_id=repo_id,
59 directories=[],
60 manifest_blob=msgpack.packb(manifest, use_bin_type=True),
61 entry_count=len(manifest),
62 created_at=_now(),
63 )
64 session.add(snap)
65 await session.commit()
66 return snap
67
68
69 async def _make_commit(
70 session: AsyncSession,
71 repo_id: str,
72 *,
73 parent_ids: list[str] | None = None,
74 snapshot_id: str | None = None,
75 seed: str = "",
76 ) -> db.MusehubCommit:
77 commit_id = _uid(f"commit-{seed}-{id(session)}")
78 row = db.MusehubCommit(
79 commit_id=commit_id,
80 repo_id=repo_id,
81 branch="main",
82 parent_ids=parent_ids or [],
83 message=f"commit {seed}",
84 author="test",
85 timestamp=_now(),
86 snapshot_id=snapshot_id,
87 )
88 session.add(row)
89 await session.commit()
90 return row
91
92
93 async def _make_object(
94 session: AsyncSession,
95 repo_id: str,
96 content: bytes,
97 ) -> str:
98 """Store an object in musehub_objects + musehub_object_refs. Returns object_id."""
99 from musehub.services.musehub_wire import get_backend
100 backend = get_backend()
101 oid = blob_id(content)
102 uri = await backend.put(oid, content)
103 row = db.MusehubObject(
104 object_id=oid,
105 path="test.dat",
106 size_bytes=len(content),
107 disk_path=uri.replace("local://", ""),
108 storage_uri=uri,
109 )
110 from sqlalchemy.dialects.postgresql import insert as pg_insert
111 await session.execute(
112 pg_insert(db.MusehubObject)
113 .values(
114 object_id=oid,
115 path="test.dat",
116 size_bytes=len(content),
117 disk_path=uri.replace("local://", ""),
118 storage_uri=uri,
119 )
120 .on_conflict_do_nothing(index_elements=["object_id"])
121 )
122 await session.execute(
123 pg_insert(db.MusehubObjectRef)
124 .values(repo_id=repo_id, object_id=oid)
125 .on_conflict_do_nothing()
126 )
127 await session.commit()
128 return oid
129
130
131 async def _collect_frames(gen: AsyncIterator[bytes]) -> list[JSONObject]:
132 frames = []
133 async for chunk in gen:
134 for frame in _unpack_chunks([chunk]):
135 frames.append(frame)
136 return frames
137
138
139 def _unpack_chunks(chunks: list[bytes]) -> list[dict]:
140 unpacker = msgpack.Unpacker(raw=False)
141 for chunk in chunks:
142 unpacker.feed(chunk)
143 return list(unpacker)
144
145
146 # ---------------------------------------------------------------------------
147 # S1 — WireFetchRequest accepts depth
148 # ---------------------------------------------------------------------------
149
150 class TestWireFetchRequestSchema:
151 def test_depth_defaults_to_none(self) -> None:
152 req = WireFetchRequest(want=[], have=[])
153 assert req.depth is None
154
155 def test_depth_accepts_positive_int(self) -> None:
156 req = WireFetchRequest(want=[], have=[], depth=1)
157 assert req.depth == 1
158
159 def test_depth_accepts_none_explicitly(self) -> None:
160 req = WireFetchRequest(want=[], have=[], depth=None)
161 assert req.depth is None
162
163
164 # ---------------------------------------------------------------------------
165 # S2 — depth=None returns full history
166 # ---------------------------------------------------------------------------
167
168 class TestFullHistory:
169 @pytest.mark.asyncio
170 async def test_no_depth_returns_all_commits(self, db_session: AsyncSession) -> None:
171 repo = await create_repo(db_session, owner="test-user-wire")
172 repo_id = str(repo.repo_id)
173
174 c1 = await _make_commit(db_session, repo_id, seed="root")
175 c2 = await _make_commit(db_session, repo_id, parent_ids=[c1.commit_id], seed="middle")
176 c3 = await _make_commit(db_session, repo_id, parent_ids=[c2.commit_id], seed="tip")
177
178 from musehub.services.musehub_wire import wire_fetch_stream
179 req = WireFetchRequest(want=[c3.commit_id], have=[], depth=None)
180 frames = await _collect_frames(wire_fetch_stream(db_session, repo_id, req))
181
182 c_frames = [f for f in frames if f.get("t") == SFRAME_COMMIT_PACK]
183 all_commit_ids = {c["commit_id"] for cf in c_frames for c in cf.get("commits", [])}
184 assert c1.commit_id in all_commit_ids
185 assert c2.commit_id in all_commit_ids
186 assert c3.commit_id in all_commit_ids
187
188
189 # ---------------------------------------------------------------------------
190 # S3 — depth=1 returns only the tip
191 # ---------------------------------------------------------------------------
192
193 class TestDepthOne:
194 @pytest.mark.asyncio
195 async def test_depth_1_returns_only_tip(self, db_session: AsyncSession) -> None:
196 repo = await create_repo(db_session, owner="test-user-wire")
197 repo_id = str(repo.repo_id)
198
199 c1 = await _make_commit(db_session, repo_id, seed="root")
200 c2 = await _make_commit(db_session, repo_id, parent_ids=[c1.commit_id], seed="middle")
201 c3 = await _make_commit(db_session, repo_id, parent_ids=[c2.commit_id], seed="tip")
202
203 from musehub.services.musehub_wire import wire_fetch_stream
204 req = WireFetchRequest(want=[c3.commit_id], have=[], depth=1)
205 frames = await _collect_frames(wire_fetch_stream(db_session, repo_id, req))
206
207 c_frames = [f for f in frames if f.get("t") == SFRAME_COMMIT_PACK]
208 all_commit_ids = {c["commit_id"] for cf in c_frames for c in cf.get("commits", [])}
209 assert c3.commit_id in all_commit_ids
210 assert c2.commit_id not in all_commit_ids
211 assert c1.commit_id not in all_commit_ids
212
213 @pytest.mark.asyncio
214 async def test_depth_1_stream_ends_cleanly(self, db_session: AsyncSession) -> None:
215 repo = await create_repo(db_session, owner="test-user-wire")
216 repo_id = str(repo.repo_id)
217
218 c1 = await _make_commit(db_session, repo_id, seed="a")
219 c2 = await _make_commit(db_session, repo_id, parent_ids=[c1.commit_id], seed="b")
220
221 from musehub.services.musehub_wire import wire_fetch_stream
222 req = WireFetchRequest(want=[c2.commit_id], have=[], depth=1)
223 frames = await _collect_frames(wire_fetch_stream(db_session, repo_id, req))
224
225 types = [f.get("t") for f in frames]
226 assert SFRAME_END in types
227
228
229 # ---------------------------------------------------------------------------
230 # S4 — depth=2 returns tip + one ancestor
231 # ---------------------------------------------------------------------------
232
233 class TestDepthTwo:
234 @pytest.mark.asyncio
235 async def test_depth_2_includes_parent_excludes_grandparent(
236 self, db_session: AsyncSession
237 ) -> None:
238 repo = await create_repo(db_session, owner="test-user-wire")
239 repo_id = str(repo.repo_id)
240
241 c1 = await _make_commit(db_session, repo_id, seed="root")
242 c2 = await _make_commit(db_session, repo_id, parent_ids=[c1.commit_id], seed="parent")
243 c3 = await _make_commit(db_session, repo_id, parent_ids=[c2.commit_id], seed="tip")
244
245 from musehub.services.musehub_wire import wire_fetch_stream
246 req = WireFetchRequest(want=[c3.commit_id], have=[], depth=2)
247 frames = await _collect_frames(wire_fetch_stream(db_session, repo_id, req))
248
249 c_frames = [f for f in frames if f.get("t") == SFRAME_COMMIT_PACK]
250 all_commit_ids = {c["commit_id"] for cf in c_frames for c in cf.get("commits", [])}
251 assert c3.commit_id in all_commit_ids
252 assert c2.commit_id in all_commit_ids
253 assert c1.commit_id not in all_commit_ids
254
255
256 # ---------------------------------------------------------------------------
257 # S5 — H frame carries shallow_commits when depth is active
258 # ---------------------------------------------------------------------------
259
260 class TestShallowHeader:
261 @pytest.mark.asyncio
262 async def test_h_frame_has_shallow_commits_on_depth(
263 self, db_session: AsyncSession
264 ) -> None:
265 repo = await create_repo(db_session, owner="test-user-wire")
266 repo_id = str(repo.repo_id)
267
268 c1 = await _make_commit(db_session, repo_id, seed="root")
269 c2 = await _make_commit(db_session, repo_id, parent_ids=[c1.commit_id], seed="tip")
270
271 from musehub.services.musehub_wire import wire_fetch_stream
272 req = WireFetchRequest(want=[c2.commit_id], have=[], depth=1)
273 frames = await _collect_frames(wire_fetch_stream(db_session, repo_id, req))
274
275 h = next(f for f in frames if f.get("t") == SFRAME_HEADER)
276 assert "shallow_commits" in h
277 assert c2.commit_id in h["shallow_commits"]
278
279 @pytest.mark.asyncio
280 async def test_h_frame_no_shallow_commits_without_depth(
281 self, db_session: AsyncSession
282 ) -> None:
283 repo = await create_repo(db_session, owner="test-user-wire")
284 repo_id = str(repo.repo_id)
285
286 c1 = await _make_commit(db_session, repo_id, seed="only")
287
288 from musehub.services.musehub_wire import wire_fetch_stream
289 req = WireFetchRequest(want=[c1.commit_id], have=[], depth=None)
290 frames = await _collect_frames(wire_fetch_stream(db_session, repo_id, req))
291
292 h = next(f for f in frames if f.get("t") == SFRAME_HEADER)
293 assert not h.get("shallow_commits")
294
295
296 # ---------------------------------------------------------------------------
297 # S6 — Only objects from depth-limited commits are sent
298 # ---------------------------------------------------------------------------
299
300 class TestShallowObjects:
301 @pytest.mark.asyncio
302 async def test_depth_1_only_sends_tip_objects(self, db_session: AsyncSession) -> None:
303 repo = await create_repo(db_session, owner="test-user-wire")
304 repo_id = str(repo.repo_id)
305
306 obj_root = await _make_object(db_session, repo_id, b"root-content")
307 obj_tip = await _make_object(db_session, repo_id, b"tip-content")
308
309 snap_root = await _make_snapshot(db_session, repo_id, {"root.dat": obj_root})
310 snap_tip = await _make_snapshot(db_session, repo_id, {"root.dat": obj_root, "tip.dat": obj_tip})
311
312 c1 = await _make_commit(db_session, repo_id, snapshot_id=snap_root.snapshot_id, seed="root")
313 c2 = await _make_commit(
314 db_session, repo_id, parent_ids=[c1.commit_id],
315 snapshot_id=snap_tip.snapshot_id, seed="tip",
316 )
317
318 from musehub.services.musehub_wire import wire_fetch_stream
319 req = WireFetchRequest(want=[c2.commit_id], have=[], depth=1)
320 frames = await _collect_frames(wire_fetch_stream(db_session, repo_id, req))
321
322 o_frames = [f for f in frames if f.get("t") == SFRAME_OBJECT]
323 received_oids = {f["id"] for f in o_frames}
324
325 # Both objects are in tip's snapshot — both should be sent
326 assert obj_tip in received_oids
327 assert obj_root in received_oids
328
329
330 # ---------------------------------------------------------------------------
331 # S7 — depth=1 on single-commit repo == full clone
332 # ---------------------------------------------------------------------------
333
334 class TestDepthOnSingleCommit:
335 @pytest.mark.asyncio
336 async def test_single_commit_depth_1_equals_full(
337 self, db_session: AsyncSession
338 ) -> None:
339 repo = await create_repo(db_session, owner="test-user-wire")
340 repo_id = str(repo.repo_id)
341
342 c1 = await _make_commit(db_session, repo_id, seed="only")
343
344 from musehub.services.musehub_wire import wire_fetch_stream
345
346 full = await _collect_frames(
347 wire_fetch_stream(db_session, repo_id, WireFetchRequest(want=[c1.commit_id], have=[], depth=None))
348 )
349 shallow = await _collect_frames(
350 wire_fetch_stream(db_session, repo_id, WireFetchRequest(want=[c1.commit_id], have=[], depth=1))
351 )
352
353 def _commit_ids(frames: list[JSONObject]) -> set[str]:
354 return {c["commit_id"] for f in frames if f.get("t") == SFRAME_COMMIT_PACK for c in f.get("commits", [])}
355
356 assert _commit_ids(full) == _commit_ids(shallow) == {c1.commit_id}
357
358
359 # ---------------------------------------------------------------------------
360 # S8 — depth on a DAG with two parents (merge commit)
361 # ---------------------------------------------------------------------------
362
363 class TestShallowDAG:
364 @pytest.mark.asyncio
365 async def test_depth_1_on_merge_commit_returns_only_merge(
366 self, db_session: AsyncSession
367 ) -> None:
368 repo = await create_repo(db_session, owner="test-user-wire")
369 repo_id = str(repo.repo_id)
370
371 base = await _make_commit(db_session, repo_id, seed="base")
372 feat = await _make_commit(db_session, repo_id, parent_ids=[base.commit_id], seed="feat")
373 merge = await _make_commit(
374 db_session, repo_id,
375 parent_ids=[base.commit_id, feat.commit_id],
376 seed="merge",
377 )
378
379 from musehub.services.musehub_wire import wire_fetch_stream
380 req = WireFetchRequest(want=[merge.commit_id], have=[], depth=1)
381 frames = await _collect_frames(wire_fetch_stream(db_session, repo_id, req))
382
383 c_frames = [f for f in frames if f.get("t") == SFRAME_COMMIT_PACK]
384 all_ids = {c["commit_id"] for cf in c_frames for c in cf.get("commits", [])}
385 assert merge.commit_id in all_ids
386 assert base.commit_id not in all_ids
387 assert feat.commit_id not in all_ids
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago