""" 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"