713 lines
24 KiB
Python
713 lines
24 KiB
Python
"""
|
|
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()
|