| """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" |
|
|
