gabriel / muse public
test_fetch_presign_routing.py python
723 lines 27.2 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """TDD — fetch client presign routing: stream vs presigned R2.
2
3 The presigned fetch path mirrors the presigned push path:
4 - Small fetches (< 500 objects AND < 50 MB) go through fetch/stream as before.
5 - Large fetches call POST /fetch/presign first; if the server returns
6 presign=True it downloads the bundle directly from the presigned URL,
7 bypassing Cloudflare entirely.
8 - LocalFileTransport never presigns (loopback / local dev).
9
10 Test plan
11 ---------
12 Unit / integration
13 FPR0 Below object threshold → _use_fetch_presign returns False.
14 FPR1 At object threshold, non-loopback → _use_fetch_presign returns True.
15 FPR2 At byte threshold (via size hint from server), non-loopback → True.
16 FPR3 Loopback URL (file://) → always False regardless of counts.
17 FPR4 Server returns presign=False (LocalBackend on staging) → falls through
18 to fetch/stream normally; no crash.
19 FPR5 Server returns presign=True → client downloads bundle from presigned URL,
20 no call to fetch/stream endpoint.
21 FPR6 Presigned download writes correct objects (bundle bytes parsed and
22 dispatched through on_object callback).
23 FPR7 fetch_presign_or_stream is the new single entry-point: small repo →
24 delegates to fetch_stream unchanged.
25 FPR8 fetch_presign_or_stream: large repo, server presigns → returns same
26 FetchStreamResult shape as fetch_stream.
27 FPR9 Network error on presigned GET → raises TransportError (not silent drop).
28 FPR10 _FETCH_PRESIGN_OBJECT_THRESHOLD and _FETCH_PRESIGN_BYTE_THRESHOLD
29 match the server-side constants.
30
31 Security
32 FPRS0 LocalFileTransport.fetch_presign_or_stream never touches the presign
33 endpoint — always delegates to fetch_stream unconditionally.
34 FPRS1 file:// and localhost URLs are classified as loopback by routing logic.
35
36 Stress / state integrity
37 FPRST0 One of N parallel GETs returns non-200 → TransportError raised.
38 FPRST1 N=10 parallel GETs all succeed → all objects dispatched via on_object.
39
40 Performance
41 FPRP0 on_object receives every object when presign=True (no silent drops).
42 FPRP1 on_object is never called when presign_response has empty object_urls.
43 """
44 from __future__ import annotations
45
46 from unittest.mock import AsyncMock, MagicMock, patch, call
47 import io
48
49 import msgpack
50 import pytest
51
52 import pathlib
53
54 from muse.core.types import fake_id
55 from muse.core.paths import muse_dir
56 from muse.core.pack import ObjectPayload
57
58 type ObjectMap = dict[str, bytes]
59 type ObjectUrlMap = dict[str, str]
60
61 # ---------------------------------------------------------------------------
62 # FPR10 — constants match server side (verified first so mismatches are obvious)
63 # ---------------------------------------------------------------------------
64
65 def test_fpr10_threshold_constants_match_server() -> None:
66 """Client-side thresholds must equal server FETCH_PRESIGN_*_THRESHOLD."""
67 from muse.cli.commands.pull import (
68 _FETCH_PRESIGN_OBJECT_THRESHOLD,
69 _FETCH_PRESIGN_BYTE_THRESHOLD,
70 )
71 assert _FETCH_PRESIGN_OBJECT_THRESHOLD == 500
72 assert _FETCH_PRESIGN_BYTE_THRESHOLD == 50 * 1024 * 1024
73
74
75 # ---------------------------------------------------------------------------
76 # Helpers
77 # ---------------------------------------------------------------------------
78
79 _SERVER_OBJECT_THRESHOLD = 500
80 _SERVER_BYTE_THRESHOLD = 50 * 1024 * 1024
81
82
83 def _is_loopback(url: str) -> bool:
84 return url.startswith("file://") or "localhost" in url or "127.0.0.1" in url
85
86
87 def _use_fetch_presign(n_objects: int, total_bytes: int, is_loopback: bool) -> bool:
88 """Mirror the client routing decision from pull.py."""
89 from muse.cli.commands.pull import (
90 _FETCH_PRESIGN_OBJECT_THRESHOLD,
91 _FETCH_PRESIGN_BYTE_THRESHOLD,
92 )
93 return (
94 not is_loopback
95 and (
96 n_objects >= _FETCH_PRESIGN_OBJECT_THRESHOLD
97 or total_bytes >= _FETCH_PRESIGN_BYTE_THRESHOLD
98 )
99 )
100
101
102 _REPO_ID = fake_id("repo")
103 _COMMIT_ID = fake_id("commit")
104 _SNAP_ID = fake_id("snap")
105
106
107 def _make_objects(n: int) -> ObjectMap:
108 """Return {oid: raw_bytes} for n objects."""
109 return {fake_id(f"obj-{i}"): f"object-{i}".encode() for i in range(n)}
110
111
112 def _make_presign_response(object_urls: ObjectUrlMap, n_objects: int = 0) -> bytes:
113 """Build a presign=True msgpack response with per-object URLs."""
114 manifest = {f"file_{i}.py": oid for i, oid in enumerate(object_urls)}
115 return msgpack.packb({
116 "presign": True,
117 "object_urls": object_urls,
118 "commits": [{
119 "commit_id": _COMMIT_ID,
120 "snapshot_id": _SNAP_ID,
121 "message": "test",
122 "author": "gabriel",
123 "committed_at": "2026-04-30T00:00:00+00:00",
124 "parent_commit_id": None,
125 "parent2_commit_id": None,
126 "agent_id": "", "model_id": "", "toolchain_id": "",
127 "signer_public_key": "", "signature": "",
128 }],
129 "snapshots": [{"snapshot_id": _SNAP_ID, "manifest": manifest, "directories": [], "created_at": ""}],
130 "branch_heads": {"main": _COMMIT_ID},
131 "repo_id": _REPO_ID,
132 "domain": "code",
133 "default_branch": "main",
134 "expires_at": "2026-04-30T02:00:00+00:00",
135 "commit_count": 1,
136 "object_count": n_objects or len(object_urls),
137 }, use_bin_type=True)
138
139
140 # ---------------------------------------------------------------------------
141 # FPR0 — below threshold → stream path
142 # ---------------------------------------------------------------------------
143
144 def test_fpr0_below_threshold_uses_stream() -> None:
145 assert not _use_fetch_presign(
146 n_objects=_SERVER_OBJECT_THRESHOLD - 1,
147 total_bytes=0,
148 is_loopback=False,
149 )
150
151
152 # ---------------------------------------------------------------------------
153 # FPR1 — at object threshold, non-loopback → presign
154 # ---------------------------------------------------------------------------
155
156 def test_fpr1_object_threshold_triggers_presign() -> None:
157 assert _use_fetch_presign(
158 n_objects=_SERVER_OBJECT_THRESHOLD,
159 total_bytes=0,
160 is_loopback=False,
161 )
162
163
164 # ---------------------------------------------------------------------------
165 # FPR2 — at byte threshold → presign
166 # ---------------------------------------------------------------------------
167
168 def test_fpr2_byte_threshold_triggers_presign() -> None:
169 assert _use_fetch_presign(
170 n_objects=1,
171 total_bytes=_SERVER_BYTE_THRESHOLD,
172 is_loopback=False,
173 )
174
175
176 # ---------------------------------------------------------------------------
177 # FPR3 — loopback URL never presigns
178 # ---------------------------------------------------------------------------
179
180 def test_fpr3_loopback_never_presigns() -> None:
181 assert not _use_fetch_presign(
182 n_objects=_SERVER_OBJECT_THRESHOLD * 10,
183 total_bytes=_SERVER_BYTE_THRESHOLD * 10,
184 is_loopback=True,
185 )
186
187
188 # ---------------------------------------------------------------------------
189 # FPR4 — server returns presign=False → fallthrough to fetch/stream
190 # ---------------------------------------------------------------------------
191
192 @pytest.mark.asyncio
193 async def test_fpr4_server_presign_false_falls_through_to_stream() -> None:
194 """When server says presign=False, client must call fetch/stream normally."""
195 from muse.core.transport import HttpTransport, FetchStreamResult
196
197 transport = HttpTransport()
198 url = "https://staging.musehub.ai/gabriel/muse"
199 want = [fake_id("want")]
200 have: list[str] = []
201
202 presign_response = msgpack.packb({
203 "presign": False,
204 "object_count": 50,
205 "commit_count": 10,
206 }, use_bin_type=True)
207
208 fake_stream_result: FetchStreamResult = FetchStreamResult(
209 repo_id=fake_id("repo"),
210 domain="code",
211 default_branch="main",
212 branch_heads={"main": want[0]},
213 commits=[],
214 snapshots=[],
215 objects_received=0,
216 shallow_commits=[],
217 )
218
219 calls: list[str] = []
220
221 class _FakeResp:
222 def __init__(self, body: bytes, status: int = 200) -> None:
223 self.status_code = status
224 self.content = body
225
226 with patch.object(transport, "fetch_stream", return_value=fake_stream_result) as mock_stream, \
227 patch("httpx.Client") as mock_client_cls:
228
229 mock_client = MagicMock()
230 mock_client.__enter__ = MagicMock(return_value=mock_client)
231 mock_client.__exit__ = MagicMock(return_value=False)
232 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
233 mock_client_cls.return_value = mock_client
234
235 result = transport.fetch_presign_or_stream(
236 url, None, want=want, have=have, on_object=None,
237 )
238
239 mock_stream.assert_called_once_with(url, None, want=want, have=have, on_object=None)
240 assert result is fake_stream_result
241
242
243 # ---------------------------------------------------------------------------
244 # FPR5 — server returns presign=True → download from URL, no fetch/stream
245 # ---------------------------------------------------------------------------
246
247 @pytest.mark.asyncio
248 async def test_fpr5_server_presign_true_skips_stream() -> None:
249 """When server returns presign=True, client downloads per-object URLs; fetch/stream not called."""
250 from muse.core.transport import HttpTransport
251
252 transport = HttpTransport()
253 url = "https://staging.musehub.ai/gabriel/muse"
254 want = [fake_id("want")]
255 have: list[str] = []
256
257 objects = _make_objects(2)
258 object_urls = {oid: f"https://r2.example.com/{oid}?sig=x" for oid in objects}
259 presign_response = _make_presign_response(object_urls)
260
261 class _FakeResp:
262 def __init__(self, body: bytes, status: int = 200) -> None:
263 self.status_code = status
264 self.content = body
265
266 stream_called = []
267
268 with patch.object(transport, "fetch_stream", side_effect=lambda *a, **kw: stream_called.append(1)), \
269 patch("httpx.Client") as mock_client_cls:
270
271 mock_client = MagicMock()
272 mock_client.__enter__ = MagicMock(return_value=mock_client)
273 mock_client.__exit__ = MagicMock(return_value=False)
274 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
275 mock_client.get = MagicMock(side_effect=lambda u, **kw: _FakeResp(
276 objects.get(next((oid for oid in objects if oid in u), ""), b"raw-content")
277 ))
278 mock_client_cls.return_value = mock_client
279
280 result = transport.fetch_presign_or_stream(
281 url, None, want=want, have=have, on_object=None,
282 )
283
284 assert not stream_called, "fetch/stream must NOT be called when presign=True"
285 assert result["objects_received"] == 2
286 assert result["commit_count"] == 1
287
288
289 # ---------------------------------------------------------------------------
290 # FPR6 — presigned bundle bytes dispatched via on_object callback
291 # ---------------------------------------------------------------------------
292
293 def test_fpr6_presigned_objects_dispatched_via_on_object() -> None:
294 """Raw bytes from each per-object GET are dispatched through on_object."""
295 from muse.core.transport import HttpTransport
296 from muse.core.pack import ObjectPayload
297
298 transport = HttpTransport()
299 url = "https://staging.musehub.ai/gabriel/muse"
300 want = [fake_id("want")]
301
302 objects = _make_objects(3)
303 object_urls = {oid: f"https://r2.example.com/{oid}?sig=x" for oid in objects}
304 presign_response = _make_presign_response(object_urls)
305
306 received: list[ObjectPayload] = []
307
308 def _on_object(obj: ObjectPayload) -> None:
309 received.append(obj)
310
311 class _FakeResp:
312 def __init__(self, body: bytes, status: int = 200) -> None:
313 self.status_code = status
314 self.content = body
315
316 def _fake_get(u: str) -> _FakeResp:
317 for oid, content in objects.items():
318 if oid in u:
319 return _FakeResp(content)
320 return _FakeResp(b"unexpected")
321
322 with patch("httpx.Client") as mock_client_cls, \
323 patch.object(transport, "fetch_stream"):
324 mock_client = MagicMock()
325 mock_client.__enter__ = MagicMock(return_value=mock_client)
326 mock_client.__exit__ = MagicMock(return_value=False)
327 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
328 mock_client.get = MagicMock(side_effect=_fake_get)
329 mock_client_cls.return_value = mock_client
330
331 transport.fetch_presign_or_stream(
332 url, None, want=want, have=[], on_object=_on_object,
333 )
334
335 assert len(received) == 3
336 for obj in received:
337 assert obj["object_id"].startswith("sha256:")
338 assert obj["content"]
339
340
341 # ---------------------------------------------------------------------------
342 # FPR7 — small repo → fetch_presign_or_stream delegates to fetch_stream
343 # ---------------------------------------------------------------------------
344
345 def test_fpr7_small_repo_delegates_to_fetch_stream() -> None:
346 """fetch_presign_or_stream with small remote info → pure fetch_stream delegation."""
347 from muse.core.transport import HttpTransport, FetchStreamResult
348
349 transport = HttpTransport()
350 url = "https://staging.musehub.ai/gabriel/timing-test"
351 want = [fake_id("want")]
352 have: list[str] = []
353
354 # Server says presign=False (small repo, below threshold)
355 presign_response = msgpack.packb({
356 "presign": False,
357 "object_count": 42,
358 "commit_count": 10,
359 }, use_bin_type=True)
360
361 fake_result: FetchStreamResult = FetchStreamResult(
362 repo_id=fake_id("repo"),
363 domain="code",
364 default_branch="main",
365 branch_heads={"main": want[0]},
366 commits=[],
367 snapshots=[],
368 objects_received=42,
369 shallow_commits=[],
370 )
371
372 class _FakeResp:
373 def __init__(self, body: bytes, status: int = 200) -> None:
374 self.status_code = status
375 self.content = body
376
377 with patch.object(transport, "fetch_stream", return_value=fake_result) as mock_stream, \
378 patch("httpx.Client") as mock_client_cls:
379 mock_client = MagicMock()
380 mock_client.__enter__ = MagicMock(return_value=mock_client)
381 mock_client.__exit__ = MagicMock(return_value=False)
382 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
383 mock_client_cls.return_value = mock_client
384
385 result = transport.fetch_presign_or_stream(url, None, want=want, have=have)
386
387 mock_stream.assert_called_once()
388 assert result["objects_received"] == 42
389
390
391 # ---------------------------------------------------------------------------
392 # FPR8 — large repo, presign=True → FetchStreamResult shape correct
393 # ---------------------------------------------------------------------------
394
395 def test_fpr8_large_repo_presign_returns_correct_shape() -> None:
396 """fetch_presign_or_stream returns FetchStreamResult-compatible dict when presign=True."""
397 from muse.core.transport import HttpTransport
398
399 transport = HttpTransport()
400 url = "https://staging.musehub.ai/gabriel/muse"
401 want = [fake_id("want")]
402
403 objects = _make_objects(1)
404 object_urls = {oid: f"https://r2.example.com/{oid}?sig=abc" for oid in objects}
405 presign_response = _make_presign_response(object_urls)
406
407 class _FakeResp:
408 def __init__(self, body: bytes, status: int = 200) -> None:
409 self.status_code = status
410 self.content = body
411
412 def _fake_get(u: str) -> _FakeResp:
413 for oid, content in objects.items():
414 if oid in u:
415 return _FakeResp(content)
416 return _FakeResp(b"unexpected")
417
418 with patch("httpx.Client") as mock_client_cls, \
419 patch.object(transport, "fetch_stream"):
420 mock_client = MagicMock()
421 mock_client.__enter__ = MagicMock(return_value=mock_client)
422 mock_client.__exit__ = MagicMock(return_value=False)
423 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
424 mock_client.get = MagicMock(side_effect=_fake_get)
425 mock_client_cls.return_value = mock_client
426
427 result = transport.fetch_presign_or_stream(url, None, want=want, have=[])
428
429 # Must have same top-level keys as FetchStreamResult
430 for key in ("repo_id", "domain", "default_branch", "branch_heads",
431 "commits", "snapshots", "objects_received", "shallow_commits"):
432 assert key in result, f"missing key: {key}"
433 assert result["objects_received"] == 1
434 assert result["commit_count"] == 1
435
436
437 # ---------------------------------------------------------------------------
438 # FPR9 — network error on presigned GET → TransportError
439 # ---------------------------------------------------------------------------
440
441 def test_fpr9_presigned_get_error_raises_transport_error() -> None:
442 """A non-200 response from the presigned URL raises TransportError."""
443 from muse.core.transport import HttpTransport, TransportError
444
445 transport = HttpTransport()
446 url = "https://staging.musehub.ai/gabriel/muse"
447 want = [fake_id("want")]
448
449 objects = _make_objects(1)
450 object_urls = {oid: f"https://r2.example.com/{oid}?sig=abc" for oid in objects}
451 presign_response = _make_presign_response(object_urls)
452
453 class _FakeResp:
454 def __init__(self, body: bytes, status: int = 200) -> None:
455 self.status_code = status
456 self.content = body
457
458 with patch("httpx.Client") as mock_client_cls, \
459 patch.object(transport, "fetch_stream"):
460 mock_client = MagicMock()
461 mock_client.__enter__ = MagicMock(return_value=mock_client)
462 mock_client.__exit__ = MagicMock(return_value=False)
463 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
464 mock_client.get = MagicMock(return_value=_FakeResp(b"Access Denied", status=403))
465 mock_client_cls.return_value = mock_client
466
467 with pytest.raises(TransportError, match="403"):
468 transport.fetch_presign_or_stream(url, None, want=want, have=[])
469
470
471 # ===========================================================================
472 # Security tests
473 # ===========================================================================
474
475 # ---------------------------------------------------------------------------
476 # FPRS0 — LocalFileTransport never calls the presign endpoint
477 # ---------------------------------------------------------------------------
478
479 def test_fprs0_local_transport_never_hits_presign_endpoint(tmp_path: pathlib.Path) -> None:
480 """LocalFileTransport.fetch_presign_or_stream must delegate to fetch_stream, not presign."""
481 from muse.core.transport import LocalFileTransport, FetchStreamResult
482 import pathlib
483
484 dot_muse = muse_dir(tmp_path)
485 dot_muse.mkdir()
486 url = f"file://{tmp_path}"
487
488 fake_result: FetchStreamResult = FetchStreamResult(
489 repo_id=fake_id("repo"),
490 domain="code",
491 default_branch="main",
492 branch_heads={},
493 commits=[],
494 snapshots=[],
495 objects_received=0,
496 shallow_commits=[],
497 )
498
499 transport = LocalFileTransport()
500 with patch.object(transport, "fetch_stream", return_value=fake_result) as mock_stream, \
501 patch("httpx.Client") as mock_http:
502 result = transport.fetch_presign_or_stream(
503 url, None, want=[fake_id("w")], have=[],
504 )
505
506 mock_stream.assert_called_once()
507 mock_http.assert_not_called()
508 assert result is fake_result
509
510
511 # ---------------------------------------------------------------------------
512 # FPRS1 — file:// and localhost URLs classified as loopback
513 # ---------------------------------------------------------------------------
514
515 def test_fprs1_loopback_url_classification() -> None:
516 """file://, localhost, and 127.0.0.1 must be classified as loopback."""
517 assert _is_loopback("file:///home/gabriel/repo")
518 assert _is_loopback("https://localhost:1337/gabriel/muse")
519 assert _is_loopback("http://127.0.0.1:8000/gabriel/muse")
520 assert not _is_loopback("https://staging.musehub.ai/gabriel/muse")
521 assert not _is_loopback("https://musehub.ai/gabriel/muse")
522
523
524 # ===========================================================================
525 # Stress / state integrity tests
526 # ===========================================================================
527
528 # ---------------------------------------------------------------------------
529 # FPRST0 — one of N parallel GETs fails → TransportError
530 # ---------------------------------------------------------------------------
531
532 def test_fprst0_one_failing_get_raises_transport_error() -> None:
533 """If any parallel presigned GET returns non-200, TransportError is raised."""
534 from muse.core.transport import HttpTransport, TransportError
535
536 transport = HttpTransport()
537 url = "https://staging.musehub.ai/gabriel/muse"
538 want = [fake_id("want")]
539
540 n = 5
541 objects = _make_objects(n)
542 object_urls = {oid: f"https://r2.example.com/{oid}?sig=st0" for oid in objects}
543 presign_response = _make_presign_response(object_urls)
544
545 call_count = 0
546
547 class _FakeResp:
548 def __init__(self, body: bytes, status: int = 200) -> None:
549 self.status_code = status
550 self.content = body
551
552 def _failing_get(u: str) -> _FakeResp:
553 nonlocal call_count
554 call_count += 1
555 # Fail on the third request
556 if call_count == 3:
557 return _FakeResp(b"Internal Server Error", status=500)
558 for oid, content in objects.items():
559 if oid in u:
560 return _FakeResp(content)
561 return _FakeResp(b"not found", status=404)
562
563 with patch("httpx.Client") as mock_client_cls, \
564 patch.object(transport, "fetch_stream"):
565 mock_client = MagicMock()
566 mock_client.__enter__ = MagicMock(return_value=mock_client)
567 mock_client.__exit__ = MagicMock(return_value=False)
568 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
569 mock_client.get = MagicMock(side_effect=_failing_get)
570 mock_client_cls.return_value = mock_client
571
572 with pytest.raises(TransportError):
573 transport.fetch_presign_or_stream(url, None, want=want, have=[])
574
575
576 # ---------------------------------------------------------------------------
577 # FPRST1 — N=10 parallel GETs succeed → all dispatched via on_object
578 # ---------------------------------------------------------------------------
579
580 def test_fprst1_n_parallel_gets_all_dispatched() -> None:
581 """When all N presigned GETs succeed, on_object receives exactly N payloads."""
582 from muse.core.transport import HttpTransport
583 from muse.core.pack import ObjectPayload
584
585 n = 10
586 transport = HttpTransport()
587 url = "https://staging.musehub.ai/gabriel/muse"
588 want = [fake_id("want")]
589
590 objects = _make_objects(n)
591 object_urls = {oid: f"https://r2.example.com/{oid}?sig=st1" for oid in objects}
592 presign_response = _make_presign_response(object_urls)
593
594 received: list[ObjectPayload] = []
595
596 class _FakeResp:
597 def __init__(self, body: bytes, status: int = 200) -> None:
598 self.status_code = status
599 self.content = body
600
601 def _fake_get(u: str) -> _FakeResp:
602 for oid, content in objects.items():
603 if oid in u:
604 return _FakeResp(content)
605 return _FakeResp(b"unexpected")
606
607 with patch("httpx.Client") as mock_client_cls, \
608 patch.object(transport, "fetch_stream"):
609 mock_client = MagicMock()
610 mock_client.__enter__ = MagicMock(return_value=mock_client)
611 mock_client.__exit__ = MagicMock(return_value=False)
612 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
613 mock_client.get = MagicMock(side_effect=_fake_get)
614 mock_client_cls.return_value = mock_client
615
616 result = transport.fetch_presign_or_stream(
617 url, None, want=want, have=[], on_object=received.append,
618 )
619
620 assert len(received) == n
621 assert result["objects_received"] == n
622
623
624 # ===========================================================================
625 # Performance tests
626 # ===========================================================================
627
628 # ---------------------------------------------------------------------------
629 # FPRP0 — on_object receives every object when presign=True
630 # ---------------------------------------------------------------------------
631
632 def test_fprp0_on_object_receives_all_presigned_objects() -> None:
633 """on_object callback must be invoked once per object in the presigned map."""
634 from muse.core.transport import HttpTransport
635 from muse.core.pack import ObjectPayload
636
637 transport = HttpTransport()
638 url = "https://staging.musehub.ai/gabriel/muse"
639 want = [fake_id("want")]
640
641 objects = _make_objects(8)
642 object_urls = {oid: f"https://r2.example.com/{oid}?sig=p0" for oid in objects}
643 presign_response = _make_presign_response(object_urls)
644
645 dispatched_ids: set[str] = set()
646
647 class _FakeResp:
648 def __init__(self, body: bytes, status: int = 200) -> None:
649 self.status_code = status
650 self.content = body
651
652 def _fake_get(u: str) -> _FakeResp:
653 for oid, content in objects.items():
654 if oid in u:
655 return _FakeResp(content)
656 return _FakeResp(b"unexpected")
657
658 def _on_object(obj: ObjectPayload) -> None:
659 dispatched_ids.add(obj["object_id"])
660
661 with patch("httpx.Client") as mock_client_cls, \
662 patch.object(transport, "fetch_stream"):
663 mock_client = MagicMock()
664 mock_client.__enter__ = MagicMock(return_value=mock_client)
665 mock_client.__exit__ = MagicMock(return_value=False)
666 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
667 mock_client.get = MagicMock(side_effect=_fake_get)
668 mock_client_cls.return_value = mock_client
669
670 transport.fetch_presign_or_stream(url, None, want=want, have=[], on_object=_on_object)
671
672 assert dispatched_ids == set(objects.keys())
673
674
675 # ---------------------------------------------------------------------------
676 # FPRP1 — on_object never called when object_urls is empty
677 # ---------------------------------------------------------------------------
678
679 def test_fprp1_no_on_object_calls_for_empty_presign_response() -> None:
680 """When server returns presign=False (object_count=0), on_object must not be called."""
681 from muse.core.transport import HttpTransport, FetchStreamResult
682
683 transport = HttpTransport()
684 url = "https://staging.musehub.ai/gabriel/muse"
685 want = [fake_id("want")]
686
687 presign_response = msgpack.packb({
688 "presign": False,
689 "object_count": 0,
690 "commit_count": 0,
691 }, use_bin_type=True)
692
693 fake_result: FetchStreamResult = FetchStreamResult(
694 repo_id=fake_id("repo"),
695 domain="code",
696 default_branch="main",
697 branch_heads={},
698 commits=[],
699 snapshots=[],
700 objects_received=0,
701 shallow_commits=[],
702 )
703
704 on_object_calls: list[ObjectPayload] = []
705
706 class _FakeResp:
707 def __init__(self, body: bytes, status: int = 200) -> None:
708 self.status_code = status
709 self.content = body
710
711 with patch.object(transport, "fetch_stream", return_value=fake_result), \
712 patch("httpx.Client") as mock_client_cls:
713 mock_client = MagicMock()
714 mock_client.__enter__ = MagicMock(return_value=mock_client)
715 mock_client.__exit__ = MagicMock(return_value=False)
716 mock_client.post = MagicMock(return_value=_FakeResp(presign_response))
717 mock_client_cls.return_value = mock_client
718
719 transport.fetch_presign_or_stream(
720 url, None, want=want, have=[], on_object=on_object_calls.append,
721 )
722
723 assert on_object_calls == [], "on_object must not be called when presign=False"
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago