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