gabriel / musehub public
test_wire_fetch_presign.py python
860 lines 35.3 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """TDD — R2 presigned fetch path (per-object URL map, not bundle).
2
3 Problem
4 -------
5 Cloudflare times out streaming responses after ~100 seconds. For large fetches
6 the same presign pattern used for push should apply — but mirrored:
7
8 Push presign: server returns {oid → presigned PUT URL}, client PUTs in parallel.
9 Fetch presign: server returns {oid → presigned GET URL}, client GETs in parallel.
10
11 The server does a BFS walk, collects needed OIDs, calls backend.presign_get(oid, ttl)
12 for each one (no object reads), and returns the map. The client downloads all
13 objects in parallel directly from R2, bypassing Cloudflare entirely.
14
15 Architecture
16 ------------
17 ``POST /{owner}/{slug}/fetch/presign``
18
19 Request body (msgpack):
20 want list[str] — commit IDs the client wants
21 have list[str] — commit IDs the client already has (ancestry cut)
22 ttl_seconds int — presigned URL TTL (default 3600)
23
24 Response (msgpack):
25 presign bool — False when below threshold or backend can't presign
26 object_urls dict[str, str] — {oid: presigned_GET_url} for each needed object
27 commits list[dict] — commit records for apply_mpack
28 snapshots list[dict] — snapshot records for apply_mpack
29 branch_heads dict[str, str] — branch → tip commit_id
30 repo_id str
31 domain str
32 default_branch str
33 expires_at str|null — ISO-8601 expiry
34 object_count int
35 commit_count int
36
37 Threshold (same as push):
38 presign when ≥ 500 objects OR total raw size ≥ 50 MB
39
40 Test plan
41 ---------
42 Unit / integration
43 FP0 Below threshold → presign=False, object_urls={}.
44 FP1 Above object threshold with S3 backend → presign=True, object_urls has all OIDs.
45 FP2 Above size threshold (few but large objects) → presign=True.
46 FP3 have set excludes commits the client already has.
47 FP4 LocalBackend (no presign_get) → presign=False regardless of size.
48 FP5 Empty want list → presign=False, object_urls={}, counts zero.
49 FP6 commit_count and object_count are accurate across a commit chain.
50 FP7 Route 404 for missing repo.
51 FP8 Route returns presign=False for small public repo (no auth needed).
52 FP9 Route returns 404 for private repo without auth (don't leak existence).
53 FP10 object_urls map keys match exactly the new OIDs in the needed manifests.
54
55 Security
56 FPS0 TTL forwarded verbatim to presign_get; no negative or zero TTL.
57 FPS1 presign_get never called for OIDs in the have set.
58 FPS2 Private repo returns 404 even to a non-owner authenticated user.
59 FPS3 object_urls count equals new_oids count — no extras, no leakage.
60
61 Performance
62 FPP0 All presign_get calls complete even when N > semaphore limit (50).
63 FPP1 Custom ttl_seconds is honoured in the presigned URL.
64
65 Stress / state integrity
66 FPST0 600 OIDs → every OID appears in object_urls (no dropped presign calls).
67 FPST1 presign_get raising for one OID propagates the exception out.
68
69 End-to-end
70 FPE0 Full HTTP route with S3 backend mock → msgpack response contains
71 correct presign=True, all object_urls present and well-formed.
72 """
73 from __future__ import annotations
74
75 from datetime import datetime, timezone
76 from unittest.mock import AsyncMock, patch
77
78 import msgpack
79 import pytest
80 from httpx import AsyncClient
81 from sqlalchemy.dialects.postgresql import insert as pg_insert
82 from sqlalchemy.ext.asyncio import AsyncSession
83
84 from muse.core.types import blob_id, fake_id
85 from musehub.db import musehub_models as db
86 from musehub.models.wire import WireFetchRequest
87 from musehub.services.musehub_wire import (
88 FETCH_PRESIGN_OBJECT_THRESHOLD,
89 FETCH_PRESIGN_SIZE_THRESHOLD,
90 wire_fetch_presign,
91 )
92 from tests.factories import create_repo
93
94 # ---------------------------------------------------------------------------
95 # Helpers
96 # ---------------------------------------------------------------------------
97
98 def _now() -> datetime:
99 return datetime.now(tz=timezone.utc)
100
101
102 def _uid(seed: str) -> str:
103 return fake_id(seed)
104
105
106 async def _store_object(
107 session: AsyncSession,
108 repo_id: str,
109 oid: str,
110 content: bytes,
111 size_override: int | None = None,
112 ) -> None:
113 from musehub.services.musehub_wire import get_backend
114 backend = get_backend()
115 uri = await backend.put(oid, content)
116 await session.execute(
117 pg_insert(db.MusehubObject)
118 .values(
119 object_id=oid,
120 path="file.dat",
121 size_bytes=size_override if size_override is not None else len(content),
122 disk_path=uri.replace("local://", ""),
123 storage_uri=uri,
124 )
125 .on_conflict_do_nothing(index_elements=["object_id"])
126 )
127 await session.execute(
128 pg_insert(db.MusehubObjectRef)
129 .values(repo_id=repo_id, object_id=oid)
130 .on_conflict_do_nothing()
131 )
132 await session.commit()
133
134
135 async def _make_commit(
136 session: AsyncSession,
137 repo_id: str,
138 *,
139 manifest: dict[str, str],
140 seed: str = "c1",
141 parent_ids: list[str] | None = None,
142 ) -> tuple[db.MusehubCommit, db.MusehubSnapshot]:
143 snap_id = _uid(f"snap-{seed}")
144 snap = db.MusehubSnapshot(
145 snapshot_id=snap_id,
146 repo_id=repo_id,
147 directories=[],
148 manifest_blob=msgpack.packb(manifest, use_bin_type=True),
149 entry_count=len(manifest),
150 created_at=_now(),
151 )
152 session.add(snap)
153 commit_id = _uid(f"commit-{seed}")
154 commit = db.MusehubCommit(
155 commit_id=commit_id,
156 repo_id=repo_id,
157 branch="main",
158 parent_ids=parent_ids or [],
159 message=f"commit {seed}",
160 author="gabriel",
161 timestamp=_now(),
162 snapshot_id=snap_id,
163 )
164 session.add(commit)
165 await session.commit()
166 return commit, snap
167
168
169 # ---------------------------------------------------------------------------
170 # FP0 — below threshold → presign=False
171 # ---------------------------------------------------------------------------
172
173 @pytest.mark.asyncio
174 async def test_fp0_below_threshold_returns_presign_false(db_session: AsyncSession) -> None:
175 repo = await create_repo(db_session, owner="gabriel", visibility="public")
176 oid = blob_id(b"small object")
177 await _store_object(db_session, repo.repo_id, oid, b"small object")
178 commit, _ = await _make_commit(
179 db_session, repo.repo_id, manifest={"file.dat": oid}, seed="c1"
180 )
181 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
182 result = await wire_fetch_presign(db_session, repo.repo_id, req)
183
184 assert result["presign"] is False
185 assert result["object_urls"] == {}
186 assert result["commit_count"] == 1
187 assert result["object_count"] == 1
188
189
190 # ---------------------------------------------------------------------------
191 # FP1 — above object threshold with S3 backend → presign=True, per-object URLs
192 # ---------------------------------------------------------------------------
193
194 @pytest.mark.asyncio
195 async def test_fp1_above_object_threshold_s3_presigns(db_session: AsyncSession) -> None:
196 from musehub.services.musehub_wire import get_backend as _get_real_backend
197 repo = await create_repo(db_session, owner="gabriel", visibility="public")
198
199 manifest: dict[str, str] = {}
200 for i in range(FETCH_PRESIGN_OBJECT_THRESHOLD):
201 oid = blob_id(f"obj-{i}".encode())
202 await _store_object(db_session, repo.repo_id, oid, f"obj-{i}".encode())
203 manifest[f"file_{i}.dat"] = oid
204
205 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="big")
206 real_backend = _get_real_backend()
207
208 class _FakeS3:
209 supports_presign = True
210 get = real_backend.get
211 exists = real_backend.exists
212 put = AsyncMock(return_value="s3://bucket/obj")
213
214 async def presign_get(self, oid: str, ttl: int) -> str:
215 return f"https://r2.example.com/{oid}?sig=fake&ttl={ttl}"
216
217 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
218
219 with patch("musehub.services.musehub_wire.get_backend", return_value=_FakeS3()):
220 result = await wire_fetch_presign(db_session, repo.repo_id, req)
221
222 assert result["presign"] is True
223 assert len(result["object_urls"]) == FETCH_PRESIGN_OBJECT_THRESHOLD
224 # Every OID in the manifest must have a presigned URL
225 for oid in manifest.values():
226 assert oid in result["object_urls"]
227 assert result["object_urls"][oid].startswith("https://r2.example.com/")
228 assert result["object_count"] == FETCH_PRESIGN_OBJECT_THRESHOLD
229 assert result["commit_count"] == 1
230 assert result["expires_at"] is not None
231 # No bundle URL — per-object map only
232 assert "url" not in result or result.get("url") is None
233
234
235 # ---------------------------------------------------------------------------
236 # FP2 — above size threshold → presign=True
237 # ---------------------------------------------------------------------------
238
239 @pytest.mark.asyncio
240 async def test_fp2_above_size_threshold_presigns(db_session: AsyncSession) -> None:
241 from musehub.services.musehub_wire import get_backend as _get_real_backend
242 repo = await create_repo(db_session, owner="gabriel", visibility="public")
243 oid = blob_id(b"large-content-placeholder")
244 await _store_object(
245 db_session, repo.repo_id, oid, b"large-content-placeholder",
246 size_override=FETCH_PRESIGN_SIZE_THRESHOLD,
247 )
248 commit, _ = await _make_commit(
249 db_session, repo.repo_id, manifest={"big.dat": oid}, seed="big2"
250 )
251 real_backend = _get_real_backend()
252
253 class _FakeS3:
254 supports_presign = True
255 get = real_backend.get
256 exists = real_backend.exists
257 put = AsyncMock(return_value="s3://bucket/obj")
258
259 async def presign_get(self, oid: str, ttl: int) -> str:
260 return f"https://r2.example.com/{oid}?sig=fake"
261
262 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
263
264 with patch("musehub.services.musehub_wire.get_backend", return_value=_FakeS3()):
265 result = await wire_fetch_presign(db_session, repo.repo_id, req)
266
267 assert result["presign"] is True
268 assert result["object_count"] == 1
269 assert oid in result["object_urls"]
270
271
272 # ---------------------------------------------------------------------------
273 # FP3 — have set excludes already-known commits/objects
274 # ---------------------------------------------------------------------------
275
276 @pytest.mark.asyncio
277 async def test_fp3_have_set_excludes_known_objects(db_session: AsyncSession) -> None:
278 repo = await create_repo(db_session, owner="gabriel", visibility="public")
279 oid_a = blob_id(b"obj-a")
280 oid_b = blob_id(b"obj-b")
281 await _store_object(db_session, repo.repo_id, oid_a, b"obj-a")
282 await _store_object(db_session, repo.repo_id, oid_b, b"obj-b")
283
284 commit_a, _ = await _make_commit(
285 db_session, repo.repo_id, manifest={"a.dat": oid_a}, seed="a"
286 )
287 commit_b, _ = await _make_commit(
288 db_session, repo.repo_id,
289 manifest={"a.dat": oid_a, "b.dat": oid_b},
290 seed="b",
291 parent_ids=[commit_a.commit_id],
292 )
293
294 req = WireFetchRequest(want=[commit_b.commit_id], have=[commit_a.commit_id], depth=None)
295 result = await wire_fetch_presign(db_session, repo.repo_id, req)
296
297 assert result["commit_count"] == 1 # only commit_b is new
298 assert result["object_count"] == 1 # only oid_b is new
299 assert result["presign"] is False # 1 object is below threshold
300
301
302 # ---------------------------------------------------------------------------
303 # FP4 — LocalBackend → presign=False regardless of size
304 # ---------------------------------------------------------------------------
305
306 @pytest.mark.asyncio
307 async def test_fp4_local_backend_never_presigns(db_session: AsyncSession) -> None:
308 repo = await create_repo(db_session, owner="gabriel", visibility="public")
309 manifest: dict[str, str] = {}
310 for i in range(FETCH_PRESIGN_OBJECT_THRESHOLD):
311 oid = blob_id(f"local-obj-{i}".encode())
312 await _store_object(db_session, repo.repo_id, oid, f"local-obj-{i}".encode())
313 manifest[f"file_{i}.dat"] = oid
314
315 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="local-big")
316 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
317 result = await wire_fetch_presign(db_session, repo.repo_id, req)
318
319 assert result["presign"] is False
320 assert result["object_urls"] == {}
321
322
323 # ---------------------------------------------------------------------------
324 # FP5 — empty want → presign=False, all counts zero
325 # ---------------------------------------------------------------------------
326
327 @pytest.mark.asyncio
328 async def test_fp5_empty_want_returns_presign_false(db_session: AsyncSession) -> None:
329 repo = await create_repo(db_session, owner="gabriel", visibility="public")
330 req = WireFetchRequest(want=[], have=[], depth=None)
331 result = await wire_fetch_presign(db_session, repo.repo_id, req)
332
333 assert result["presign"] is False
334 assert result["object_urls"] == {}
335 assert result["commit_count"] == 0
336 assert result["object_count"] == 0
337
338
339 # ---------------------------------------------------------------------------
340 # FP6 — commit_count and object_count accurate across commit chain
341 # ---------------------------------------------------------------------------
342
343 @pytest.mark.asyncio
344 async def test_fp6_counts_accurate(db_session: AsyncSession) -> None:
345 repo = await create_repo(db_session, owner="gabriel", visibility="public")
346 oids = [blob_id(f"obj-{i}".encode()) for i in range(3)]
347 for i, oid in enumerate(oids):
348 await _store_object(db_session, repo.repo_id, oid, f"obj-{i}".encode())
349
350 c1, _ = await _make_commit(db_session, repo.repo_id, manifest={"a.dat": oids[0]}, seed="s1")
351 c2, _ = await _make_commit(
352 db_session, repo.repo_id,
353 manifest={"a.dat": oids[0], "b.dat": oids[1]},
354 seed="s2", parent_ids=[c1.commit_id],
355 )
356 c3, _ = await _make_commit(
357 db_session, repo.repo_id,
358 manifest={"a.dat": oids[0], "b.dat": oids[1], "c.dat": oids[2]},
359 seed="s3", parent_ids=[c2.commit_id],
360 )
361
362 req = WireFetchRequest(want=[c3.commit_id], have=[], depth=None)
363 result = await wire_fetch_presign(db_session, repo.repo_id, req)
364
365 assert result["commit_count"] == 3
366 assert result["object_count"] == 3 # 3 unique objects across all commits
367
368
369 # ---------------------------------------------------------------------------
370 # FP7 — route 404 for missing repo
371 # ---------------------------------------------------------------------------
372
373 @pytest.mark.asyncio
374 async def test_fp7_route_404_missing_repo(client: AsyncClient) -> None:
375 resp = await client.post(
376 "/nobody/no-such-repo/fetch/presign",
377 content=msgpack.packb({"want": [], "have": []}, use_bin_type=True),
378 headers={"Content-Type": "application/x-msgpack"},
379 )
380 assert resp.status_code == 404
381
382
383 # ---------------------------------------------------------------------------
384 # FP8 — route 200/presign=False for small public repo
385 # ---------------------------------------------------------------------------
386
387 @pytest.mark.asyncio
388 async def test_fp8_route_small_public_repo(
389 client: AsyncClient, db_session: AsyncSession, wire_headers: dict[str, str]
390 ) -> None:
391 repo = await create_repo(db_session, owner="gabriel", visibility="public")
392 oid = blob_id(b"tiny")
393 await _store_object(db_session, repo.repo_id, oid, b"tiny")
394 commit, _ = await _make_commit(
395 db_session, repo.repo_id, manifest={"t.dat": oid}, seed="tiny"
396 )
397
398 resp = await client.post(
399 f"/gabriel/{repo.slug}/fetch/presign",
400 content=msgpack.packb({"want": [commit.commit_id], "have": []}, use_bin_type=True),
401 headers={**wire_headers, "Content-Type": "application/x-msgpack"},
402 )
403 assert resp.status_code == 200
404 data = msgpack.unpackb(resp.content, raw=False)
405 assert data["presign"] is False
406
407
408 # ---------------------------------------------------------------------------
409 # FP9 — private repo returns 404 to non-owner
410 # ---------------------------------------------------------------------------
411
412 @pytest.mark.asyncio
413 async def test_fp9_route_private_repo_non_owner_gets_404(
414 client: AsyncClient, db_session: AsyncSession, wire_headers: dict[str, str]
415 ) -> None:
416 repo = await create_repo(db_session, owner="gabriel", visibility="private")
417 resp = await client.post(
418 f"/gabriel/{repo.slug}/fetch/presign",
419 content=msgpack.packb({"want": [], "have": []}, use_bin_type=True),
420 headers={**wire_headers, "Content-Type": "application/x-msgpack"},
421 )
422 assert resp.status_code == 404
423
424
425 # ---------------------------------------------------------------------------
426 # FP10 — object_urls keys match exactly the new OIDs in the needed manifests
427 # ---------------------------------------------------------------------------
428
429 @pytest.mark.asyncio
430 async def test_fp10_object_urls_keys_match_manifest_oids(db_session: AsyncSession) -> None:
431 """object_urls must contain exactly the OIDs from needed commits, not more, not less."""
432 from musehub.services.musehub_wire import get_backend as _get_real_backend
433 repo = await create_repo(db_session, owner="gabriel", visibility="public")
434 real_backend = _get_real_backend()
435
436 manifest: dict[str, str] = {}
437 expected_oids: set[str] = set()
438 for i in range(FETCH_PRESIGN_OBJECT_THRESHOLD):
439 oid = blob_id(f"fp10-obj-{i}".encode())
440 await _store_object(db_session, repo.repo_id, oid, f"fp10-obj-{i}".encode())
441 manifest[f"file_{i}.dat"] = oid
442 expected_oids.add(oid)
443
444 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="fp10")
445
446 class _FakeS3:
447 supports_presign = True
448 get = real_backend.get
449 exists = real_backend.exists
450 put = AsyncMock(return_value="s3://bucket/obj")
451
452 async def presign_get(self, oid: str, ttl: int) -> str:
453 return f"https://r2.example.com/{oid}?sig=fp10"
454
455 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
456
457 with patch("musehub.services.musehub_wire.get_backend", return_value=_FakeS3()):
458 result = await wire_fetch_presign(db_session, repo.repo_id, req)
459
460 assert result["presign"] is True
461 assert set(result["object_urls"].keys()) == expected_oids
462
463
464 # ===========================================================================
465 # Security tests
466 # ===========================================================================
467
468 # ---------------------------------------------------------------------------
469 # FPS0 — TTL forwarded verbatim to presign_get; never negative or zero
470 # ---------------------------------------------------------------------------
471
472 @pytest.mark.asyncio
473 async def test_fps0_ttl_forwarded_to_presign_get(db_session: AsyncSession) -> None:
474 """presign_get must receive the exact ttl_seconds argument; cannot produce negative TTL."""
475 from musehub.services.musehub_wire import get_backend as _get_real_backend
476 repo = await create_repo(db_session, owner="gabriel", visibility="public")
477 real_backend = _get_real_backend()
478
479 manifest: dict[str, str] = {}
480 for i in range(FETCH_PRESIGN_OBJECT_THRESHOLD):
481 oid = blob_id(f"fps0-{i}".encode())
482 await _store_object(db_session, repo.repo_id, oid, f"fps0-{i}".encode())
483 manifest[f"f{i}.dat"] = oid
484 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="fps0")
485
486 received_ttls: list[int] = []
487
488 class _RecordTTL:
489 supports_presign = True
490 get = real_backend.get
491 exists = real_backend.exists
492 put = AsyncMock(return_value="s3://bucket/obj")
493
494 async def presign_get(self, oid: str, ttl: int) -> str:
495 received_ttls.append(ttl)
496 assert ttl > 0, "TTL must be positive"
497 return f"https://r2.example.com/{oid}?ttl={ttl}"
498
499 custom_ttl = 1800
500 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
501
502 with patch("musehub.services.musehub_wire.get_backend", return_value=_RecordTTL()):
503 result = await wire_fetch_presign(db_session, repo.repo_id, req, ttl_seconds=custom_ttl)
504
505 assert result["presign"] is True
506 assert all(t == custom_ttl for t in received_ttls), "All presign_get calls must use the custom TTL"
507 assert len(received_ttls) == FETCH_PRESIGN_OBJECT_THRESHOLD
508
509
510 # ---------------------------------------------------------------------------
511 # FPS1 — presign_get never called for have-set OIDs
512 # ---------------------------------------------------------------------------
513
514 @pytest.mark.asyncio
515 async def test_fps1_presign_get_not_called_for_have_oids(db_session: AsyncSession) -> None:
516 """presign_get must not be invoked for objects the client already has.
517
518 Commit A carries base_oid (client already has commit A). Commit B adds
519 THRESHOLD new objects. The delta is exactly those THRESHOLD objects —
520 large enough to trigger presign. base_oid must not be presigned.
521 """
522 from musehub.services.musehub_wire import get_backend as _get_real_backend
523 repo = await create_repo(db_session, owner="gabriel", visibility="public")
524 real_backend = _get_real_backend()
525
526 # Commit A: one object that the client already has.
527 base_oid = blob_id(b"fps1-base")
528 await _store_object(db_session, repo.repo_id, base_oid, b"fps1-base")
529 commit_a, _ = await _make_commit(
530 db_session, repo.repo_id, manifest={"base.dat": base_oid}, seed="fps1a",
531 )
532
533 # Commit B: THRESHOLD new objects (delta large enough to trigger presign).
534 manifest_b: dict[str, str] = {"base.dat": base_oid}
535 new_oids: set[str] = set()
536 for i in range(FETCH_PRESIGN_OBJECT_THRESHOLD):
537 oid = blob_id(f"fps1-new-{i}".encode())
538 await _store_object(db_session, repo.repo_id, oid, f"fps1-new-{i}".encode())
539 manifest_b[f"new_{i}.dat"] = oid
540 new_oids.add(oid)
541 commit_b, _ = await _make_commit(
542 db_session, repo.repo_id, manifest=manifest_b,
543 seed="fps1b", parent_ids=[commit_a.commit_id],
544 )
545
546 presigned_oids: set[str] = set()
547
548 class _TrackCalls:
549 supports_presign = True
550 get = real_backend.get
551 exists = real_backend.exists
552 put = AsyncMock(return_value="s3://bucket/obj")
553
554 async def presign_get(self, oid: str, ttl: int) -> str:
555 presigned_oids.add(oid)
556 return f"https://r2.example.com/{oid}?sig=fps1"
557
558 # Client has commit_a — only the THRESHOLD new objects are delta.
559 req = WireFetchRequest(want=[commit_b.commit_id], have=[commit_a.commit_id], depth=None)
560
561 with patch("musehub.services.musehub_wire.get_backend", return_value=_TrackCalls()):
562 result = await wire_fetch_presign(db_session, repo.repo_id, req)
563
564 assert result["presign"] is True
565 assert base_oid not in presigned_oids, "base OID known by client must not be presigned"
566 assert new_oids == presigned_oids, "exactly the new OIDs must be presigned"
567 assert result["object_count"] == FETCH_PRESIGN_OBJECT_THRESHOLD
568
569
570 # ---------------------------------------------------------------------------
571 # FPS2 — private repo returns 404 to non-owner authenticated user
572 # ---------------------------------------------------------------------------
573
574 @pytest.mark.asyncio
575 async def test_fps2_private_repo_404_for_non_owner(
576 client: AsyncClient, db_session: AsyncSession, wire_headers: dict[str, str],
577 ) -> None:
578 """Authenticated non-owner must not discover a private repo via fetch/presign."""
579 # wire_headers injects handle="test-user-wire"; repo owner is "gabriel" — different user.
580 repo = await create_repo(db_session, owner="gabriel", visibility="private")
581 resp = await client.post(
582 f"/gabriel/{repo.slug}/fetch/presign",
583 content=msgpack.packb({"want": [], "have": []}, use_bin_type=True),
584 headers={**wire_headers, "Content-Type": "application/x-msgpack"},
585 )
586 assert resp.status_code == 404
587
588
589 # ---------------------------------------------------------------------------
590 # FPS3 — object_urls count equals new_oids count, no leakage
591 # ---------------------------------------------------------------------------
592
593 @pytest.mark.asyncio
594 async def test_fps3_object_urls_no_extras(db_session: AsyncSession) -> None:
595 """object_urls must contain exactly as many keys as new OIDs — no leakage."""
596 from musehub.services.musehub_wire import get_backend as _get_real_backend
597 repo = await create_repo(db_session, owner="gabriel", visibility="public")
598 real_backend = _get_real_backend()
599
600 manifest: dict[str, str] = {}
601 for i in range(FETCH_PRESIGN_OBJECT_THRESHOLD):
602 oid = blob_id(f"fps3-{i}".encode())
603 await _store_object(db_session, repo.repo_id, oid, f"fps3-{i}".encode())
604 manifest[f"f{i}.dat"] = oid
605 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="fps3")
606
607 class _FakeS3:
608 supports_presign = True
609 get = real_backend.get
610 exists = real_backend.exists
611 put = AsyncMock(return_value="s3://bucket/obj")
612
613 async def presign_get(self, oid: str, ttl: int) -> str:
614 return f"https://r2.example.com/{oid}?sig=fps3"
615
616 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
617 with patch("musehub.services.musehub_wire.get_backend", return_value=_FakeS3()):
618 result = await wire_fetch_presign(db_session, repo.repo_id, req)
619
620 assert result["presign"] is True
621 assert len(result["object_urls"]) == result["object_count"]
622 assert set(result["object_urls"].keys()) == set(manifest.values())
623
624
625 # ===========================================================================
626 # Performance tests
627 # ===========================================================================
628
629 # ---------------------------------------------------------------------------
630 # FPP0 — all presign_get calls complete when N > semaphore limit (50)
631 # ---------------------------------------------------------------------------
632
633 @pytest.mark.asyncio
634 async def test_fpp0_all_presign_calls_complete_above_semaphore_limit(
635 db_session: AsyncSession,
636 ) -> None:
637 """asyncio.gather with Semaphore(50) must complete all N>50 presign calls."""
638 from musehub.services.musehub_wire import get_backend as _get_real_backend
639 repo = await create_repo(db_session, owner="gabriel", visibility="public")
640 real_backend = _get_real_backend()
641
642 n = 75 # deliberately above the semaphore limit of 50
643 manifest: dict[str, str] = {}
644 for i in range(n):
645 oid = blob_id(f"fpp0-{i}".encode())
646 await _store_object(db_session, repo.repo_id, oid, f"fpp0-{i}".encode())
647 manifest[f"f{i}.dat"] = oid
648
649 # Force threshold so 75 objects triggers presign even though it's below 500.
650 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="fpp0")
651 presign_count = 0
652
653 class _CountingS3:
654 supports_presign = True
655 get = real_backend.get
656 exists = real_backend.exists
657 put = AsyncMock(return_value="s3://bucket/obj")
658
659 async def presign_get(self, oid: str, ttl: int) -> str:
660 nonlocal presign_count
661 presign_count += 1
662 return f"https://r2.example.com/{oid}?sig=fpp0"
663
664 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
665 # Patch both the backend AND the threshold so n=75 crosses it.
666 with patch("musehub.services.musehub_wire.get_backend", return_value=_CountingS3()), \
667 patch("musehub.services.musehub_wire.FETCH_PRESIGN_OBJECT_THRESHOLD", n - 1):
668 result = await wire_fetch_presign(db_session, repo.repo_id, req)
669
670 assert result["presign"] is True
671 assert presign_count == n, f"Expected {n} presign_get calls, got {presign_count}"
672 assert len(result["object_urls"]) == n
673
674
675 # ---------------------------------------------------------------------------
676 # FPP1 — custom ttl_seconds appears in expires_at timestamp
677 # ---------------------------------------------------------------------------
678
679 @pytest.mark.asyncio
680 async def test_fpp1_custom_ttl_reflected_in_expires_at(db_session: AsyncSession) -> None:
681 """expires_at must be approximately now() + ttl_seconds."""
682 from musehub.services.musehub_wire import get_backend as _get_real_backend
683 import dateutil.parser
684
685 repo = await create_repo(db_session, owner="gabriel", visibility="public")
686 real_backend = _get_real_backend()
687
688 manifest: dict[str, str] = {}
689 for i in range(FETCH_PRESIGN_OBJECT_THRESHOLD):
690 oid = blob_id(f"fpp1-{i}".encode())
691 await _store_object(db_session, repo.repo_id, oid, f"fpp1-{i}".encode())
692 manifest[f"f{i}.dat"] = oid
693 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="fpp1")
694
695 class _FakeS3:
696 supports_presign = True
697 get = real_backend.get
698 exists = real_backend.exists
699 put = AsyncMock(return_value="s3://bucket/obj")
700
701 async def presign_get(self, oid: str, ttl: int) -> str:
702 return f"https://r2.example.com/{oid}?ttl={ttl}"
703
704 before = datetime.now(tz=timezone.utc)
705 custom_ttl = 300
706 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
707
708 with patch("musehub.services.musehub_wire.get_backend", return_value=_FakeS3()):
709 result = await wire_fetch_presign(db_session, repo.repo_id, req, ttl_seconds=custom_ttl)
710
711 after = datetime.now(tz=timezone.utc)
712 assert result["expires_at"] is not None
713 expires_dt = dateutil.parser.parse(result["expires_at"])
714 from datetime import timedelta
715 assert before + timedelta(seconds=custom_ttl - 5) <= expires_dt <= after + timedelta(seconds=custom_ttl + 5)
716
717
718 # ===========================================================================
719 # Stress / state integrity tests
720 # ===========================================================================
721
722 # ---------------------------------------------------------------------------
723 # FPST0 — 600 OIDs, every one appears in object_urls
724 # ---------------------------------------------------------------------------
725
726 @pytest.mark.asyncio
727 async def test_fpst0_600_oids_all_presigned(db_session: AsyncSession) -> None:
728 """All 600 OIDs must appear in object_urls — no dropped presign calls."""
729 from musehub.services.musehub_wire import get_backend as _get_real_backend
730 repo = await create_repo(db_session, owner="gabriel", visibility="public")
731 real_backend = _get_real_backend()
732
733 n = 600 # well above FETCH_PRESIGN_OBJECT_THRESHOLD (500)
734 manifest: dict[str, str] = {}
735 all_oids: set[str] = set()
736 for i in range(n):
737 oid = blob_id(f"fpst0-{i}".encode())
738 await _store_object(db_session, repo.repo_id, oid, f"fpst0-{i}".encode())
739 manifest[f"f{i}.dat"] = oid
740 all_oids.add(oid)
741 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="fpst0")
742
743 class _FakeS3:
744 supports_presign = True
745 get = real_backend.get
746 exists = real_backend.exists
747 put = AsyncMock(return_value="s3://bucket/obj")
748
749 async def presign_get(self, oid: str, ttl: int) -> str:
750 return f"https://r2.example.com/{oid}?sig=fpst0"
751
752 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
753
754 with patch("musehub.services.musehub_wire.get_backend", return_value=_FakeS3()):
755 result = await wire_fetch_presign(db_session, repo.repo_id, req)
756
757 assert result["presign"] is True
758 assert result["object_count"] == n
759 missing = all_oids - set(result["object_urls"].keys())
760 assert not missing, f"Missing presigned URLs for {len(missing)} OIDs"
761
762
763 # ---------------------------------------------------------------------------
764 # FPST1 — presign_get raising propagates the exception
765 # ---------------------------------------------------------------------------
766
767 @pytest.mark.asyncio
768 async def test_fpst1_presign_get_exception_propagates(db_session: AsyncSession) -> None:
769 """If presign_get raises, wire_fetch_presign must propagate — no silent partial failure."""
770 from musehub.services.musehub_wire import get_backend as _get_real_backend
771 repo = await create_repo(db_session, owner="gabriel", visibility="public")
772 real_backend = _get_real_backend()
773
774 manifest: dict[str, str] = {}
775 for i in range(FETCH_PRESIGN_OBJECT_THRESHOLD):
776 oid = blob_id(f"fpst1-{i}".encode())
777 await _store_object(db_session, repo.repo_id, oid, f"fpst1-{i}".encode())
778 manifest[f"f{i}.dat"] = oid
779 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="fpst1")
780
781 call_count = 0
782
783 class _FailingS3:
784 supports_presign = True
785 get = real_backend.get
786 exists = real_backend.exists
787 put = AsyncMock(return_value="s3://bucket/obj")
788
789 async def presign_get(self, oid: str, ttl: int) -> str:
790 nonlocal call_count
791 call_count += 1
792 if call_count == 10:
793 raise RuntimeError("R2 presign service unavailable")
794 return f"https://r2.example.com/{oid}?sig=fpst1"
795
796 req = WireFetchRequest(want=[commit.commit_id], have=[], depth=None)
797
798 with patch("musehub.services.musehub_wire.get_backend", return_value=_FailingS3()), \
799 pytest.raises(RuntimeError, match="R2 presign service unavailable"):
800 await wire_fetch_presign(db_session, repo.repo_id, req)
801
802
803 # ===========================================================================
804 # End-to-end tests
805 # ===========================================================================
806
807 # ---------------------------------------------------------------------------
808 # FPE0 — full HTTP route with mocked S3 → msgpack response correct
809 # ---------------------------------------------------------------------------
810
811 @pytest.mark.asyncio
812 async def test_fpe0_route_presign_true_full_response(
813 client: AsyncClient, db_session: AsyncSession, wire_headers: dict[str, str],
814 ) -> None:
815 """Full HTTP round-trip: route → service → FakeS3 → msgpack response with object_urls."""
816 from musehub.services.musehub_wire import get_backend as _get_real_backend
817 real_backend = _get_real_backend()
818
819 repo = await create_repo(db_session, owner="gabriel", visibility="public")
820 manifest: dict[str, str] = {}
821 all_oids: set[str] = set()
822 for i in range(FETCH_PRESIGN_OBJECT_THRESHOLD):
823 oid = blob_id(f"fpe0-{i}".encode())
824 await _store_object(db_session, repo.repo_id, oid, f"fpe0-{i}".encode())
825 manifest[f"f{i}.dat"] = oid
826 all_oids.add(oid)
827 commit, _ = await _make_commit(db_session, repo.repo_id, manifest=manifest, seed="fpe0")
828
829 class _FakeS3:
830 supports_presign = True
831 get = real_backend.get
832 exists = real_backend.exists
833 put = AsyncMock(return_value="s3://bucket/obj")
834
835 async def presign_get(self, oid: str, ttl: int) -> str:
836 return f"https://r2.example.com/{oid}?sig=fpe0&ttl={ttl}"
837
838 with patch("musehub.services.musehub_wire.get_backend", return_value=_FakeS3()):
839 resp = await client.post(
840 f"/gabriel/{repo.slug}/fetch/presign",
841 content=msgpack.packb(
842 {"want": [commit.commit_id], "have": []}, use_bin_type=True
843 ),
844 headers={**wire_headers, "Content-Type": "application/x-msgpack"},
845 )
846
847 assert resp.status_code == 200
848 data = msgpack.unpackb(resp.content, raw=False)
849
850 assert data["presign"] is True
851 assert set(data["object_urls"].keys()) == all_oids
852 for oid, url in data["object_urls"].items():
853 assert url.startswith("https://r2.example.com/"), f"Unexpected URL: {url}"
854 assert oid in url, "OID must appear in its own presigned URL"
855 assert data["commit_count"] == 1
856 assert data["object_count"] == FETCH_PRESIGN_OBJECT_THRESHOLD
857 assert data["expires_at"] is not None
858 assert data["repo_id"] == repo.repo_id
859 # No legacy bundle URL field
860 assert "url" not in data or data.get("url") is None
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago