Agent Zero
a3a26d1f66
B-STOR: storage.py um 5 Funktionen erweitert - validate_mime(): MIME-Erkennung (python-magic/mimetypes) + Allowlist-Prüfung - validate_size(): Dateigrößen-Prüfung (Default 50MB, konfigurierbar) - compute_hash(): SHA256/MD5/SHA1 Hash-Berechnung - save_with_metadata(): save + validate + hash in einem Call - get_file_metadata(): File-Stat ohne Content zu laden - config.py: storage_max_file_size_mb, storage_allowed_mimes Settings - Backward compatible: save/read/delete/exists unverändert B-STOR-TEST: 27 Tests in test_storage.py — alle grün - Path-Traversal, MIME-Validation, Size-Limits, Hashing, save_with_metadata, LocalStorage, Factory B-STOR-MIG: Bereits erledigt — DMS, Mail, Report-Generator, Attachments nutzen bereits get_storage_backend()
287 lines
9.6 KiB
Python
287 lines
9.6 KiB
Python
"""Tests for app/core/storage.py — validation, hashing, metadata, LocalStorage, factory."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
# Set test env BEFORE importing app modules
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-with-at-least-32-characters-for-testing-only!!")
|
|
os.environ.setdefault("ENVIRONMENT", "testing")
|
|
os.environ.setdefault("SESSION_COOKIE_SECURE", "false")
|
|
|
|
from app.core.storage import ( # noqa: E402
|
|
LocalStorage,
|
|
compute_hash,
|
|
get_file_metadata,
|
|
get_storage_backend,
|
|
reset_storage_backend,
|
|
save_with_metadata,
|
|
validate_mime,
|
|
validate_size,
|
|
)
|
|
|
|
|
|
# ─── Fixtures ───
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_storage(tmp_path, monkeypatch):
|
|
"""Provide a LocalStorage backed by a temp directory."""
|
|
monkeypatch.setenv("STORAGE_PATH", str(tmp_path))
|
|
monkeypatch.setenv("STORAGE_BACKEND", "local")
|
|
reset_storage_backend()
|
|
backend = LocalStorage(base_path=str(tmp_path))
|
|
yield backend
|
|
reset_storage_backend()
|
|
|
|
|
|
# ─── 1. Path-Traversal Tests ───
|
|
|
|
|
|
class TestPathTraversal:
|
|
@pytest.mark.asyncio
|
|
async def test_save_traversal_relative_parent(self, tmp_storage):
|
|
with pytest.raises(ValueError, match="[Pp]ath traversal"):
|
|
await tmp_storage.save("../etc/passwd", b"data")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_traversal_double_parent(self, tmp_storage):
|
|
with pytest.raises(ValueError, match="[Pp]ath traversal"):
|
|
await tmp_storage.save("../../secret", b"data")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_normal_path_ok(self, tmp_storage):
|
|
result = await tmp_storage.save("normal/path/file.txt", b"hello")
|
|
assert result == "normal/path/file.txt"
|
|
|
|
|
|
# ─── 2. MIME Tests ───
|
|
|
|
|
|
class TestValidateMime:
|
|
def test_text_plain_allowed(self):
|
|
mime = validate_mime("file.txt", b"hello", ["text/plain"])
|
|
assert mime == "text/plain"
|
|
|
|
def test_exe_rejected(self):
|
|
with pytest.raises(ValueError, match="not allowed"):
|
|
validate_mime("file.exe", b"MZ\x90\x00", ["text/plain"])
|
|
|
|
def test_pdf_default_allowlist(self):
|
|
# PDF magic bytes
|
|
mime = validate_mime("file.pdf", b"%PDF-1.4 test", None)
|
|
assert mime is not None
|
|
|
|
def test_empty_allowlist_all_allowed(self):
|
|
mime = validate_mime("file.xyz", b"data", [])
|
|
assert mime is not None
|
|
|
|
def test_custom_allowlist_reject(self):
|
|
with pytest.raises(ValueError, match="not allowed"):
|
|
validate_mime("file.txt", b"hello", ["application/pdf"])
|
|
|
|
|
|
# ─── 3. Size-Limit Tests ───
|
|
|
|
|
|
class TestValidateSize:
|
|
def test_small_file_ok(self):
|
|
validate_size(b"x" * 100, max_size_mb=1)
|
|
|
|
def test_oversized_file_raises(self):
|
|
with pytest.raises(ValueError, match="exceeds limit"):
|
|
validate_size(b"x" * (60 * 1024 * 1024), max_size_mb=50)
|
|
|
|
def test_exact_limit_ok(self):
|
|
# Exactly at the limit should pass (<=)
|
|
validate_size(b"x" * (1024 * 1024), max_size_mb=1)
|
|
|
|
def test_default_from_config(self):
|
|
# Should use config default (50 MB) — small data passes
|
|
validate_size(b"x" * 100)
|
|
|
|
|
|
# ─── 4. Hash Tests ───
|
|
|
|
|
|
class TestComputeHash:
|
|
def test_sha256_known_value(self):
|
|
result = compute_hash(b"hello")
|
|
expected = hashlib.sha256(b"hello").hexdigest()
|
|
assert result == expected
|
|
assert result == "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
|
|
|
|
def test_md5_known_value(self):
|
|
result = compute_hash(b"hello", algorithm="md5")
|
|
expected = hashlib.md5(b"hello").hexdigest()
|
|
assert result == expected
|
|
assert result == "5d41402abc4b2a76b9719d911017c592"
|
|
|
|
def test_empty_data(self):
|
|
result = compute_hash(b"")
|
|
assert result == hashlib.sha256(b"").hexdigest()
|
|
|
|
def test_sha1(self):
|
|
result = compute_hash(b"test", algorithm="sha1")
|
|
assert result == hashlib.sha1(b"test").hexdigest()
|
|
|
|
|
|
# ─── 5. save_with_metadata Tests ───
|
|
|
|
|
|
class TestSaveWithMetadata:
|
|
@pytest.mark.asyncio
|
|
async def test_save_returns_full_metadata(self, tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
|
|
monkeypatch.setenv("STORAGE_BACKEND", "local")
|
|
reset_storage_backend()
|
|
# Patch the singleton to use our tmp backend
|
|
import app.core.storage as storage_mod
|
|
storage_mod._storage_backend = tmp_storage
|
|
|
|
data = b"hello world"
|
|
result = await save_with_metadata("test/file.txt", data)
|
|
|
|
assert result["path"] == "test/file.txt"
|
|
assert result["size"] == len(data)
|
|
assert result["hash"] == compute_hash(data)
|
|
assert "mime_type" in result
|
|
assert result["storage_path"] == "test/file.txt"
|
|
reset_storage_backend()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_oversized_raises(self, tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
|
|
monkeypatch.setenv("STORAGE_BACKEND", "local")
|
|
reset_storage_backend()
|
|
import app.core.storage as storage_mod
|
|
storage_mod._storage_backend = tmp_storage
|
|
|
|
big_data = b"x" * (60 * 1024 * 1024)
|
|
with pytest.raises(ValueError, match="exceeds limit"):
|
|
await save_with_metadata("big.txt", big_data, max_size_mb=50)
|
|
reset_storage_backend()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_invalid_mime_raises(self, tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
|
|
monkeypatch.setenv("STORAGE_BACKEND", "local")
|
|
reset_storage_backend()
|
|
import app.core.storage as storage_mod
|
|
storage_mod._storage_backend = tmp_storage
|
|
|
|
with pytest.raises(ValueError, match="not allowed"):
|
|
await save_with_metadata("file.xyz", b"data", allowed_mimes=["application/pdf"])
|
|
reset_storage_backend()
|
|
|
|
|
|
# ─── 6. get_file_metadata Tests ───
|
|
|
|
|
|
class TestGetFileMetadata:
|
|
def test_existing_file(self, tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
|
|
monkeypatch.setenv("STORAGE_BACKEND", "local")
|
|
reset_storage_backend()
|
|
import app.core.storage as storage_mod
|
|
storage_mod._storage_backend = tmp_storage
|
|
|
|
# Save a file first
|
|
asyncio.run(tmp_storage.save("meta/test.txt", b"metadata test"))
|
|
|
|
meta = get_file_metadata("meta/test.txt")
|
|
assert meta["exists"] is True
|
|
assert meta["size"] == len(b"metadata test")
|
|
assert meta["modified"] is not None
|
|
reset_storage_backend()
|
|
|
|
def test_non_existing_file(self, tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
|
|
monkeypatch.setenv("STORAGE_BACKEND", "local")
|
|
reset_storage_backend()
|
|
import app.core.storage as storage_mod
|
|
storage_mod._storage_backend = tmp_storage
|
|
|
|
meta = get_file_metadata("nonexistent/file.txt")
|
|
assert meta["exists"] is False
|
|
assert meta["size"] is None
|
|
assert meta["modified"] is None
|
|
reset_storage_backend()
|
|
|
|
|
|
# ─── 7. LocalStorage Tests ───
|
|
|
|
|
|
class TestLocalStorage:
|
|
@pytest.mark.asyncio
|
|
async def test_save_read_delete_cycle(self, tmp_storage):
|
|
data = b"cycle test data"
|
|
path = await tmp_storage.save("cycle/test.txt", data)
|
|
assert path == "cycle/test.txt"
|
|
|
|
read_data = await tmp_storage.read("cycle/test.txt")
|
|
assert read_data == data
|
|
|
|
deleted = await tmp_storage.delete("cycle/test.txt")
|
|
assert deleted is True
|
|
|
|
deleted_again = await tmp_storage.delete("cycle/test.txt")
|
|
assert deleted_again is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_stream_read_cycle(self, tmp_storage):
|
|
async def chunk_gen():
|
|
yield b"chunk1-"
|
|
yield b"chunk2-"
|
|
yield b"chunk3"
|
|
|
|
total = await tmp_storage.save_stream("stream/test.bin", chunk_gen())
|
|
assert total == len(b"chunk1-chunk2-chunk3")
|
|
|
|
read_data = await tmp_storage.read("stream/test.bin")
|
|
assert read_data == b"chunk1-chunk2-chunk3"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exists_check(self, tmp_storage):
|
|
assert await tmp_storage.exists("exists/no.txt") is False
|
|
await tmp_storage.save("exists/yes.txt", b"yes")
|
|
assert await tmp_storage.exists("exists/yes.txt") is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_files(self, tmp_storage):
|
|
await tmp_storage.save("list/a.txt", b"a")
|
|
await tmp_storage.save("list/sub/b.txt", b"b")
|
|
files = await tmp_storage.list_files("list")
|
|
assert len(files) == 2
|
|
# Paths are relative to base_path
|
|
assert any("a.txt" in f for f in files)
|
|
assert any("b.txt" in f for f in files)
|
|
|
|
|
|
# ─── 8. Backend Factory Tests ───
|
|
|
|
|
|
class TestBackendFactory:
|
|
def test_get_storage_backend_local_default(self, monkeypatch):
|
|
monkeypatch.setenv("STORAGE_BACKEND", "local")
|
|
monkeypatch.setenv("STORAGE_PATH", tempfile.mkdtemp())
|
|
reset_storage_backend()
|
|
backend = get_storage_backend()
|
|
assert isinstance(backend, LocalStorage)
|
|
reset_storage_backend()
|
|
|
|
def test_reset_storage_backend(self, monkeypatch):
|
|
monkeypatch.setenv("STORAGE_BACKEND", "local")
|
|
monkeypatch.setenv("STORAGE_PATH", tempfile.mkdtemp())
|
|
reset_storage_backend()
|
|
backend1 = get_storage_backend()
|
|
reset_storage_backend()
|
|
backend2 = get_storage_backend()
|
|
assert backend1 is not backend2
|
|
reset_storage_backend()
|