184 lines
6.6 KiB
Python
184 lines
6.6 KiB
Python
"""
|
|
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__]) |