302 lines
9.8 KiB
Python
302 lines
9.8 KiB
Python
"""
|
|
Tests for storage base module.
|
|
|
|
TDD Phase 1: RED - Write tests first, then implement to pass.
|
|
"""
|
|
|
|
from abc import ABC
|
|
from pathlib import Path
|
|
from typing import BinaryIO
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
class TestStorageBackendInterface:
|
|
"""Tests for StorageBackend abstract base class."""
|
|
|
|
def test_cannot_instantiate_directly(self) -> None:
|
|
"""Test that StorageBackend cannot be instantiated."""
|
|
from shared.storage.base import StorageBackend
|
|
|
|
with pytest.raises(TypeError):
|
|
StorageBackend() # type: ignore
|
|
|
|
def test_is_abstract_base_class(self) -> None:
|
|
"""Test that StorageBackend is an ABC."""
|
|
from shared.storage.base import StorageBackend
|
|
|
|
assert issubclass(StorageBackend, ABC)
|
|
|
|
def test_subclass_must_implement_upload(self) -> None:
|
|
"""Test that subclass must implement upload method."""
|
|
from shared.storage.base import StorageBackend
|
|
|
|
class IncompleteBackend(StorageBackend):
|
|
def download(self, remote_path: str, local_path: Path) -> Path:
|
|
return local_path
|
|
|
|
def exists(self, remote_path: str) -> bool:
|
|
return False
|
|
|
|
def list_files(self, prefix: str) -> list[str]:
|
|
return []
|
|
|
|
def delete(self, remote_path: str) -> bool:
|
|
return True
|
|
|
|
def get_url(self, remote_path: str) -> str:
|
|
return ""
|
|
|
|
with pytest.raises(TypeError):
|
|
IncompleteBackend() # type: ignore
|
|
|
|
def test_subclass_must_implement_download(self) -> None:
|
|
"""Test that subclass must implement download method."""
|
|
from shared.storage.base import StorageBackend
|
|
|
|
class IncompleteBackend(StorageBackend):
|
|
def upload(
|
|
self, local_path: Path, remote_path: str, overwrite: bool = False
|
|
) -> str:
|
|
return remote_path
|
|
|
|
def exists(self, remote_path: str) -> bool:
|
|
return False
|
|
|
|
def list_files(self, prefix: str) -> list[str]:
|
|
return []
|
|
|
|
def delete(self, remote_path: str) -> bool:
|
|
return True
|
|
|
|
def get_url(self, remote_path: str) -> str:
|
|
return ""
|
|
|
|
with pytest.raises(TypeError):
|
|
IncompleteBackend() # type: ignore
|
|
|
|
def test_subclass_must_implement_exists(self) -> None:
|
|
"""Test that subclass must implement exists method."""
|
|
from shared.storage.base import StorageBackend
|
|
|
|
class IncompleteBackend(StorageBackend):
|
|
def upload(
|
|
self, local_path: Path, remote_path: str, overwrite: bool = False
|
|
) -> str:
|
|
return remote_path
|
|
|
|
def download(self, remote_path: str, local_path: Path) -> Path:
|
|
return local_path
|
|
|
|
def list_files(self, prefix: str) -> list[str]:
|
|
return []
|
|
|
|
def delete(self, remote_path: str) -> bool:
|
|
return True
|
|
|
|
def get_url(self, remote_path: str) -> str:
|
|
return ""
|
|
|
|
with pytest.raises(TypeError):
|
|
IncompleteBackend() # type: ignore
|
|
|
|
def test_subclass_must_implement_list_files(self) -> None:
|
|
"""Test that subclass must implement list_files method."""
|
|
from shared.storage.base import StorageBackend
|
|
|
|
class IncompleteBackend(StorageBackend):
|
|
def upload(
|
|
self, local_path: Path, remote_path: str, overwrite: bool = False
|
|
) -> str:
|
|
return remote_path
|
|
|
|
def download(self, remote_path: str, local_path: Path) -> Path:
|
|
return local_path
|
|
|
|
def exists(self, remote_path: str) -> bool:
|
|
return False
|
|
|
|
def delete(self, remote_path: str) -> bool:
|
|
return True
|
|
|
|
def get_url(self, remote_path: str) -> str:
|
|
return ""
|
|
|
|
with pytest.raises(TypeError):
|
|
IncompleteBackend() # type: ignore
|
|
|
|
def test_subclass_must_implement_delete(self) -> None:
|
|
"""Test that subclass must implement delete method."""
|
|
from shared.storage.base import StorageBackend
|
|
|
|
class IncompleteBackend(StorageBackend):
|
|
def upload(
|
|
self, local_path: Path, remote_path: str, overwrite: bool = False
|
|
) -> str:
|
|
return remote_path
|
|
|
|
def download(self, remote_path: str, local_path: Path) -> Path:
|
|
return local_path
|
|
|
|
def exists(self, remote_path: str) -> bool:
|
|
return False
|
|
|
|
def list_files(self, prefix: str) -> list[str]:
|
|
return []
|
|
|
|
def get_url(self, remote_path: str) -> str:
|
|
return ""
|
|
|
|
with pytest.raises(TypeError):
|
|
IncompleteBackend() # type: ignore
|
|
|
|
def test_subclass_must_implement_get_url(self) -> None:
|
|
"""Test that subclass must implement get_url method."""
|
|
from shared.storage.base import StorageBackend
|
|
|
|
class IncompleteBackend(StorageBackend):
|
|
def upload(
|
|
self, local_path: Path, remote_path: str, overwrite: bool = False
|
|
) -> str:
|
|
return remote_path
|
|
|
|
def download(self, remote_path: str, local_path: Path) -> Path:
|
|
return local_path
|
|
|
|
def exists(self, remote_path: str) -> bool:
|
|
return False
|
|
|
|
def list_files(self, prefix: str) -> list[str]:
|
|
return []
|
|
|
|
def delete(self, remote_path: str) -> bool:
|
|
return True
|
|
|
|
with pytest.raises(TypeError):
|
|
IncompleteBackend() # type: ignore
|
|
|
|
def test_valid_subclass_can_be_instantiated(self) -> None:
|
|
"""Test that a complete subclass can be instantiated."""
|
|
from shared.storage.base import StorageBackend
|
|
|
|
class CompleteBackend(StorageBackend):
|
|
def upload(
|
|
self, local_path: Path, remote_path: str, overwrite: bool = False
|
|
) -> str:
|
|
return remote_path
|
|
|
|
def download(self, remote_path: str, local_path: Path) -> Path:
|
|
return local_path
|
|
|
|
def exists(self, remote_path: str) -> bool:
|
|
return False
|
|
|
|
def list_files(self, prefix: str) -> list[str]:
|
|
return []
|
|
|
|
def delete(self, remote_path: str) -> bool:
|
|
return True
|
|
|
|
def get_url(self, remote_path: str) -> str:
|
|
return ""
|
|
|
|
def get_presigned_url(
|
|
self, remote_path: str, expires_in_seconds: int = 3600
|
|
) -> str:
|
|
return ""
|
|
|
|
backend = CompleteBackend()
|
|
assert isinstance(backend, StorageBackend)
|
|
|
|
|
|
class TestStorageError:
|
|
"""Tests for StorageError exception."""
|
|
|
|
def test_storage_error_is_exception(self) -> None:
|
|
"""Test that StorageError is an Exception."""
|
|
from shared.storage.base import StorageError
|
|
|
|
assert issubclass(StorageError, Exception)
|
|
|
|
def test_storage_error_with_message(self) -> None:
|
|
"""Test StorageError with message."""
|
|
from shared.storage.base import StorageError
|
|
|
|
error = StorageError("Upload failed")
|
|
assert str(error) == "Upload failed"
|
|
|
|
def test_storage_error_can_be_raised(self) -> None:
|
|
"""Test that StorageError can be raised and caught."""
|
|
from shared.storage.base import StorageError
|
|
|
|
with pytest.raises(StorageError, match="test error"):
|
|
raise StorageError("test error")
|
|
|
|
|
|
class TestFileNotFoundError:
|
|
"""Tests for FileNotFoundStorageError exception."""
|
|
|
|
def test_file_not_found_is_storage_error(self) -> None:
|
|
"""Test that FileNotFoundStorageError is a StorageError."""
|
|
from shared.storage.base import FileNotFoundStorageError, StorageError
|
|
|
|
assert issubclass(FileNotFoundStorageError, StorageError)
|
|
|
|
def test_file_not_found_with_path(self) -> None:
|
|
"""Test FileNotFoundStorageError with path."""
|
|
from shared.storage.base import FileNotFoundStorageError
|
|
|
|
error = FileNotFoundStorageError("images/test.png")
|
|
assert "images/test.png" in str(error)
|
|
|
|
|
|
class TestStorageConfig:
|
|
"""Tests for StorageConfig dataclass."""
|
|
|
|
def test_storage_config_creation(self) -> None:
|
|
"""Test creating StorageConfig."""
|
|
from shared.storage.base import StorageConfig
|
|
|
|
config = StorageConfig(
|
|
backend_type="azure_blob",
|
|
connection_string="DefaultEndpointsProtocol=https;...",
|
|
container_name="training-images",
|
|
)
|
|
|
|
assert config.backend_type == "azure_blob"
|
|
assert config.connection_string == "DefaultEndpointsProtocol=https;..."
|
|
assert config.container_name == "training-images"
|
|
|
|
def test_storage_config_defaults(self) -> None:
|
|
"""Test StorageConfig with defaults."""
|
|
from shared.storage.base import StorageConfig
|
|
|
|
config = StorageConfig(backend_type="local")
|
|
|
|
assert config.backend_type == "local"
|
|
assert config.connection_string is None
|
|
assert config.container_name is None
|
|
assert config.base_path is None
|
|
|
|
def test_storage_config_with_base_path(self) -> None:
|
|
"""Test StorageConfig with base_path for local backend."""
|
|
from shared.storage.base import StorageConfig
|
|
|
|
config = StorageConfig(
|
|
backend_type="local",
|
|
base_path=Path("/data/images"),
|
|
)
|
|
|
|
assert config.backend_type == "local"
|
|
assert config.base_path == Path("/data/images")
|
|
|
|
def test_storage_config_immutable(self) -> None:
|
|
"""Test that StorageConfig is immutable (frozen)."""
|
|
from shared.storage.base import StorageConfig
|
|
|
|
config = StorageConfig(backend_type="local")
|
|
|
|
with pytest.raises(AttributeError):
|
|
config.backend_type = "azure_blob" # type: ignore
|