gabriel / musehub public
test_storage_backends.py python
870 lines 36.4 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 122 days ago
1 """Section 33 — Storage Backends: 7-layer test suite.
2
3 Covers:
4 musehub/storage/backends.py — StorageBackend protocol, LocalBackend, S3Backend, get_backend, decode_b64
5
6 Key behaviour under test:
7 LocalBackend:
8 - put/get/exists/delete round-trip (per-repo layout)
9 - _path requires repo_root — raises ValueError without it
10 - _path(oid, repo_root) produces objects/sha256/<2hex>/<62hex> layout
11 - put is idempotent (second write to same path is a no-op)
12 - uri_for returns "local://" prefix
13 - missing object → get returns None, exists returns False, delete is a no-op
14
15 S3Backend:
16 - _key replaces colons with underscores
17 - uri_for returns "s3://<bucket>/..." prefix
18 - put/get/exists/delete delegated to boto3 client (tested via mock)
19 - credential/boto3 errors on get/exists → return None/False (no raise)
20 - _s3_put always calls put_object (idempotent — HeadObject check removed; deduplication is upstream)
21
22 get_backend:
23 - returns LocalBackend when aws_s3_asset_bucket is unset
24 - returns S3Backend when aws_s3_asset_bucket is set
25
26 decode_b64:
27 - decodes standard base64
28 - handles missing padding (1, 2, 3 missing "=" chars)
29 """
30 from __future__ import annotations
31
32 import asyncio
33 import secrets
34 import time
35 from pathlib import Path
36 from types import SimpleNamespace
37 from typing import TypedDict
38 from unittest.mock import MagicMock, patch
39
40 import pytest
41
42 from musehub.storage.backends import LocalBackend, S3Backend, decode_b64, get_backend
43 from muse.core.types import long_id
44
45
46 # ── helpers ───────────────────────────────────────────────────────────────────
47
48
49 class _S3GetResponse(TypedDict):
50 Body: MagicMock
51
52
53 def _uid() -> str:
54 return secrets.token_hex(16)
55
56
57 def _backend() -> LocalBackend:
58 return LocalBackend()
59
60
61 def _repo_root(tmp_path: Path) -> Path:
62 return tmp_path / "repos" / "gabriel" / "test-repo"
63
64
65 # ═══════════════════════════════════════════════════════════════════════════════
66 # Layer 1 — Unit (pure logic, no filesystem I/O)
67 # ═══════════════════════════════════════════════════════════════════════════════
68
69
70 class TestUnitLocalBackend:
71 def test_path_requires_repo_root(self, tmp_path: Path) -> None:
72 b = _backend()
73 with pytest.raises(ValueError, match="repo_root"):
74 b._path(long_id("a" * 64), repo_root=None)
75
76 def test_path_with_repo_root_uses_algo_shard_layout(self, tmp_path: Path) -> None:
77 b = _backend()
78 repo_root = _repo_root(tmp_path)
79 hex64 = "ab" + "c" * 62
80 oid = long_id(hex64)
81 p = b._path(oid, repo_root=repo_root)
82 expected = repo_root / "objects" / "sha256" / "ab" / ("c" * 62)
83 assert p == expected
84
85 def test_uri_for_has_local_prefix(self, tmp_path: Path) -> None:
86 b = _backend()
87 repo_root = _repo_root(tmp_path)
88 oid = long_id("a" * 64)
89 uri = b.uri_for(oid, repo_root=repo_root)
90 assert uri.startswith("local://")
91
92 def test_uri_for_contains_obj(self, tmp_path: Path) -> None:
93 b = _backend()
94 repo_root = _repo_root(tmp_path)
95 oid = long_id("a" * 64)
96 uri = b.uri_for(oid, repo_root=repo_root)
97 assert "sha256" in uri
98
99
100 class TestUnitS3Backend:
101 def test_key_basic(self) -> None:
102 b = S3Backend(bucket="test-bucket", region="us-east-1")
103 key = b._key("abc123")
104 assert "abc123" in key
105
106 def test_key_replaces_colon(self) -> None:
107 b = S3Backend(bucket="b", region="us-east-1")
108 key = b._key("sha256:deadbeef")
109 assert ":" not in key
110 assert "sha256_deadbeef" in key
111
112 def test_uri_for_s3_prefix(self) -> None:
113 b = S3Backend(bucket="my-bucket", region="us-east-1")
114 uri = b.uri_for("obj1")
115 assert uri.startswith("s3://my-bucket/")
116
117 def test_uri_for_no_colon_in_key(self) -> None:
118 b = S3Backend(bucket="b", region="us-east-1")
119 uri = b.uri_for("sha256:abc")
120 assert "sha256:abc" not in uri
121 assert "sha256_abc" in uri
122
123
124 class TestUnitDecodeb64:
125 def test_standard_base64(self) -> None:
126 import base64
127 data = b"hello world"
128 encoded = base64.b64encode(data).decode()
129 assert decode_b64(encoded) == data
130
131 def test_missing_one_padding(self) -> None:
132 import base64
133 data = b"hi"
134 encoded = base64.b64encode(data).decode().rstrip("=")
135 assert decode_b64(encoded) == data
136
137 def test_missing_two_padding(self) -> None:
138 import base64
139 data = b"h"
140 encoded = base64.b64encode(data).decode().rstrip("=")
141 assert decode_b64(encoded) == data
142
143 def test_already_padded(self) -> None:
144 import base64
145 data = b"test"
146 encoded = base64.b64encode(data).decode()
147 assert decode_b64(encoded) == data
148
149 def test_empty_string(self) -> None:
150 assert decode_b64("") == b""
151
152
153 class TestUnitGetBackend:
154 def test_get_backend_returns_local_when_no_s3_bucket(self) -> None:
155 with patch("musehub.storage.backends.settings") as mock_settings:
156 mock_settings.r2_bucket = None
157 mock_settings.aws_s3_asset_bucket = None
158 result = get_backend()
159 assert isinstance(result, LocalBackend)
160
161 def test_get_backend_returns_s3_when_bucket_set(self) -> None:
162 with patch("musehub.storage.backends.settings") as mock_settings:
163 mock_settings.aws_s3_asset_bucket = "my-bucket"
164 mock_settings.aws_region = "us-east-1"
165 mock_settings.r2_bucket = None
166 result = get_backend()
167 assert isinstance(result, S3Backend)
168
169 def test_get_backend_with_owner_slug_returns_bound_local(self, tmp_path: Path) -> None:
170 """get_backend(owner, slug) returns a LocalBackend pre-bound to that repo_root."""
171 with patch("musehub.storage.backends.settings") as mock_settings:
172 mock_settings.r2_bucket = None
173 mock_settings.aws_s3_asset_bucket = None
174 mock_settings.musehub_repos_dir = str(tmp_path)
175 result = get_backend("gabriel", "myrepo")
176 assert isinstance(result, LocalBackend)
177 assert result._repo_root is not None
178
179 async def test_get_backend_bound_local_get_needs_no_repo_root_kwarg(self, tmp_path: Path) -> None:
180 """A backend from get_backend(owner, slug) satisfies storage.get(oid) with no extra args."""
181 with patch("musehub.storage.backends.settings") as mock_settings:
182 mock_settings.r2_bucket = None
183 mock_settings.aws_s3_asset_bucket = None
184 mock_settings.musehub_repos_dir = str(tmp_path)
185 backend = get_backend("gabriel", "myrepo")
186
187 oid = long_id("a" * 64)
188 await backend.put(oid, b"hello")
189 result = await backend.get(oid)
190 assert result == b"hello"
191
192 async def test_get_backend_bound_local_exists_needs_no_repo_root_kwarg(self, tmp_path: Path) -> None:
193 with patch("musehub.storage.backends.settings") as mock_settings:
194 mock_settings.r2_bucket = None
195 mock_settings.aws_s3_asset_bucket = None
196 mock_settings.musehub_repos_dir = str(tmp_path)
197 backend = get_backend("gabriel", "myrepo")
198
199 oid = long_id("b" * 64)
200 assert await backend.exists(oid) is False
201 await backend.put(oid, b"data")
202 assert await backend.exists(oid) is True
203
204 def test_get_backend_no_args_local_returns_unbound(self) -> None:
205 """get_backend() with no owner/slug still returns LocalBackend — unbound."""
206 with patch("musehub.storage.backends.settings") as mock_settings:
207 mock_settings.r2_bucket = None
208 mock_settings.aws_s3_asset_bucket = None
209 result = get_backend()
210 assert isinstance(result, LocalBackend)
211 assert result._repo_root is None
212
213
214 class TestUnitLocalBackendBoundInit:
215 """LocalBackend(repo_root=path) binds repo_root at construction; methods need no kwarg."""
216
217 def test_bound_backend_stores_repo_root(self, tmp_path: Path) -> None:
218 root = tmp_path / "gabriel" / "myrepo"
219 b = LocalBackend(repo_root=root)
220 assert b._repo_root == root
221
222 def test_unbound_backend_has_none_repo_root(self) -> None:
223 b = LocalBackend()
224 assert b._repo_root is None
225
226 async def test_bound_put_and_get_no_kwarg(self, tmp_path: Path) -> None:
227 root = tmp_path / "gabriel" / "myrepo"
228 b = LocalBackend(repo_root=root)
229 oid = long_id("c" * 64)
230 await b.put(oid, b"bound put")
231 result = await b.get(oid)
232 assert result == b"bound put"
233
234 async def test_bound_exists_no_kwarg(self, tmp_path: Path) -> None:
235 root = tmp_path / "gabriel" / "myrepo"
236 b = LocalBackend(repo_root=root)
237 oid = long_id("d" * 64)
238 assert await b.exists(oid) is False
239 await b.put(oid, b"x")
240 assert await b.exists(oid) is True
241
242 async def test_bound_delete_no_kwarg(self, tmp_path: Path) -> None:
243 root = tmp_path / "gabriel" / "myrepo"
244 b = LocalBackend(repo_root=root)
245 oid = long_id("e" * 64)
246 await b.put(oid, b"y")
247 await b.delete(oid)
248 assert await b.exists(oid) is False
249
250 async def test_kwarg_overrides_bound_repo_root(self, tmp_path: Path) -> None:
251 """Explicit repo_root kwarg at call time takes precedence over bound root."""
252 root_a = tmp_path / "a" / "repo"
253 root_b = tmp_path / "b" / "repo"
254 b = LocalBackend(repo_root=root_a)
255 oid = long_id("f" * 64)
256 await b.put(oid, b"in-b", repo_root=root_b)
257 assert await b.exists(oid, repo_root=root_b) is True
258 assert await b.exists(oid, repo_root=root_a) is False
259 assert await b.get(oid) is None # root_a doesn't have it
260 assert await b.get(oid, repo_root=root_b) == b"in-b"
261
262 async def test_unbound_get_without_kwarg_raises(self) -> None:
263 b = LocalBackend()
264 oid = long_id("0" * 64)
265 with pytest.raises(ValueError, match="repo_root"):
266 await b.get(oid)
267
268
269 # ═══════════════════════════════════════════════════════════════════════════════
270 # Layer 2 — Integration (real filesystem I/O via tmp_path)
271 # ═══════════════════════════════════════════════════════════════════════════════
272
273
274 class TestIntegrationLocalBackend:
275 async def test_put_and_get_round_trip(self, tmp_path: Path) -> None:
276 b = _backend()
277 repo_root = _repo_root(tmp_path)
278 data = b"hello storage"
279 oid = long_id("a" * 64)
280 uri = await b.put(oid, data, repo_root=repo_root)
281 assert uri.startswith("local://")
282 result = await b.get(oid, repo_root=repo_root)
283 assert result == data
284
285 async def test_get_missing_returns_none(self, tmp_path: Path) -> None:
286 b = _backend()
287 repo_root = _repo_root(tmp_path)
288 oid = long_id("b" * 64)
289 result = await b.get(oid, repo_root=repo_root)
290 assert result is None
291
292 async def test_exists_after_put(self, tmp_path: Path) -> None:
293 b = _backend()
294 repo_root = _repo_root(tmp_path)
295 oid = long_id("c" * 64)
296 assert await b.exists(oid, repo_root=repo_root) is False
297 await b.put(oid, b"data", repo_root=repo_root)
298 assert await b.exists(oid, repo_root=repo_root) is True
299
300 async def test_exists_missing_returns_false(self, tmp_path: Path) -> None:
301 b = _backend()
302 repo_root = _repo_root(tmp_path)
303 oid = long_id("d" * 64)
304 assert await b.exists(oid, repo_root=repo_root) is False
305
306 async def test_delete_removes_object(self, tmp_path: Path) -> None:
307 b = _backend()
308 repo_root = _repo_root(tmp_path)
309 oid = long_id("e" * 64)
310 await b.put(oid, b"bye", repo_root=repo_root)
311 assert await b.exists(oid, repo_root=repo_root) is True
312 await b.delete(oid, repo_root=repo_root)
313 assert await b.exists(oid, repo_root=repo_root) is False
314
315 async def test_delete_nonexistent_is_noop(self, tmp_path: Path) -> None:
316 b = _backend()
317 repo_root = _repo_root(tmp_path)
318 oid = long_id("f" * 64)
319 await b.delete(oid, repo_root=repo_root)
320
321 async def test_put_idempotent_same_bytes(self, tmp_path: Path) -> None:
322 """Putting the same bytes twice is a no-op — content-addressed dedup."""
323 b = _backend()
324 repo_root = _repo_root(tmp_path)
325 oid = long_id("0" * 64)
326 await b.put(oid, b"content", repo_root=repo_root)
327 await b.put(oid, b"content", repo_root=repo_root)
328 result = await b.get(oid, repo_root=repo_root)
329 assert result == b"content"
330
331 async def test_put_creates_parent_dirs(self, tmp_path: Path) -> None:
332 b = _backend()
333 repo_root = _repo_root(tmp_path)
334 oid = long_id("1" * 64)
335 await b.put(oid, b"data", repo_root=repo_root)
336 assert await b.exists(oid, repo_root=repo_root) is True
337
338 async def test_sha256_oid_round_trip(self, tmp_path: Path) -> None:
339 b = _backend()
340 repo_root = _repo_root(tmp_path)
341 oid = long_id("ab" + "c" * 62)
342 data = b"sha256 content"
343 await b.put(oid, data, repo_root=repo_root)
344 result = await b.get(oid, repo_root=repo_root)
345 assert result == data
346
347 async def test_per_repo_isolation(self, tmp_path: Path) -> None:
348 """Objects in different repos are isolated — same oid, different repo_root."""
349 b = _backend()
350 repo_a = tmp_path / "repos" / "alice" / "repo-a"
351 repo_b = tmp_path / "repos" / "bob" / "repo-b"
352 oid = long_id("2" * 64)
353 await b.put(oid, b"alice data", repo_root=repo_a)
354 assert await b.exists(oid, repo_root=repo_b) is False
355 assert await b.get(oid, repo_root=repo_a) == b"alice data"
356
357
358 class TestIntegrationS3BackendMocked:
359 """S3Backend with a mock boto3 client — no real AWS calls."""
360
361 def _mock_s3_backend(self, *, head_raises: bool = False) -> S3Backend:
362 b = S3Backend(bucket="test-bucket", region="us-east-1")
363 mock_client = MagicMock()
364 if head_raises:
365 mock_client.head_object.side_effect = Exception("NoSuchKey")
366 b._client = mock_client
367 return b
368
369 async def test_put_calls_put_object_directly(self) -> None:
370 """put() is idempotent — no HeadObject check, just put_object unconditionally."""
371 b = self._mock_s3_backend()
372 await b.put("obj1", b"data")
373 b._client.put_object.assert_called_once()
374 b._client.head_object.assert_not_called()
375
376 async def test_put_is_idempotent(self) -> None:
377 """Uploading the same object twice both succeed — put_object called twice."""
378 b = self._mock_s3_backend()
379 await b.put("obj1", b"data")
380 await b.put("obj1", b"data")
381 assert b._client.put_object.call_count == 2
382
383 async def test_get_returns_body_bytes(self) -> None:
384 b = self._mock_s3_backend()
385 mock_body = MagicMock()
386 mock_body.read.return_value = b"s3 content"
387 b._client.get_object.return_value = {"Body": mock_body}
388 result = await b.get("obj1")
389 assert result == b"s3 content"
390
391 async def test_get_returns_none_on_error(self) -> None:
392 b = self._mock_s3_backend()
393 b._client.get_object.side_effect = Exception("NoSuchKey")
394 result = await b.get("obj1")
395 assert result is None
396
397 async def test_exists_returns_true_when_head_succeeds(self) -> None:
398 b = self._mock_s3_backend()
399 b._client.head_object.return_value = {}
400 assert await b.exists("obj1") is True
401
402 async def test_exists_returns_false_on_error(self) -> None:
403 b = self._mock_s3_backend(head_raises=True)
404 assert await b.exists("obj1") is False
405
406 async def test_delete_calls_delete_object(self) -> None:
407 b = self._mock_s3_backend()
408 await b.delete("obj1")
409 b._client.delete_object.assert_called_once()
410
411 async def test_get_batch_returns_all_found_objects(self) -> None:
412 """get_batch maps every object_id to its bytes when S3 finds all of them."""
413 b = self._mock_s3_backend()
414
415 def _get_object(Bucket: str, Key: str) -> _S3GetResponse:
416 mock_body = MagicMock()
417 mock_body.read.return_value = Key.encode()
418 return {"Body": mock_body}
419
420 b._client.get_object.side_effect = _get_object
421 result = await b.get_batch(["oid-a", "oid-b", "oid-c"])
422 assert set(result.keys()) == {"oid-a", "oid-b", "oid-c"}
423
424 async def test_get_batch_omits_missing_objects(self) -> None:
425 """get_batch omits object_ids where S3 raises (NoSuchKey etc.)."""
426 b = self._mock_s3_backend()
427 b._client.get_object.side_effect = Exception("NoSuchKey")
428 result = await b.get_batch(["oid-1", "oid-2"])
429 assert result == {}
430
431 async def test_get_batch_is_parallel_not_sequential(self) -> None:
432 """N parallel S3 gets must complete in ~1× per-object latency, not N×."""
433 import asyncio
434
435 DELAY = 0.04
436 N = 6
437
438 b = S3Backend(bucket="test-bucket", region="us-east-1")
439
440 async def _slow_get(oid: str) -> bytes | None:
441 await asyncio.sleep(DELAY)
442 return oid.encode()
443
444 setattr(b, "get", _slow_get)
445
446 obj_ids = [f"oid-{i}" for i in range(N)]
447 start = time.perf_counter()
448 result = await b.get_batch(obj_ids)
449 elapsed = time.perf_counter() - start
450
451 assert len(result) == N
452 assert elapsed < DELAY * 2.5, (
453 f"get_batch took {elapsed * 1000:.0f} ms; expected ≈{DELAY * 1000:.0f} ms "
454 f"(parallel). Sequential would take ≥{DELAY * N * 1000:.0f} ms. "
455 "S3Backend.get_batch must use asyncio.gather, not the base-class loop."
456 )
457
458
459 # ═══════════════════════════════════════════════════════════════════════════════
460 # Layer 3 — End-to-End (LocalBackend as the full stack backend)
461 # ═══════════════════════════════════════════════════════════════════════════════
462
463
464 class TestE2ELocalBackend:
465 async def test_full_lifecycle(self, tmp_path: Path) -> None:
466 """put → exists(True) → get → delete → exists(False) → get(None)."""
467 b = _backend()
468 repo_root = _repo_root(tmp_path)
469 obj_id = long_id("3" * 64)
470 data = b"end-to-end content"
471
472 uri = await b.put(obj_id, data, repo_root=repo_root)
473 assert uri.startswith("local://")
474
475 assert await b.exists(obj_id, repo_root=repo_root) is True
476 assert await b.get(obj_id, repo_root=repo_root) == data
477
478 await b.delete(obj_id, repo_root=repo_root)
479 assert await b.exists(obj_id, repo_root=repo_root) is False
480 assert await b.get(obj_id, repo_root=repo_root) is None
481
482 async def test_binary_data_preserved(self, tmp_path: Path) -> None:
483 b = _backend()
484 repo_root = _repo_root(tmp_path)
485 oid = long_id("4" * 64)
486 data = bytes(range(256))
487 await b.put(oid, data, repo_root=repo_root)
488 result = await b.get(oid, repo_root=repo_root)
489 assert result == data
490
491 async def test_large_object(self, tmp_path: Path) -> None:
492 b = _backend()
493 repo_root = _repo_root(tmp_path)
494 oid = long_id("5" * 64)
495 data = b"x" * (4 * 1024 * 1024)
496 await b.put(oid, data, repo_root=repo_root)
497 result = await b.get(oid, repo_root=repo_root)
498 assert result == data
499
500 async def test_uri_resolves_to_disk_path(self, tmp_path: Path) -> None:
501 b = _backend()
502 repo_root = _repo_root(tmp_path)
503 oid = long_id("6" * 64)
504 data = b"uri content"
505 uri = await b.put(oid, data, repo_root=repo_root)
506 disk_path = Path(uri.replace("local://", ""))
507 assert disk_path.exists()
508 assert disk_path.read_bytes() == data
509
510
511 # ═══════════════════════════════════════════════════════════════════════════════
512 # Layer 4 — Stress
513 # ═══════════════════════════════════════════════════════════════════════════════
514
515
516 class TestStressLocalBackend:
517 async def test_write_many_objects(self, tmp_path: Path) -> None:
518 b = _backend()
519 repo_root = _repo_root(tmp_path)
520 obj_ids = [long_id(str(i).zfill(64)) for i in range(100)]
521 for oid in obj_ids:
522 await b.put(oid, oid.encode(), repo_root=repo_root)
523 for oid in obj_ids:
524 assert await b.exists(oid, repo_root=repo_root) is True
525
526 async def test_concurrent_writes_different_objects(self, tmp_path: Path) -> None:
527 b = _backend()
528 repo_root = _repo_root(tmp_path)
529 obj_ids = [long_id(str(i).zfill(64)) for i in range(20)]
530 await asyncio.gather(*[b.put(oid, oid.encode(), repo_root=repo_root) for oid in obj_ids])
531 for oid in obj_ids:
532 assert await b.get(oid, repo_root=repo_root) == oid.encode()
533
534 async def test_concurrent_reads(self, tmp_path: Path) -> None:
535 b = _backend()
536 repo_root = _repo_root(tmp_path)
537 oid = long_id("7" * 64)
538 await b.put(oid, b"shared content", repo_root=repo_root)
539 results = await asyncio.gather(*[b.get(oid, repo_root=repo_root) for _ in range(20)])
540 assert all(r == b"shared content" for r in results)
541
542
543 # ═══════════════════════════════════════════════════════════════════════════════
544 # Layer 5 — Data Integrity
545 # ═══════════════════════════════════════════════════════════════════════════════
546
547
548 class TestDataIntegrityLocalBackend:
549 async def test_put_returns_deterministic_uri(self, tmp_path: Path) -> None:
550 b = _backend()
551 repo_root = _repo_root(tmp_path)
552 oid = long_id("8" * 64)
553 uri1 = await b.put(oid, b"data", repo_root=repo_root)
554 uri2 = b.uri_for(oid, repo_root=repo_root)
555 assert uri1 == uri2
556
557 async def test_data_not_corrupted(self, tmp_path: Path) -> None:
558 b = _backend()
559 repo_root = _repo_root(tmp_path)
560 oid = long_id("9" * 64)
561 payload = b"\x00\x01\x02\x03" * 1000
562 await b.put(oid, payload, repo_root=repo_root)
563 result = await b.get(oid, repo_root=repo_root)
564 assert result == payload
565
566 async def test_put_is_idempotent_for_identical_bytes(self, tmp_path: Path) -> None:
567 b = _backend()
568 repo_root = _repo_root(tmp_path)
569 oid = long_id("a" * 64)
570 await b.put(oid, b"original", repo_root=repo_root)
571 await b.put(oid, b"original", repo_root=repo_root)
572 result = await b.get(oid, repo_root=repo_root)
573 assert result == b"original"
574
575 async def test_put_repairs_corrupted_file(self, tmp_path: Path) -> None:
576 """put() must overwrite an existing file whose bytes differ from the new data."""
577 import zlib
578 b = _backend()
579 repo_root = _repo_root(tmp_path)
580 raw = b"# real raw content"
581 corrupt = zlib.compress(raw)
582 oid = long_id("b" * 64)
583
584 path = b._path(oid, repo_root=repo_root)
585 path.parent.mkdir(parents=True, exist_ok=True)
586 path.write_bytes(corrupt)
587 import stat
588 path.chmod(stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
589
590 await b.put(oid, raw, repo_root=repo_root)
591
592 result = await b.get(oid, repo_root=repo_root)
593 assert result == raw, (
594 f"expected raw content but got {result[:20]!r} — "
595 "LocalBackend.put() failed to repair the corrupted file"
596 )
597
598 async def test_delete_only_removes_target(self, tmp_path: Path) -> None:
599 b = _backend()
600 repo_root = _repo_root(tmp_path)
601 oid_keep = long_id("c" * 64)
602 oid_remove = long_id("d" * 64)
603 await b.put(oid_keep, b"keep", repo_root=repo_root)
604 await b.put(oid_remove, b"remove", repo_root=repo_root)
605 await b.delete(oid_remove, repo_root=repo_root)
606 assert await b.get(oid_keep, repo_root=repo_root) == b"keep"
607 assert await b.get(oid_remove, repo_root=repo_root) is None
608
609 async def test_empty_bytes_stored_and_retrieved(self, tmp_path: Path) -> None:
610 b = _backend()
611 repo_root = _repo_root(tmp_path)
612 oid = long_id("e" * 64)
613 await b.put(oid, b"", repo_root=repo_root)
614 result = await b.get(oid, repo_root=repo_root)
615 assert result == b""
616
617
618 # ═══════════════════════════════════════════════════════════════════════════════
619 # Layer 6 — Security
620 # ═══════════════════════════════════════════════════════════════════════════════
621
622
623 class TestSecurityLocalBackend:
624 def test_s3_key_no_colon_injection(self) -> None:
625 b = S3Backend(bucket="b", region="us-east-1")
626 key = b._key("sha256:evil:colons")
627 assert ":" not in key
628
629 async def test_put_without_repo_root_raises(self, tmp_path: Path) -> None:
630 b = _backend()
631 oid = long_id("f" * 64)
632 with pytest.raises(ValueError, match="repo_root"):
633 await b.put(oid, b"data", repo_root=None)
634
635 async def test_get_without_repo_root_raises(self, tmp_path: Path) -> None:
636 b = _backend()
637 oid = long_id("0" * 64)
638 with pytest.raises(ValueError, match="repo_root"):
639 await b.get(oid, repo_root=None)
640
641 async def test_exists_without_repo_root_raises(self, tmp_path: Path) -> None:
642 b = _backend()
643 oid = long_id("1" * 64)
644 with pytest.raises(ValueError, match="repo_root"):
645 await b.exists(oid, repo_root=None)
646
647
648 # ═══════════════════════════════════════════════════════════════════════════════
649 # Layer 7 — Performance
650 # ═══════════════════════════════════════════════════════════════════════════════
651
652
653 class TestPerformanceLocalBackend:
654 async def test_put_latency(self, tmp_path: Path) -> None:
655 b = _backend()
656 repo_root = _repo_root(tmp_path)
657 oid = long_id("2" * 64)
658 data = b"perf-test-payload" * 100
659 start = time.perf_counter()
660 await b.put(oid, data, repo_root=repo_root)
661 elapsed = time.perf_counter() - start
662 assert elapsed < 0.5
663
664 async def test_get_latency(self, tmp_path: Path) -> None:
665 b = _backend()
666 repo_root = _repo_root(tmp_path)
667 oid = long_id("3" * 64)
668 data = b"x" * (512 * 1024)
669 await b.put(oid, data, repo_root=repo_root)
670
671 start = time.perf_counter()
672 result = await b.get(oid, repo_root=repo_root)
673 elapsed = time.perf_counter() - start
674
675 assert result == data
676 assert elapsed < 0.5
677
678 async def test_exists_latency(self, tmp_path: Path) -> None:
679 b = _backend()
680 repo_root = _repo_root(tmp_path)
681 oid = long_id("4" * 64)
682 await b.put(oid, b"data", repo_root=repo_root)
683
684 start = time.perf_counter()
685 for _ in range(50):
686 await b.exists(oid, repo_root=repo_root)
687 elapsed = time.perf_counter() - start
688
689 assert elapsed < 1.0
690
691 async def test_50_sequential_puts_under_budget(self, tmp_path: Path) -> None:
692 b = _backend()
693 repo_root = _repo_root(tmp_path)
694 data = b"payload" * 100
695
696 start = time.perf_counter()
697 for i in range(50):
698 oid = long_id(str(i).zfill(64))
699 await b.put(oid, data, repo_root=repo_root)
700 elapsed = time.perf_counter() - start
701
702 assert elapsed < 2.0
703
704
705 # ═══════════════════════════════════════════════════════════════════════════════
706 # Layer 8 — Atomic Write Safety (Phase 8 crash-safety fix)
707 # ═══════════════════════════════════════════════════════════════════════════════
708
709
710 class TestAtomicWrite:
711 """_write() uses mkstemp → fsync → os.replace — no partial files visible to readers."""
712
713 async def test_write_leaves_complete_file(self, tmp_path: Path) -> None:
714 b = _backend()
715 repo_root = _repo_root(tmp_path)
716 oid = long_id("5" * 64)
717 data = b"atomic content"
718 await b.put(oid, data, repo_root=repo_root)
719 path = b._path(oid, repo_root=repo_root)
720 assert path.exists()
721 assert path.read_bytes() == data
722
723 async def test_write_no_tmp_file_remains_after_success(self, tmp_path: Path) -> None:
724 b = _backend()
725 repo_root = _repo_root(tmp_path)
726 oid = long_id("6" * 64)
727 await b.put(oid, b"data", repo_root=repo_root)
728 objects_dir = repo_root / "objects"
729 tmp_files = list(objects_dir.rglob("tmp*"))
730 assert tmp_files == [], f"stray temp files: {tmp_files}"
731
732 async def test_write_file_is_immutable_after_put(self, tmp_path: Path) -> None:
733 import stat as _stat
734 b = _backend()
735 repo_root = _repo_root(tmp_path)
736 oid = long_id("7" * 64)
737 await b.put(oid, b"readonly data", repo_root=repo_root)
738 path = b._path(oid, repo_root=repo_root)
739 mode = _stat.S_IMODE(path.stat().st_mode)
740 assert mode == 0o444, f"expected 0o444 but got {oct(mode)}"
741
742 async def test_write_tmp_file_cleaned_up_on_fsync_failure(self, tmp_path: Path) -> None:
743 import os as _os
744 b = _backend()
745 repo_root = _repo_root(tmp_path)
746 oid = long_id("8" * 64)
747
748 def _failing_fsync(fd: int) -> None:
749 raise OSError("simulated fsync failure")
750
751 import unittest.mock as _mock
752 objects_dir = repo_root / "objects"
753 objects_dir.mkdir(parents=True, exist_ok=True)
754
755 with _mock.patch("os.fsync", side_effect=_failing_fsync):
756 with pytest.raises(OSError, match="simulated fsync failure"):
757 await b.put(oid, b"never written", repo_root=repo_root)
758
759 tmp_files = list(objects_dir.rglob("tmp*"))
760 assert tmp_files == [], f"orphan temp files after fsync failure: {tmp_files}"
761
762 async def test_write_idempotent_identical_bytes_skips_io(self, tmp_path: Path) -> None:
763 import time as _time
764 b = _backend()
765 repo_root = _repo_root(tmp_path)
766 oid = long_id("9" * 64)
767 data = b"idempotent"
768 await b.put(oid, data, repo_root=repo_root)
769 path = b._path(oid, repo_root=repo_root)
770 mtime_before = path.stat().st_mtime_ns
771 _time.sleep(0.01)
772 await b.put(oid, data, repo_root=repo_root)
773 mtime_after = path.stat().st_mtime_ns
774 assert mtime_before == mtime_after, "second put with identical bytes must not touch the file"
775
776 async def test_write_repairs_corrupted_file_atomically(self, tmp_path: Path) -> None:
777 import stat as _stat
778 b = _backend()
779 repo_root = _repo_root(tmp_path)
780 raw = b"correct content"
781 corrupt = b"wrong zlib garbage"
782 oid = long_id("aa" + "b" * 62)
783
784 path = b._path(oid, repo_root=repo_root)
785 path.parent.mkdir(parents=True, exist_ok=True)
786 path.write_bytes(corrupt)
787 path.chmod(_stat.S_IRUSR | _stat.S_IRGRP | _stat.S_IROTH)
788
789 await b.put(oid, raw, repo_root=repo_root)
790 assert await b.get(oid, repo_root=repo_root) == raw
791
792 async def test_write_concurrent_same_object_safe(self, tmp_path: Path) -> None:
793 import asyncio as _asyncio
794 b = _backend()
795 repo_root = _repo_root(tmp_path)
796 oid = long_id("bb" + "c" * 62)
797 data = b"concurrent data"
798 results = await _asyncio.gather(*[b.put(oid, data, repo_root=repo_root) for _ in range(10)])
799 assert all(r.startswith("local://") for r in results)
800 assert await b.get(oid, repo_root=repo_root) == data
801
802
803 # ═══════════════════════════════════════════════════════════════════════════════
804 # read_object_bytes — unified adapter
805 # ═══════════════════════════════════════════════════════════════════════════════
806
807
808 class TestUnitReadObjectBytes:
809 """read_object_bytes(obj) must handle every storage case without callers
810 knowing which backend is active."""
811
812 def _obj(self, **kwargs):
813 """Build a minimal MusehubObject-like namespace."""
814 defaults = dict(
815 object_id=long_id("aa" * 32),
816 content_cache=None,
817 disk_path="",
818 storage_uri="",
819 )
820 defaults.update(kwargs)
821 return SimpleNamespace(**defaults)
822
823 async def test_returns_content_cache_when_present(self) -> None:
824 from musehub.storage.backends import read_object_bytes
825 obj = self._obj(content_cache=b"cached")
826 assert await read_object_bytes(obj) == b"cached"
827
828 async def test_content_cache_takes_priority_over_disk(self, tmp_path: Path) -> None:
829 from musehub.storage.backends import read_object_bytes
830 disk_file = tmp_path / "obj"
831 disk_file.write_bytes(b"disk content")
832 obj = self._obj(content_cache=b"cached", disk_path=str(disk_file))
833 assert await read_object_bytes(obj) == b"cached"
834
835 async def test_reads_local_disk_path(self, tmp_path: Path) -> None:
836 from musehub.storage.backends import read_object_bytes
837 disk_file = tmp_path / "obj"
838 disk_file.write_bytes(b"from disk")
839 obj = self._obj(disk_path=str(disk_file))
840 assert await read_object_bytes(obj) == b"from disk"
841
842 async def test_reads_local_uri_prefix(self, tmp_path: Path) -> None:
843 from musehub.storage.backends import read_object_bytes
844 disk_file = tmp_path / "obj"
845 disk_file.write_bytes(b"from local uri")
846 obj = self._obj(disk_path=f"local://{disk_file}")
847 assert await read_object_bytes(obj) == b"from local uri"
848
849 async def test_reads_from_s3_when_s3_uri(self) -> None:
850 from musehub.storage.backends import read_object_bytes
851 obj = self._obj(disk_path="s3://my-bucket/objects/sha256_abc123")
852 mock_backend = MagicMock()
853
854 async def _fake_get(oid, **kw):
855 return b"s3 bytes"
856
857 mock_backend.get = _fake_get
858 with patch("musehub.storage.backends.get_backend", return_value=mock_backend):
859 result = await read_object_bytes(obj)
860 assert result == b"s3 bytes"
861
862 async def test_missing_disk_file_returns_none(self, tmp_path: Path) -> None:
863 from musehub.storage.backends import read_object_bytes
864 obj = self._obj(disk_path=str(tmp_path / "does_not_exist"))
865 assert await read_object_bytes(obj) is None
866
867 async def test_no_path_no_cache_returns_none(self) -> None:
868 from musehub.storage.backends import read_object_bytes
869 obj = self._obj()
870 assert await read_object_bytes(obj) is None
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 122 days ago