This commit is contained in:
Yaojia Wang
2026-02-01 00:08:40 +01:00
parent 33ada0350d
commit a516de4320
90 changed files with 11642 additions and 398 deletions

View File

@@ -0,0 +1,712 @@
"""
Tests for LocalStorageBackend.
TDD Phase 1: RED - Write tests first, then implement to pass.
"""
import shutil
import tempfile
from pathlib import Path
import pytest
@pytest.fixture
def temp_storage_dir() -> Path:
"""Create a temporary directory for storage tests."""
temp_dir = Path(tempfile.mkdtemp())
yield temp_dir
shutil.rmtree(temp_dir, ignore_errors=True)
@pytest.fixture
def sample_file(temp_storage_dir: Path) -> Path:
"""Create a sample file for testing."""
file_path = temp_storage_dir / "sample.txt"
file_path.write_text("Hello, World!")
return file_path
@pytest.fixture
def sample_image(temp_storage_dir: Path) -> Path:
"""Create a sample PNG file for testing."""
file_path = temp_storage_dir / "sample.png"
# Minimal valid PNG (1x1 transparent pixel)
png_data = bytes(
[
0x89,
0x50,
0x4E,
0x47,
0x0D,
0x0A,
0x1A,
0x0A, # PNG signature
0x00,
0x00,
0x00,
0x0D, # IHDR length
0x49,
0x48,
0x44,
0x52, # IHDR
0x00,
0x00,
0x00,
0x01, # width: 1
0x00,
0x00,
0x00,
0x01, # height: 1
0x08,
0x06,
0x00,
0x00,
0x00, # 8-bit RGBA
0x1F,
0x15,
0xC4,
0x89, # CRC
0x00,
0x00,
0x00,
0x0A, # IDAT length
0x49,
0x44,
0x41,
0x54, # IDAT
0x78,
0x9C,
0x63,
0x00,
0x01,
0x00,
0x00,
0x05,
0x00,
0x01, # compressed data
0x0D,
0x0A,
0x2D,
0xB4, # CRC
0x00,
0x00,
0x00,
0x00, # IEND length
0x49,
0x45,
0x4E,
0x44, # IEND
0xAE,
0x42,
0x60,
0x82, # CRC
]
)
file_path.write_bytes(png_data)
return file_path
class TestLocalStorageBackendCreation:
"""Tests for LocalStorageBackend instantiation."""
def test_create_with_base_path(self, temp_storage_dir: Path) -> None:
"""Test creating backend with base path."""
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
assert backend.base_path == temp_storage_dir
def test_create_with_string_path(self, temp_storage_dir: Path) -> None:
"""Test creating backend with string path."""
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=str(temp_storage_dir))
assert backend.base_path == temp_storage_dir
def test_create_creates_directory_if_not_exists(
self, temp_storage_dir: Path
) -> None:
"""Test that base directory is created if it doesn't exist."""
from shared.storage.local import LocalStorageBackend
new_dir = temp_storage_dir / "new_storage"
assert not new_dir.exists()
backend = LocalStorageBackend(base_path=new_dir)
assert new_dir.exists()
assert backend.base_path == new_dir
def test_is_storage_backend_subclass(self, temp_storage_dir: Path) -> None:
"""Test that LocalStorageBackend is a StorageBackend."""
from shared.storage.base import StorageBackend
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
assert isinstance(backend, StorageBackend)
class TestLocalStorageBackendUpload:
"""Tests for LocalStorageBackend.upload method."""
def test_upload_file(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test uploading a file."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
result = backend.upload(sample_file, "uploads/sample.txt")
assert result == "uploads/sample.txt"
assert (storage_dir / "uploads" / "sample.txt").exists()
assert (storage_dir / "uploads" / "sample.txt").read_text() == "Hello, World!"
def test_upload_creates_subdirectories(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that upload creates necessary subdirectories."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
result = backend.upload(sample_file, "deep/nested/path/sample.txt")
assert (storage_dir / "deep" / "nested" / "path" / "sample.txt").exists()
def test_upload_fails_if_file_exists_without_overwrite(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that upload fails if file exists and overwrite is False."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
# First upload succeeds
backend.upload(sample_file, "sample.txt")
# Second upload should fail
with pytest.raises(StorageError, match="already exists"):
backend.upload(sample_file, "sample.txt", overwrite=False)
def test_upload_succeeds_with_overwrite(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that upload succeeds with overwrite=True."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
# First upload
backend.upload(sample_file, "sample.txt")
# Modify original file
sample_file.write_text("Modified content")
# Second upload with overwrite
result = backend.upload(sample_file, "sample.txt", overwrite=True)
assert result == "sample.txt"
assert (storage_dir / "sample.txt").read_text() == "Modified content"
def test_upload_nonexistent_file_fails(self, temp_storage_dir: Path) -> None:
"""Test that uploading nonexistent file fails."""
from shared.storage.base import FileNotFoundStorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(FileNotFoundStorageError):
backend.upload(Path("/nonexistent/file.txt"), "sample.txt")
def test_upload_binary_file(
self, temp_storage_dir: Path, sample_image: Path
) -> None:
"""Test uploading a binary file."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
result = backend.upload(sample_image, "images/sample.png")
assert result == "images/sample.png"
uploaded_content = (storage_dir / "images" / "sample.png").read_bytes()
assert uploaded_content == sample_image.read_bytes()
class TestLocalStorageBackendDownload:
"""Tests for LocalStorageBackend.download method."""
def test_download_file(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test downloading a file."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
download_dir = temp_storage_dir / "downloads"
download_dir.mkdir()
backend = LocalStorageBackend(base_path=storage_dir)
# First upload
backend.upload(sample_file, "sample.txt")
# Then download
local_path = download_dir / "downloaded.txt"
result = backend.download("sample.txt", local_path)
assert result == local_path
assert local_path.exists()
assert local_path.read_text() == "Hello, World!"
def test_download_creates_parent_directories(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that download creates parent directories."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "sample.txt")
local_path = temp_storage_dir / "deep" / "nested" / "downloaded.txt"
result = backend.download("sample.txt", local_path)
assert local_path.exists()
assert local_path.read_text() == "Hello, World!"
def test_download_nonexistent_file_fails(self, temp_storage_dir: Path) -> None:
"""Test that downloading nonexistent file fails."""
from shared.storage.base import FileNotFoundStorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(FileNotFoundStorageError, match="nonexistent.txt"):
backend.download("nonexistent.txt", Path("/tmp/file.txt"))
def test_download_nested_file(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test downloading a file from nested path."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "a/b/c/sample.txt")
local_path = temp_storage_dir / "downloaded.txt"
result = backend.download("a/b/c/sample.txt", local_path)
assert local_path.read_text() == "Hello, World!"
class TestLocalStorageBackendExists:
"""Tests for LocalStorageBackend.exists method."""
def test_exists_returns_true_for_existing_file(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test exists returns True for existing file."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "sample.txt")
assert backend.exists("sample.txt") is True
def test_exists_returns_false_for_nonexistent_file(
self, temp_storage_dir: Path
) -> None:
"""Test exists returns False for nonexistent file."""
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
assert backend.exists("nonexistent.txt") is False
def test_exists_with_nested_path(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test exists with nested path."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "a/b/sample.txt")
assert backend.exists("a/b/sample.txt") is True
assert backend.exists("a/b/other.txt") is False
class TestLocalStorageBackendListFiles:
"""Tests for LocalStorageBackend.list_files method."""
def test_list_files_empty_storage(self, temp_storage_dir: Path) -> None:
"""Test listing files in empty storage."""
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
assert backend.list_files("") == []
def test_list_files_returns_all_files(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test listing all files."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
# Upload multiple files
backend.upload(sample_file, "file1.txt")
backend.upload(sample_file, "file2.txt")
backend.upload(sample_file, "subdir/file3.txt")
files = backend.list_files("")
assert len(files) == 3
assert "file1.txt" in files
assert "file2.txt" in files
assert "subdir/file3.txt" in files
def test_list_files_with_prefix(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test listing files with prefix filter."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "images/a.png")
backend.upload(sample_file, "images/b.png")
backend.upload(sample_file, "labels/a.txt")
files = backend.list_files("images/")
assert len(files) == 2
assert "images/a.png" in files
assert "images/b.png" in files
assert "labels/a.txt" not in files
def test_list_files_returns_sorted(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that list_files returns sorted list."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "c.txt")
backend.upload(sample_file, "a.txt")
backend.upload(sample_file, "b.txt")
files = backend.list_files("")
assert files == ["a.txt", "b.txt", "c.txt"]
class TestLocalStorageBackendDelete:
"""Tests for LocalStorageBackend.delete method."""
def test_delete_existing_file(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test deleting an existing file."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "sample.txt")
result = backend.delete("sample.txt")
assert result is True
assert not (storage_dir / "sample.txt").exists()
def test_delete_nonexistent_file_returns_false(
self, temp_storage_dir: Path
) -> None:
"""Test deleting nonexistent file returns False."""
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
result = backend.delete("nonexistent.txt")
assert result is False
def test_delete_nested_file(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test deleting a nested file."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "a/b/sample.txt")
result = backend.delete("a/b/sample.txt")
assert result is True
assert not (storage_dir / "a" / "b" / "sample.txt").exists()
class TestLocalStorageBackendGetUrl:
"""Tests for LocalStorageBackend.get_url method."""
def test_get_url_returns_file_path(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test get_url returns file:// URL."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "sample.txt")
url = backend.get_url("sample.txt")
# Should return file:// URL or absolute path
assert "sample.txt" in url
# URL should be usable to locate the file
expected_path = storage_dir / "sample.txt"
assert str(expected_path) in url or expected_path.as_uri() == url
def test_get_url_nonexistent_file(self, temp_storage_dir: Path) -> None:
"""Test get_url for nonexistent file."""
from shared.storage.base import FileNotFoundStorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(FileNotFoundStorageError):
backend.get_url("nonexistent.txt")
class TestLocalStorageBackendUploadBytes:
"""Tests for LocalStorageBackend.upload_bytes method."""
def test_upload_bytes(self, temp_storage_dir: Path) -> None:
"""Test uploading bytes directly."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
data = b"Binary content here"
result = backend.upload_bytes(data, "binary.dat")
assert result == "binary.dat"
assert (storage_dir / "binary.dat").read_bytes() == data
def test_upload_bytes_creates_subdirectories(
self, temp_storage_dir: Path
) -> None:
"""Test that upload_bytes creates subdirectories."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
data = b"content"
backend.upload_bytes(data, "a/b/c/file.dat")
assert (storage_dir / "a" / "b" / "c" / "file.dat").exists()
class TestLocalStorageBackendDownloadBytes:
"""Tests for LocalStorageBackend.download_bytes method."""
def test_download_bytes(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test downloading file as bytes."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
backend.upload(sample_file, "sample.txt")
data = backend.download_bytes("sample.txt")
assert data == b"Hello, World!"
def test_download_bytes_nonexistent(self, temp_storage_dir: Path) -> None:
"""Test downloading nonexistent file as bytes."""
from shared.storage.base import FileNotFoundStorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(FileNotFoundStorageError):
backend.download_bytes("nonexistent.txt")
class TestLocalStorageBackendSecurity:
"""Security tests for LocalStorageBackend - path traversal prevention."""
def test_path_traversal_with_dotdot_blocked(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that path traversal using ../ is blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
with pytest.raises(StorageError, match="Path traversal not allowed"):
backend.upload(sample_file, "../escape.txt")
def test_path_traversal_with_nested_dotdot_blocked(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that nested path traversal is blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
with pytest.raises(StorageError, match="Path traversal not allowed"):
backend.upload(sample_file, "subdir/../../escape.txt")
def test_path_traversal_with_many_dotdot_blocked(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that deeply nested path traversal is blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
with pytest.raises(StorageError, match="Path traversal not allowed"):
backend.upload(sample_file, "a/b/c/../../../../escape.txt")
def test_absolute_path_unix_blocked(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that absolute Unix paths are blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(StorageError, match="Absolute paths not allowed"):
backend.upload(sample_file, "/etc/passwd")
def test_absolute_path_windows_blocked(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that absolute Windows paths are blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(StorageError, match="Absolute paths not allowed"):
backend.upload(sample_file, "C:\\Windows\\System32\\config")
def test_download_path_traversal_blocked(
self, temp_storage_dir: Path
) -> None:
"""Test that path traversal in download is blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(StorageError, match="Path traversal not allowed"):
backend.download("../escape.txt", Path("/tmp/file.txt"))
def test_exists_path_traversal_blocked(
self, temp_storage_dir: Path
) -> None:
"""Test that path traversal in exists is blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(StorageError, match="Path traversal not allowed"):
backend.exists("../escape.txt")
def test_delete_path_traversal_blocked(
self, temp_storage_dir: Path
) -> None:
"""Test that path traversal in delete is blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(StorageError, match="Path traversal not allowed"):
backend.delete("../escape.txt")
def test_get_url_path_traversal_blocked(
self, temp_storage_dir: Path
) -> None:
"""Test that path traversal in get_url is blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(StorageError, match="Path traversal not allowed"):
backend.get_url("../escape.txt")
def test_upload_bytes_path_traversal_blocked(
self, temp_storage_dir: Path
) -> None:
"""Test that path traversal in upload_bytes is blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(StorageError, match="Path traversal not allowed"):
backend.upload_bytes(b"content", "../escape.txt")
def test_download_bytes_path_traversal_blocked(
self, temp_storage_dir: Path
) -> None:
"""Test that path traversal in download_bytes is blocked."""
from shared.storage.base import StorageError
from shared.storage.local import LocalStorageBackend
backend = LocalStorageBackend(base_path=temp_storage_dir)
with pytest.raises(StorageError, match="Path traversal not allowed"):
backend.download_bytes("../escape.txt")
def test_valid_nested_path_still_works(
self, temp_storage_dir: Path, sample_file: Path
) -> None:
"""Test that valid nested paths still work after security fix."""
from shared.storage.local import LocalStorageBackend
storage_dir = temp_storage_dir / "storage"
backend = LocalStorageBackend(base_path=storage_dir)
# Valid nested paths should still work
result = backend.upload(sample_file, "a/b/c/d/file.txt")
assert result == "a/b/c/d/file.txt"
assert (storage_dir / "a" / "b" / "c" / "d" / "file.txt").exists()