349 lines
12 KiB
Python
349 lines
12 KiB
Python
"""
|
|
Tests for storage configuration file loader.
|
|
|
|
TDD Phase 1: RED - Write tests first, then implement to pass.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir() -> Path:
|
|
"""Create a temporary directory for tests."""
|
|
temp_dir = Path(tempfile.mkdtemp())
|
|
yield temp_dir
|
|
shutil.rmtree(temp_dir, ignore_errors=True)
|
|
|
|
|
|
class TestEnvVarSubstitution:
|
|
"""Tests for environment variable substitution in config values."""
|
|
|
|
def test_substitute_simple_env_var(self) -> None:
|
|
"""Test substituting a simple environment variable."""
|
|
from shared.storage.config_loader import substitute_env_vars
|
|
|
|
with patch.dict(os.environ, {"MY_VAR": "my_value"}):
|
|
result = substitute_env_vars("${MY_VAR}")
|
|
assert result == "my_value"
|
|
|
|
def test_substitute_env_var_with_default(self) -> None:
|
|
"""Test substituting env var with default when var is not set."""
|
|
from shared.storage.config_loader import substitute_env_vars
|
|
|
|
# Ensure var is not set
|
|
os.environ.pop("UNSET_VAR", None)
|
|
|
|
result = substitute_env_vars("${UNSET_VAR:-default_value}")
|
|
assert result == "default_value"
|
|
|
|
def test_substitute_env_var_ignores_default_when_set(self) -> None:
|
|
"""Test that default is ignored when env var is set."""
|
|
from shared.storage.config_loader import substitute_env_vars
|
|
|
|
with patch.dict(os.environ, {"SET_VAR": "actual_value"}):
|
|
result = substitute_env_vars("${SET_VAR:-default_value}")
|
|
assert result == "actual_value"
|
|
|
|
def test_substitute_multiple_env_vars(self) -> None:
|
|
"""Test substituting multiple env vars in one string."""
|
|
from shared.storage.config_loader import substitute_env_vars
|
|
|
|
with patch.dict(os.environ, {"HOST": "localhost", "PORT": "5432"}):
|
|
result = substitute_env_vars("postgres://${HOST}:${PORT}/db")
|
|
assert result == "postgres://localhost:5432/db"
|
|
|
|
def test_substitute_preserves_non_env_text(self) -> None:
|
|
"""Test that non-env-var text is preserved."""
|
|
from shared.storage.config_loader import substitute_env_vars
|
|
|
|
with patch.dict(os.environ, {"VAR": "value"}):
|
|
result = substitute_env_vars("prefix_${VAR}_suffix")
|
|
assert result == "prefix_value_suffix"
|
|
|
|
def test_substitute_empty_string_when_not_set_and_no_default(self) -> None:
|
|
"""Test that empty string is returned when var not set and no default."""
|
|
from shared.storage.config_loader import substitute_env_vars
|
|
|
|
os.environ.pop("MISSING_VAR", None)
|
|
|
|
result = substitute_env_vars("${MISSING_VAR}")
|
|
assert result == ""
|
|
|
|
|
|
class TestLoadStorageConfigYaml:
|
|
"""Tests for loading storage configuration from YAML files."""
|
|
|
|
def test_load_local_backend_config(self, temp_dir: Path) -> None:
|
|
"""Test loading configuration for local backend."""
|
|
from shared.storage.config_loader import load_storage_config
|
|
|
|
config_path = temp_dir / "storage.yaml"
|
|
config_path.write_text("""
|
|
backend: local
|
|
presigned_url_expiry: 3600
|
|
|
|
local:
|
|
base_path: ./data/storage
|
|
""")
|
|
|
|
config = load_storage_config(config_path)
|
|
|
|
assert config.backend_type == "local"
|
|
assert config.presigned_url_expiry == 3600
|
|
assert config.local is not None
|
|
assert config.local.base_path == Path("./data/storage")
|
|
|
|
def test_load_azure_backend_config(self, temp_dir: Path) -> None:
|
|
"""Test loading configuration for Azure backend."""
|
|
from shared.storage.config_loader import load_storage_config
|
|
|
|
config_path = temp_dir / "storage.yaml"
|
|
config_path.write_text("""
|
|
backend: azure_blob
|
|
presigned_url_expiry: 7200
|
|
|
|
azure:
|
|
connection_string: DefaultEndpointsProtocol=https;AccountName=test
|
|
container_name: documents
|
|
create_container: true
|
|
""")
|
|
|
|
config = load_storage_config(config_path)
|
|
|
|
assert config.backend_type == "azure_blob"
|
|
assert config.presigned_url_expiry == 7200
|
|
assert config.azure is not None
|
|
assert config.azure.connection_string == "DefaultEndpointsProtocol=https;AccountName=test"
|
|
assert config.azure.container_name == "documents"
|
|
assert config.azure.create_container is True
|
|
|
|
def test_load_s3_backend_config(self, temp_dir: Path) -> None:
|
|
"""Test loading configuration for S3 backend."""
|
|
from shared.storage.config_loader import load_storage_config
|
|
|
|
config_path = temp_dir / "storage.yaml"
|
|
config_path.write_text("""
|
|
backend: s3
|
|
presigned_url_expiry: 1800
|
|
|
|
s3:
|
|
bucket_name: my-bucket
|
|
region_name: us-west-2
|
|
endpoint_url: http://localhost:9000
|
|
create_bucket: false
|
|
""")
|
|
|
|
config = load_storage_config(config_path)
|
|
|
|
assert config.backend_type == "s3"
|
|
assert config.presigned_url_expiry == 1800
|
|
assert config.s3 is not None
|
|
assert config.s3.bucket_name == "my-bucket"
|
|
assert config.s3.region_name == "us-west-2"
|
|
assert config.s3.endpoint_url == "http://localhost:9000"
|
|
assert config.s3.create_bucket is False
|
|
|
|
def test_load_config_with_env_var_substitution(self, temp_dir: Path) -> None:
|
|
"""Test that environment variables are substituted in config."""
|
|
from shared.storage.config_loader import load_storage_config
|
|
|
|
config_path = temp_dir / "storage.yaml"
|
|
config_path.write_text("""
|
|
backend: ${STORAGE_BACKEND:-local}
|
|
|
|
local:
|
|
base_path: ${STORAGE_PATH:-./default/path}
|
|
""")
|
|
|
|
with patch.dict(os.environ, {"STORAGE_BACKEND": "local", "STORAGE_PATH": "/custom/path"}):
|
|
config = load_storage_config(config_path)
|
|
|
|
assert config.backend_type == "local"
|
|
assert config.local is not None
|
|
assert config.local.base_path == Path("/custom/path")
|
|
|
|
def test_load_config_file_not_found_raises(self, temp_dir: Path) -> None:
|
|
"""Test that FileNotFoundError is raised for missing config file."""
|
|
from shared.storage.config_loader import load_storage_config
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
load_storage_config(temp_dir / "nonexistent.yaml")
|
|
|
|
def test_load_config_invalid_yaml_raises(self, temp_dir: Path) -> None:
|
|
"""Test that ValueError is raised for invalid YAML."""
|
|
from shared.storage.config_loader import load_storage_config
|
|
|
|
config_path = temp_dir / "storage.yaml"
|
|
config_path.write_text("invalid: yaml: content: [")
|
|
|
|
with pytest.raises(ValueError, match="Invalid"):
|
|
load_storage_config(config_path)
|
|
|
|
def test_load_config_missing_backend_raises(self, temp_dir: Path) -> None:
|
|
"""Test that ValueError is raised when backend is missing."""
|
|
from shared.storage.config_loader import load_storage_config
|
|
|
|
config_path = temp_dir / "storage.yaml"
|
|
config_path.write_text("""
|
|
local:
|
|
base_path: ./data
|
|
""")
|
|
|
|
with pytest.raises(ValueError, match="backend"):
|
|
load_storage_config(config_path)
|
|
|
|
def test_load_config_default_presigned_url_expiry(self, temp_dir: Path) -> None:
|
|
"""Test default presigned_url_expiry when not specified."""
|
|
from shared.storage.config_loader import load_storage_config
|
|
|
|
config_path = temp_dir / "storage.yaml"
|
|
config_path.write_text("""
|
|
backend: local
|
|
|
|
local:
|
|
base_path: ./data
|
|
""")
|
|
|
|
config = load_storage_config(config_path)
|
|
|
|
assert config.presigned_url_expiry == 3600 # Default value
|
|
|
|
|
|
class TestStorageFileConfig:
|
|
"""Tests for StorageFileConfig dataclass."""
|
|
|
|
def test_storage_file_config_is_immutable(self) -> None:
|
|
"""Test that StorageFileConfig is frozen (immutable)."""
|
|
from shared.storage.config_loader import StorageFileConfig
|
|
|
|
config = StorageFileConfig(backend_type="local")
|
|
|
|
with pytest.raises(AttributeError):
|
|
config.backend_type = "azure_blob" # type: ignore
|
|
|
|
def test_storage_file_config_defaults(self) -> None:
|
|
"""Test StorageFileConfig default values."""
|
|
from shared.storage.config_loader import StorageFileConfig
|
|
|
|
config = StorageFileConfig(backend_type="local")
|
|
|
|
assert config.backend_type == "local"
|
|
assert config.local is None
|
|
assert config.azure is None
|
|
assert config.s3 is None
|
|
assert config.presigned_url_expiry == 3600
|
|
|
|
|
|
class TestLocalConfig:
|
|
"""Tests for LocalConfig dataclass."""
|
|
|
|
def test_local_config_creation(self) -> None:
|
|
"""Test creating LocalConfig."""
|
|
from shared.storage.config_loader import LocalConfig
|
|
|
|
config = LocalConfig(base_path=Path("/data/storage"))
|
|
|
|
assert config.base_path == Path("/data/storage")
|
|
|
|
def test_local_config_is_immutable(self) -> None:
|
|
"""Test that LocalConfig is frozen."""
|
|
from shared.storage.config_loader import LocalConfig
|
|
|
|
config = LocalConfig(base_path=Path("/data"))
|
|
|
|
with pytest.raises(AttributeError):
|
|
config.base_path = Path("/other") # type: ignore
|
|
|
|
|
|
class TestAzureConfig:
|
|
"""Tests for AzureConfig dataclass."""
|
|
|
|
def test_azure_config_creation(self) -> None:
|
|
"""Test creating AzureConfig."""
|
|
from shared.storage.config_loader import AzureConfig
|
|
|
|
config = AzureConfig(
|
|
connection_string="test_connection",
|
|
container_name="test_container",
|
|
create_container=True,
|
|
)
|
|
|
|
assert config.connection_string == "test_connection"
|
|
assert config.container_name == "test_container"
|
|
assert config.create_container is True
|
|
|
|
def test_azure_config_defaults(self) -> None:
|
|
"""Test AzureConfig default values."""
|
|
from shared.storage.config_loader import AzureConfig
|
|
|
|
config = AzureConfig(
|
|
connection_string="conn",
|
|
container_name="container",
|
|
)
|
|
|
|
assert config.create_container is False
|
|
|
|
def test_azure_config_is_immutable(self) -> None:
|
|
"""Test that AzureConfig is frozen."""
|
|
from shared.storage.config_loader import AzureConfig
|
|
|
|
config = AzureConfig(
|
|
connection_string="conn",
|
|
container_name="container",
|
|
)
|
|
|
|
with pytest.raises(AttributeError):
|
|
config.container_name = "other" # type: ignore
|
|
|
|
|
|
class TestS3Config:
|
|
"""Tests for S3Config dataclass."""
|
|
|
|
def test_s3_config_creation(self) -> None:
|
|
"""Test creating S3Config."""
|
|
from shared.storage.config_loader import S3Config
|
|
|
|
config = S3Config(
|
|
bucket_name="my-bucket",
|
|
region_name="us-east-1",
|
|
access_key_id="AKIAIOSFODNN7EXAMPLE",
|
|
secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
|
|
endpoint_url="http://localhost:9000",
|
|
create_bucket=True,
|
|
)
|
|
|
|
assert config.bucket_name == "my-bucket"
|
|
assert config.region_name == "us-east-1"
|
|
assert config.access_key_id == "AKIAIOSFODNN7EXAMPLE"
|
|
assert config.secret_access_key == "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
|
|
assert config.endpoint_url == "http://localhost:9000"
|
|
assert config.create_bucket is True
|
|
|
|
def test_s3_config_minimal(self) -> None:
|
|
"""Test S3Config with only required fields."""
|
|
from shared.storage.config_loader import S3Config
|
|
|
|
config = S3Config(bucket_name="bucket")
|
|
|
|
assert config.bucket_name == "bucket"
|
|
assert config.region_name is None
|
|
assert config.access_key_id is None
|
|
assert config.secret_access_key is None
|
|
assert config.endpoint_url is None
|
|
assert config.create_bucket is False
|
|
|
|
def test_s3_config_is_immutable(self) -> None:
|
|
"""Test that S3Config is frozen."""
|
|
from shared.storage.config_loader import S3Config
|
|
|
|
config = S3Config(bucket_name="bucket")
|
|
|
|
with pytest.raises(AttributeError):
|
|
config.bucket_name = "other" # type: ignore
|