201 lines
7.1 KiB
Python
201 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import threading
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from game_data_capture.storage.blobs import (
|
|
BlobCorruptionError,
|
|
BlobLimitError,
|
|
BlobLockError,
|
|
ContentAddressedBlobStore,
|
|
)
|
|
|
|
|
|
def test_blob_is_verified_deduplicated_and_private(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
payload = b"capture-bytes"
|
|
digest, size = store.put([payload], max_bytes=100)
|
|
assert digest == hashlib.sha256(payload).hexdigest()
|
|
assert store.put([payload], max_bytes=100) == (digest, size)
|
|
assert b"".join(store.read(digest, max_bytes=100)) == payload
|
|
assert (store._path(digest).stat().st_mode & 0o777) == 0o600
|
|
|
|
|
|
def test_corruption_and_partial_files_are_detected_or_reconciled(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
digest, _ = store.put([b"good"], max_bytes=100)
|
|
store._path(digest).write_bytes(b"bad")
|
|
with pytest.raises(BlobCorruptionError):
|
|
store.verify(digest)
|
|
store.temporary.mkdir(parents=True, exist_ok=True)
|
|
partial = store.temporary / ".crash.partial"
|
|
partial.write_bytes(b"partial")
|
|
assert store.reconcile() == [".crash.partial"]
|
|
assert not partial.exists()
|
|
|
|
|
|
def test_rejects_hash_and_generator_failures_without_partial_file(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
with pytest.raises(BlobCorruptionError):
|
|
store.put([b"value"], max_bytes=100, expected_sha256="0" * 64)
|
|
|
|
def broken() -> Iterator[bytes]:
|
|
yield b"prefix"
|
|
raise RuntimeError("source failed")
|
|
|
|
with pytest.raises(RuntimeError):
|
|
store.put(broken(), max_bytes=100)
|
|
assert not list(store.temporary.glob("*.partial"))
|
|
|
|
|
|
def test_root_symlinks_are_not_traversed(tmp_path: Path) -> None:
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
root = (tmp_path / "data").resolve()
|
|
root.symlink_to(outside, target_is_directory=True)
|
|
store = ContentAddressedBlobStore(root)
|
|
with pytest.raises(BlobCorruptionError):
|
|
store.put([b"x"], max_bytes=1)
|
|
|
|
|
|
def test_limits_and_root_symlink_reconciliation_are_safe(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
for limit in (0, -1, True):
|
|
with pytest.raises(BlobLimitError):
|
|
store.put([b"x"], max_bytes=limit)
|
|
digest, _ = store.put([b"x"], max_bytes=1)
|
|
for limit in (0, -1, True):
|
|
with pytest.raises(BlobLimitError):
|
|
list(store.read(digest, max_bytes=limit))
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
partial = outside / ".outside.partial"
|
|
partial.write_bytes(b"outside")
|
|
store.temporary.rmdir()
|
|
store.temporary.symlink_to(outside, target_is_directory=True)
|
|
assert store.reconcile() == []
|
|
assert partial.read_bytes() == b"outside"
|
|
|
|
|
|
def test_corrupt_read_rejects_before_yielding(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
digest, _ = store.put([b"good"], max_bytes=10)
|
|
store._path(digest).write_bytes(b"bad")
|
|
with pytest.raises(BlobCorruptionError):
|
|
list(store.read(digest, max_bytes=10))
|
|
|
|
|
|
def test_oversize_and_valid_digest_mismatch_leave_no_partial(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
with pytest.raises(BlobLimitError):
|
|
store.put([b"too-large"], max_bytes=1)
|
|
assert not list(store.temporary.glob("*.partial"))
|
|
with pytest.raises(BlobCorruptionError):
|
|
store.put([b"value"], max_bytes=100, expected_sha256="0" * 64)
|
|
assert not list(store.temporary.glob("*.partial"))
|
|
|
|
|
|
def test_blob_root_symlink_purge_does_not_inspect_outside(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
store.put([b"inside"], max_bytes=100)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
outside_file = outside / "outside-blob"
|
|
outside_file.write_bytes(b"outside")
|
|
for entry in store.blobs.iterdir():
|
|
if entry.is_file():
|
|
entry.unlink()
|
|
else:
|
|
for child in entry.iterdir():
|
|
child.unlink()
|
|
entry.rmdir()
|
|
store.blobs.rmdir()
|
|
store.blobs.symlink_to(outside, target_is_directory=True)
|
|
assert store.purge_dry_run(set()) == []
|
|
assert outside_file.read_bytes() == b"outside"
|
|
|
|
|
|
def test_purge_dry_run_does_not_follow_symlink(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
digest, _ = store.put([b"inside"], max_bytes=100)
|
|
outside = tmp_path / "outside"
|
|
outside.write_bytes(b"outside")
|
|
unsafe = store.blobs / "ff"
|
|
unsafe.symlink_to(tmp_path)
|
|
assert store.purge_dry_run(set()) == [store._path(digest).relative_to(store.root)]
|
|
assert outside.read_bytes() == b"outside"
|
|
|
|
|
|
def test_operation_lock_rejects_symlink_and_unsafe_mode(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
store.root.mkdir(mode=0o700)
|
|
outside = tmp_path / "outside"
|
|
outside.write_bytes(b"x")
|
|
store.lock_path.symlink_to(outside)
|
|
with pytest.raises(BlobLockError), store.operation_lock():
|
|
pass
|
|
store.lock_path.unlink()
|
|
store.lock_path.write_bytes(b"")
|
|
store.lock_path.chmod(0o644)
|
|
with pytest.raises(BlobLockError), store.operation_lock():
|
|
pass
|
|
|
|
|
|
def test_operation_lock_timeout_and_release_after_exception(tmp_path: Path) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
entered = threading.Event()
|
|
outcome: list[type[Exception]] = []
|
|
|
|
def contender() -> None:
|
|
entered.set()
|
|
try:
|
|
with store.operation_lock(timeout_seconds=0.02):
|
|
pass
|
|
except BlobLockError as error:
|
|
outcome.append(type(error))
|
|
|
|
with store.operation_lock():
|
|
thread = threading.Thread(target=contender)
|
|
thread.start()
|
|
assert entered.wait(0.2)
|
|
thread.join(0.5)
|
|
assert not thread.is_alive()
|
|
assert outcome == [BlobLockError]
|
|
with pytest.raises(RuntimeError), store.operation_lock():
|
|
raise RuntimeError("fault")
|
|
with store.operation_lock(timeout_seconds=0.02):
|
|
pass
|
|
for invalid in (True, "1", 0, 6):
|
|
with (
|
|
pytest.raises(ValueError),
|
|
store.operation_lock(
|
|
timeout_seconds=invalid # type: ignore[arg-type]
|
|
),
|
|
):
|
|
pass
|
|
|
|
|
|
def test_expected_size_mismatch_rejects_before_blob_read(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
store = ContentAddressedBlobStore((tmp_path / "data").resolve())
|
|
digest, _ = store.put([b"value"], max_bytes=100)
|
|
store._path(digest).write_bytes(b"x" * 4096)
|
|
reads = 0
|
|
original = os.read
|
|
|
|
def forbidden_read(fd: int, size: int) -> bytes:
|
|
nonlocal reads
|
|
reads += 1
|
|
return original(fd, size)
|
|
|
|
monkeypatch.setattr(os, "read", forbidden_read)
|
|
with pytest.raises(BlobCorruptionError):
|
|
store.verify(digest, expected_size=5)
|
|
assert reads == 0
|