142 lines
5.1 KiB
Python
142 lines
5.1 KiB
Python
"""
|
|
Tests for SecretsManager.
|
|
|
|
Validates Docker Secrets and environment variable handling.
|
|
"""
|
|
|
|
import os
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from src.security.secrets_manager import SecretsManager
|
|
|
|
|
|
class TestSecretsManager:
|
|
"""Tests for SecretsManager functionality."""
|
|
|
|
@pytest.fixture
|
|
def temp_secrets_dir(self, tmp_path):
|
|
"""Create a temporary secrets directory with test files."""
|
|
secrets_dir = tmp_path / "secrets"
|
|
secrets_dir.mkdir()
|
|
|
|
# Create some test secret files
|
|
(secrets_dir / "POSTGRES_PASSWORD").write_text("test_password_123")
|
|
(secrets_dir / "api_key").write_text("sk-test-key-abc")
|
|
|
|
return secrets_dir
|
|
|
|
def test_load_docker_secrets(self, temp_secrets_dir):
|
|
"""Test loading secrets from Docker Secrets directory."""
|
|
manager = SecretsManager(secrets_path=temp_secrets_dir)
|
|
|
|
password = manager.get("POSTGRES_PASSWORD")
|
|
|
|
assert password == "test_password_123"
|
|
|
|
def test_secret_case_insensitive(self, temp_secrets_dir):
|
|
"""Test that secret names are case-insensitive."""
|
|
manager = SecretsManager(secrets_path=temp_secrets_dir)
|
|
|
|
# File is 'api_key' but accessed as 'API_KEY'
|
|
key = manager.get("API_KEY")
|
|
|
|
assert key == "sk-test-key-abc"
|
|
|
|
def test_fallback_to_environment(self):
|
|
"""Test fallback to environment variables."""
|
|
manager = SecretsManager(secrets_path=Path("/nonexistent"))
|
|
|
|
with patch.dict(os.environ, {"TEST_SECRET": "env_value"}):
|
|
value = manager.get("TEST_SECRET")
|
|
|
|
assert value == "env_value"
|
|
|
|
def test_return_default_when_not_found(self):
|
|
"""Test returning default value when secret not found."""
|
|
manager = SecretsManager(secrets_path=Path("/nonexistent"))
|
|
|
|
value = manager.get("NONEXISTENT_SECRET", default="default_value")
|
|
|
|
assert value == "default_value"
|
|
|
|
def test_required_secret_raises_error(self):
|
|
"""Test that required secrets raise ValueError when missing."""
|
|
manager = SecretsManager(secrets_path=Path("/nonexistent"))
|
|
|
|
with pytest.raises(ValueError) as exc_info:
|
|
manager.get("MISSING_SECRET", required=True)
|
|
|
|
assert "MISSING_SECRET" in str(exc_info.value)
|
|
|
|
def test_validate_required_secrets(self, temp_secrets_dir):
|
|
"""Test validating multiple required secrets."""
|
|
manager = SecretsManager(secrets_path=temp_secrets_dir)
|
|
|
|
# POSTGRES_PASSWORD exists, MISSING_ONE doesn't
|
|
missing = manager.validate_required_secrets([
|
|
"POSTGRES_PASSWORD",
|
|
"MISSING_ONE",
|
|
"MISSING_TWO"
|
|
])
|
|
|
|
assert "POSTGRES_PASSWORD" not in missing
|
|
assert "MISSING_ONE" in missing
|
|
assert "MISSING_TWO" in missing
|
|
|
|
def test_get_required_secrets_list(self):
|
|
"""Test getting the list of required secrets."""
|
|
required = SecretsManager.get_required_secrets()
|
|
|
|
assert "POSTGRES_USER" in required
|
|
assert "POSTGRES_PASSWORD" in required
|
|
assert "POSTGRES_DB" in required
|
|
assert "MAIL_PASSWORD" in required
|
|
|
|
def test_docker_secrets_priority_over_env(self, temp_secrets_dir):
|
|
"""Test that Docker Secrets take priority over environment."""
|
|
manager = SecretsManager(secrets_path=temp_secrets_dir)
|
|
|
|
# Set both Docker Secret and env var
|
|
with patch.dict(os.environ, {"POSTGRES_PASSWORD": "env_password"}):
|
|
password = manager.get("POSTGRES_PASSWORD")
|
|
|
|
# Should use Docker Secret, not env var
|
|
assert password == "test_password_123"
|
|
|
|
def test_caching(self, temp_secrets_dir):
|
|
"""Test that secrets are cached after first load."""
|
|
manager = SecretsManager(secrets_path=temp_secrets_dir)
|
|
|
|
# First access loads secrets
|
|
manager.get("POSTGRES_PASSWORD")
|
|
assert manager._loaded is True
|
|
|
|
# Modify the file (would not be picked up due to caching)
|
|
(temp_secrets_dir / "POSTGRES_PASSWORD").write_text("new_password")
|
|
|
|
# Should still return cached value
|
|
password = manager.get("POSTGRES_PASSWORD")
|
|
assert password == "test_password_123"
|
|
|
|
def test_empty_secrets_directory(self, tmp_path):
|
|
"""Test handling of empty secrets directory."""
|
|
empty_dir = tmp_path / "empty_secrets"
|
|
empty_dir.mkdir()
|
|
|
|
manager = SecretsManager(secrets_path=empty_dir)
|
|
|
|
# Should not raise, just return None/default
|
|
value = manager.get("ANY_SECRET", default="fallback")
|
|
assert value == "fallback"
|
|
|
|
def test_strip_whitespace_from_secrets(self, temp_secrets_dir):
|
|
"""Test that whitespace is stripped from secret values."""
|
|
(temp_secrets_dir / "WHITESPACE_SECRET").write_text(" value_with_spaces \n")
|
|
|
|
manager = SecretsManager(secrets_path=temp_secrets_dir)
|
|
value = manager.get("WHITESPACE_SECRET")
|
|
|
|
assert value == "value_with_spaces"
|