""" Basic tests for Portainer Core MCP Server. This module contains basic tests to verify the setup and configuration. """ import pytest import os from unittest.mock import patch from portainer_core.config import PortainerConfig, get_config from portainer_core.utils.errors import PortainerError, PortainerAuthenticationError from portainer_core.utils.logging import get_logger, set_correlation_id class TestConfiguration: """Test configuration management.""" def test_config_validation_with_api_key(self): """Test configuration validation with API key.""" with patch.dict(os.environ, { 'PORTAINER_URL': 'https://test.example.com', 'PORTAINER_API_KEY': 'test-api-key' }): config = PortainerConfig() config.validate_auth_config() assert config.portainer_url == 'https://test.example.com' assert config.portainer_api_key == 'test-api-key' assert config.use_api_key_auth is True assert config.use_credentials_auth is False def test_config_validation_with_credentials(self): """Test configuration validation with username/password.""" with patch.dict(os.environ, { 'PORTAINER_URL': 'https://test.example.com', 'PORTAINER_USERNAME': 'admin', 'PORTAINER_PASSWORD': 'password' }): config = PortainerConfig() config.validate_auth_config() assert config.portainer_url == 'https://test.example.com' assert config.portainer_username == 'admin' assert config.portainer_password == 'password' assert config.use_api_key_auth is False assert config.use_credentials_auth is True def test_config_validation_no_auth(self): """Test configuration validation without authentication.""" with patch.dict(os.environ, { 'PORTAINER_URL': 'https://test.example.com' }, clear=True): config = PortainerConfig() with pytest.raises(ValueError, match="Either PORTAINER_API_KEY or both"): config.validate_auth_config() def test_invalid_url(self): """Test invalid URL validation.""" with patch.dict(os.environ, { 'PORTAINER_URL': 'invalid-url', 'PORTAINER_API_KEY': 'test-key' }): with pytest.raises(ValueError, match="Invalid URL format"): PortainerConfig() def test_url_trailing_slash_removal(self): """Test URL trailing slash removal.""" with patch.dict(os.environ, { 'PORTAINER_URL': 'https://test.example.com/', 'PORTAINER_API_KEY': 'test-key' }): config = PortainerConfig() assert config.portainer_url == 'https://test.example.com' def test_api_base_url(self): """Test API base URL construction.""" with patch.dict(os.environ, { 'PORTAINER_URL': 'https://test.example.com', 'PORTAINER_API_KEY': 'test-key' }): config = PortainerConfig() assert config.api_base_url == 'https://test.example.com/api' class TestErrors: """Test error handling utilities.""" def test_portainer_error_basic(self): """Test basic PortainerError.""" error = PortainerError("Test error") assert str(error) == "Test error" assert error.message == "Test error" assert error.status_code is None assert error.details == {} def test_portainer_error_with_status_code(self): """Test PortainerError with status code.""" error = PortainerError("Test error", status_code=400) assert str(error) == "[400] Test error" assert error.status_code == 400 def test_portainer_authentication_error(self): """Test PortainerAuthenticationError.""" error = PortainerAuthenticationError() assert error.status_code == 401 assert "Authentication failed" in str(error) def test_error_mapping(self): """Test HTTP error mapping.""" from portainer_core.utils.errors import map_http_error error = map_http_error(404, "Not found") assert error.__class__.__name__ == "PortainerNotFoundError" assert error.status_code == 404 error = map_http_error(500, "Server error") assert error.__class__.__name__ == "PortainerServerError" assert error.status_code == 500 class TestLogging: """Test logging utilities.""" def test_get_logger(self): """Test logger creation.""" logger = get_logger("test") assert logger is not None def test_correlation_id(self): """Test correlation ID functionality.""" correlation_id = set_correlation_id("test-id") assert correlation_id == "test-id" from portainer_core.utils.logging import get_correlation_id assert get_correlation_id() == "test-id" def test_correlation_id_auto_generation(self): """Test automatic correlation ID generation.""" correlation_id = set_correlation_id() assert correlation_id is not None assert len(correlation_id) > 0 class TestCircuitBreaker: """Test circuit breaker functionality.""" def test_circuit_breaker_initial_state(self): """Test circuit breaker initial state.""" from portainer_core.services.base import CircuitBreaker, CircuitBreakerState cb = CircuitBreaker() assert cb.state == CircuitBreakerState.CLOSED assert cb.can_execute() is True assert cb.failure_count == 0 def test_circuit_breaker_failure_threshold(self): """Test circuit breaker failure threshold.""" from portainer_core.services.base import CircuitBreaker, CircuitBreakerState cb = CircuitBreaker(failure_threshold=2) # First failure cb.record_failure() assert cb.state == CircuitBreakerState.CLOSED assert cb.can_execute() is True # Second failure - should open cb.record_failure() assert cb.state == CircuitBreakerState.OPEN assert cb.can_execute() is False def test_circuit_breaker_success_reset(self): """Test circuit breaker success reset.""" from portainer_core.services.base import CircuitBreaker, CircuitBreakerState cb = CircuitBreaker() cb.record_failure() cb.record_success() assert cb.failure_count == 0 assert cb.last_failure_time is None if __name__ == "__main__": pytest.main([__file__])