104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
# #region Test.Storage.Service [C:3] [TYPE Module] [SEMANTICS test,storage,io,security]
|
|
# @BRIEF Verify StorageService filesystem operations and path-traversal protection.
|
|
# @RELATION VERIFIES -> [Storage.Service]
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.models.storage import FileCategory
|
|
from src.services.storage_service import StorageService
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_root():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
yield Path(td)
|
|
|
|
|
|
@pytest.fixture
|
|
def svc(tmp_root: Path) -> StorageService:
|
|
return StorageService(root=tmp_root)
|
|
|
|
|
|
class TestSaveFromPath:
|
|
def test_saves_and_returns_path(self, svc: StorageService, tmp_root: Path):
|
|
src = tmp_root / "source.txt"
|
|
src.write_text("hello")
|
|
dest = svc.save_from_path(FileCategory.BACKUP, "env/file.zip", src)
|
|
assert dest.exists()
|
|
assert dest.read_bytes() == b"hello"
|
|
|
|
def test_creates_intermediate_dirs(self, svc: StorageService, tmp_root: Path):
|
|
src = tmp_root / "data.bin"
|
|
src.write_bytes(b"abc")
|
|
dest = svc.save_from_path(FileCategory.MIGRATION_FAILED, "a/b/c/archive.zip", src)
|
|
assert dest.parent.exists()
|
|
|
|
|
|
class TestSaveBytes:
|
|
def test_writes_content(self, svc: StorageService, tmp_root: Path):
|
|
dest = svc.save_bytes(FileCategory.CHAT_UPLOADS, "img.png", b"\x89PNG")
|
|
assert dest.read_bytes() == b"\x89PNG"
|
|
|
|
|
|
class TestGetPath:
|
|
def test_returns_valid_path(self, svc: StorageService, tmp_root: Path):
|
|
p = svc.get_path(FileCategory.REPOSITORY, "git/repo")
|
|
assert str(p).startswith(str(tmp_root))
|
|
assert "repositories/git/repo" in str(p)
|
|
|
|
def test_empty_rel_path_returns_category_root(self, svc: StorageService, tmp_root: Path):
|
|
p = svc.get_path(FileCategory.BACKUP)
|
|
assert p == tmp_root / "backups"
|
|
|
|
@pytest.mark.parametrize("bad_path", [
|
|
"../../../etc/passwd",
|
|
"../secrets",
|
|
"foo/../../etc/shadow",
|
|
])
|
|
def test_blocks_path_traversal(self, svc: StorageService, bad_path: str):
|
|
with pytest.raises(ValueError, match="Path traversal"):
|
|
svc.get_path(FileCategory.BACKUP, bad_path)
|
|
|
|
|
|
class TestDelete:
|
|
def test_deletes_file(self, svc: StorageService, tmp_root: Path):
|
|
src = tmp_root / "todel.txt"
|
|
src.write_text("x")
|
|
svc.save_from_path(FileCategory.BACKUP, "todel.txt", src)
|
|
assert svc.delete(FileCategory.BACKUP, "todel.txt")
|
|
assert not (tmp_root / "backups" / "todel.txt").exists()
|
|
|
|
def test_delete_missing_is_noop(self, svc: StorageService):
|
|
assert not svc.delete(FileCategory.BACKUP, "ghost/dir")
|
|
|
|
def test_deletes_directory_tree(self, svc: StorageService, tmp_root: Path):
|
|
d = tmp_root / "backups" / "nested" / "deep"
|
|
d.mkdir(parents=True)
|
|
(d / "f.txt").write_text("ok")
|
|
assert svc.delete(FileCategory.BACKUP, "nested")
|
|
assert not (tmp_root / "backups" / "nested").exists()
|
|
|
|
|
|
class TestList:
|
|
def test_lists_files(self, svc: StorageService, tmp_root: Path):
|
|
(tmp_root / "backups").mkdir(parents=True, exist_ok=True)
|
|
(tmp_root / "backups" / "a.zip").write_text("a")
|
|
(tmp_root / "backups" / "b.zip").write_text("b")
|
|
items = svc.list(FileCategory.BACKUP)
|
|
names = [p.name for p in items]
|
|
assert names == ["a.zip", "b.zip"]
|
|
|
|
def test_empty_when_missing(self, svc: StorageService):
|
|
assert svc.list(FileCategory.MIGRATION_FAILED) == []
|
|
|
|
|
|
class TestEnsure:
|
|
def test_creates_category_root(self, svc: StorageService, tmp_root: Path):
|
|
d = svc.ensure_category(FileCategory.MIGRATION_FAILED)
|
|
assert d.exists()
|
|
assert d.is_dir()
|
|
# #endregion Test.Storage.Service
|