File size: 2,519 Bytes
08565c0 d0eee72 08565c0 d0eee72 08565c0 d0eee72 08565c0 d0eee72 08565c0 d0eee72 08565c0 d0eee72 08565c0 d0eee72 08565c0 d0eee72 08565c0 d0eee72 08565c0 d0eee72 08565c0 d0eee72 08565c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """FileStore tests.
On the Space this same store writes into a mounted bucket, so the properties
that matter are the filesystem ones: sharding, atomic replacement, and a clean
NotFound for anything absent.
"""
import json
import pytest
import storage
@pytest.fixture
def store(tmp_path):
return storage.FileStore(str(tmp_path / "traces"))
def test_roundtrip(store):
record = {"id": "abc", "payload": {"messages": []}}
store.put("abc", record)
assert store.get("abc") == record
def test_sharded_layout(store, tmp_path):
store.put("abcdef", {"id": "abcdef"})
assert (tmp_path / "traces" / "ab" / "abcdef.json").is_file()
def test_missing_raises_notfound(store):
with pytest.raises(storage.NotFound):
store.get("nope")
def test_unicode_survives(store):
store.put("abc", {"title": "hé ✓", "payload": "日本語"})
assert store.get("abc")["title"] == "hé ✓"
def test_no_temp_files_left_behind(store, tmp_path):
store.put("abc", {"id": "abc"})
assert not list((tmp_path / "traces").rglob("*.tmp"))
def test_write_is_atomic(store, tmp_path, monkeypatch):
"""A failed write must leave no partial file for a reader to find."""
real_replace = storage.os.replace
def boom(src, dst):
raise OSError("disk full")
monkeypatch.setattr(storage.os, "replace", boom)
with pytest.raises(OSError):
store.put("abc", {"id": "abc"})
monkeypatch.setattr(storage.os, "replace", real_replace)
with pytest.raises(storage.NotFound):
store.get("abc")
assert not list((tmp_path / "traces").rglob("*.tmp"))
def test_build_store_creates_dir(monkeypatch, tmp_path):
target = tmp_path / "mounted" / "traces"
monkeypatch.setenv("TRACE_DIR", str(target))
store = storage.build_store()
assert target.is_dir()
assert store.root == target
def test_build_store_rejects_readonly_dir(monkeypatch, tmp_path):
"""A bucket accidentally mounted read-only should fail at boot, not on the
first upload."""
target = tmp_path / "ro"
target.mkdir()
target.chmod(0o500)
monkeypatch.setenv("TRACE_DIR", str(target))
try:
with pytest.raises(RuntimeError, match="not writable"):
storage.build_store()
finally:
target.chmod(0o700)
def test_record_is_plain_json_on_disk(store, tmp_path):
store.put("abcdef", {"id": "abcdef", "title": "t"})
body = json.loads((tmp_path / "traces" / "ab" / "abcdef.json").read_text())
assert body["title"] == "t"
|