137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Tests for local-first storage with NAS fallback (umbraaudio.storage)
|
|
and the prune planner in scripts/nas_lifecycle."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
|
|
|
import nas_lifecycle # noqa: E402
|
|
from umbraaudio import storage as st # noqa: E402
|
|
|
|
|
|
@pytest.fixture
|
|
def stores(tmp_path):
|
|
local = tmp_path / "local" / "UmbraAudio"
|
|
nas = tmp_path / "nas" / "UmbraAudio"
|
|
for root in (local, nas):
|
|
root.mkdir(parents=True)
|
|
return local, nas
|
|
|
|
|
|
def _mirror(local: Path, nas: Path, rel: str) -> None:
|
|
src = local / rel
|
|
dst = nas / rel
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src, dst)
|
|
|
|
|
|
class TestResolvePath:
|
|
def test_local_hit(self, stores):
|
|
local, nas = stores
|
|
f = local / "a.txt"
|
|
f.write_text("x")
|
|
assert st.resolve_path(f, local_root=local, nas_root=nas) == f
|
|
|
|
def test_relative_resolves_against_local_root(self, stores):
|
|
local, nas = stores
|
|
(local / "b.txt").write_text("y")
|
|
out = st.resolve_path("b.txt", local_root=local, nas_root=nas)
|
|
assert out == local / "b.txt"
|
|
|
|
def test_fallback_restores_from_nas(self, stores):
|
|
local, nas = stores
|
|
(nas / "models").mkdir()
|
|
(nas / "models" / "big.bin").write_bytes(b"0" * 128)
|
|
out = st.resolve_path("models/big.bin", local_root=local, nas_root=nas)
|
|
assert out.read_bytes() == b"0" * 128
|
|
assert out.exists()
|
|
|
|
def test_missing_everywhere_raises(self, stores):
|
|
local, nas = stores
|
|
with pytest.raises(FileNotFoundError):
|
|
st.resolve_path("nope.txt", local_root=local, nas_root=nas)
|
|
|
|
def test_restore_false_skips_fallback(self, stores):
|
|
local, nas = stores
|
|
out = st.resolve_path("gone.txt", local_root=local,
|
|
nas_root=nas, restore=False)
|
|
assert out == local / "gone.txt"
|
|
assert not out.exists()
|
|
|
|
|
|
def _age_file(path: Path, days: int) -> None:
|
|
import os
|
|
old = __import__("time").time() - days * 86400
|
|
os.utime(path, (old, old))
|
|
|
|
|
|
class TestPrunePlan:
|
|
def _seed(self, local: Path, nas: Path) -> None:
|
|
(local / "output").mkdir(parents=True)
|
|
(local / "scripts").mkdir()
|
|
for rel in ("output/old.wav", "output/new.wav", "scripts/gen.py"):
|
|
f = local / rel
|
|
f.write_text(rel)
|
|
if "old" in rel or rel.endswith("gen.py"):
|
|
_age_file(f, 90)
|
|
|
|
def test_only_old_unprotected_verified_files_pruned(self, stores):
|
|
local, nas = stores
|
|
self._seed(local, nas)
|
|
_mirror(local, nas, "output/old.wav")
|
|
plan = nas_lifecycle.prune_plan(local, nas,
|
|
nas_lifecycle.DEFAULT_POLICY)
|
|
assert plan == [local / "output" / "old.wav"]
|
|
|
|
def test_unverified_files_never_pruned(self, stores):
|
|
local, nas = stores
|
|
self._seed(local, nas)
|
|
assert nas_lifecycle.prune_plan(local, nas,
|
|
nas_lifecycle.DEFAULT_POLICY) == []
|
|
|
|
def test_emergency_includes_protected_dirs(self, stores):
|
|
local, nas = stores
|
|
self._seed(local, nas)
|
|
_mirror(local, nas, "scripts/gen.py")
|
|
plan = nas_lifecycle.prune_plan(local, nas,
|
|
nas_lifecycle.DEFAULT_POLICY,
|
|
emergency=True)
|
|
assert local / "scripts" / "gen.py" in plan
|
|
|
|
def test_recent_files_excluded_even_if_backed_up(self, stores):
|
|
local, nas = stores
|
|
self._seed(local, nas)
|
|
_mirror(local, nas, "output/new.wav")
|
|
assert local / "output/new.wav" not in nas_lifecycle.prune_plan(
|
|
local, nas, nas_lifecycle.DEFAULT_POLICY)
|
|
|
|
|
|
class TestVerifiedOnNas:
|
|
def test_size_mismatch_rejects(self, stores):
|
|
local, nas = stores
|
|
f = local / "x.bin"
|
|
f.write_bytes(b"12345")
|
|
m = nas / "x.bin"
|
|
m.write_bytes(b"12")
|
|
_age_file(f, 60)
|
|
assert not nas_lifecycle.verified_on_nas(f, local, nas)
|
|
|
|
def test_match_accepts(self, stores):
|
|
local, nas = stores
|
|
f = local / "x.bin"
|
|
f.write_bytes(b"12345")
|
|
_mirror(local, nas, "x.bin")
|
|
_age_file(f, 60)
|
|
_age_file(nas / "x.bin", 60)
|
|
assert nas_lifecycle.verified_on_nas(f, local, nas)
|