"""
Comprehensive error handling and edge case tests.
Tests error scenarios, edge cases, validation, and resilience across
all components of the Unimus MCP Server.
"""
import pytest
from unittest.mock import Mock, patch, MagicMock
import json
import tempfile
import os
from pathlib import Path
from unimus_client import (
UnimusRestClient,
UnimusError,
UnimusNotFoundError,
UnimusValidationError,
UnimusAuthenticationError
)
from config import UnimusConfig, ConfigurationManager, load_config
import server
from tests.fixtures.mock_data import MockDataFactory
class TestConfigurationErrorHandling:
"""Test error handling in configuration system."""
def test_missing_required_config_url(self, clean_environment):
"""Test error when URL is missing."""
with pytest.raises(ValueError, match="Unimus URL must be specified"):
UnimusConfig(url="", token="test-token")
def test_missing_required_config_token(self, clean_environment):
"""Test error when token is missing."""
with pytest.raises(ValueError, match="Unimus token must be specified"):
UnimusConfig(url="https://test.com", token="")
def test_invalid_timeout_value(self, clean_environment):
"""Test error with invalid timeout."""
with pytest.raises(ValueError, match="Timeout must be positive"):
UnimusConfig(url="https://test.com", token="token", timeout=-1)
with pytest.raises(ValueError, match="Timeout must be positive"):
UnimusConfig(url="https://test.com", token="token", timeout=0)
def test_invalid_port_values(self, clean_environment):
"""Test error with invalid port numbers."""
with pytest.raises(ValueError, match="Health check port must be between 1 and 65535"):
UnimusConfig(url="https://test.com", token="token", health_check_port=0)
with pytest.raises(ValueError, match="Health check port must be between 1 and 65535"):
UnimusConfig(url="https://test.com", token="token", health_check_port=70000)
def test_invalid_log_level(self, clean_environment):
"""Test error with invalid log level."""
with pytest.raises(ValueError, match="Invalid log level"):
UnimusConfig(url="https://test.com", token="token", log_level="INVALID")
def test_invalid_page_size(self, clean_environment):
"""Test error with invalid page size."""
with pytest.raises(ValueError, match="Default page size must be positive"):
UnimusConfig(url="https://test.com", token="token", default_page_size=0)
with pytest.raises(ValueError, match="Max search results must be positive"):
UnimusConfig(url="https://test.com", token="token", max_search_results=-5)
def test_config_file_not_found(self, clean_environment):
"""Test handling of non-existent config file."""
# Should not raise exception, fall back to environment variables
os.environ["UNIMUS_URL"] = "https://fallback.com"
os.environ["UNIMUS_TOKEN"] = "fallback-token"
config = load_config("/nonexistent/config.yaml")
assert config.url == "https://fallback.com"
assert config.token == "fallback-token"
def test_corrupted_yaml_config(self, clean_environment):
"""Test handling of corrupted YAML config file."""
corrupted_yaml = """
url: "https://test.com"
token: "test-token"
invalid_yaml: [unclosed list
"""
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
f.write(corrupted_yaml)
temp_path = f.name
try:
with pytest.raises(ValueError, match="Failed to load configuration"):
load_config(temp_path)
finally:
os.unlink(temp_path)
def test_corrupted_toml_config(self, clean_environment):
"""Test handling of corrupted TOML config file."""
corrupted_toml = """
url = "https://test.com"
token = "test-token"
[invalid_section
"""
with tempfile.NamedTemporaryFile(mode='w', suffix='.toml', delete=False) as f:
f.write(corrupted_toml)
temp_path = f.name
try:
with pytest.raises(ValueError, match="Failed to load configuration"):
load_config(temp_path)
finally:
os.unlink(temp_path)
def test_invalid_environment_variables(self, clean_environment):
"""Test handling of invalid environment variable values."""
os.environ["UNIMUS_URL"] = "https://test.com"
os.environ["UNIMUS_TOKEN"] = "test-token"
os.environ["UNIMUS_TIMEOUT"] = "invalid_number"
os.environ["UNIMUS_HEALTH_PORT"] = "not_a_port"
# Should use defaults for invalid values and issue warnings
config = load_config()
assert config.timeout == 30 # default value
assert config.health_check_port == 8080 # default value
class TestUnimusClientErrorHandling:
"""Test error handling in Unimus REST client."""
@pytest.fixture
def client(self):
return UnimusRestClient(
url="https://test.unimus.com",
token="test-token"
)
def test_client_initialization_errors(self):
"""Test client initialization with invalid parameters."""
with pytest.raises(ValueError, match="Base URL is required"):
UnimusRestClient(url="", token="token")
with pytest.raises(ValueError, match="API token is required"):
UnimusRestClient(url="https://test.com", token="")
@patch('unimus_client.requests.get')
def test_network_connection_errors(self, mock_get, client):
"""Test handling of network connection errors."""
import requests
# Connection error
mock_get.side_effect = requests.exceptions.ConnectionError("Network unreachable")
with pytest.raises(UnimusError, match="Network unreachable"):
client.get_health()
# Timeout error
mock_get.side_effect = requests.exceptions.Timeout("Request timeout")
with pytest.raises(UnimusError, match="Request timeout"):
client.get_health()
# DNS resolution error
mock_get.side_effect = requests.exceptions.ConnectionError("Name resolution failed")
with pytest.raises(UnimusError, match="Name resolution failed"):
client.get_health()
@patch('unimus_client.requests.get')
def test_http_error_status_codes(self, mock_get, client):
"""Test handling of various HTTP error status codes."""
error_scenarios = [
(400, "Bad Request", UnimusValidationError),
(401, "Unauthorized", UnimusAuthenticationError),
(403, "Forbidden", UnimusAuthenticationError),
(404, "Not Found", UnimusNotFoundError),
(429, "Too Many Requests", UnimusError),
(500, "Internal Server Error", UnimusError),
(502, "Bad Gateway", UnimusError),
(503, "Service Unavailable", UnimusError),
(504, "Gateway Timeout", UnimusError),
]
for status_code, error_message, expected_exception in error_scenarios:
mock_response = Mock()
mock_response.status_code = status_code
mock_response.json.return_value = {"error": error_message}
mock_get.return_value = mock_response
with pytest.raises(expected_exception, match=error_message):
client.get_health()
@patch('unimus_client.requests.get')
def test_invalid_json_response(self, mock_get, client):
"""Test handling of invalid JSON responses."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.side_effect = json.JSONDecodeError("Invalid JSON", "", 0)
mock_response.text = "This is not JSON"
mock_get.return_value = mock_response
with pytest.raises(UnimusError, match="This is not JSON"):
client.get_health()
@patch('unimus_client.requests.get')
def test_empty_response_body(self, mock_get, client):
"""Test handling of empty response body."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = None
mock_get.return_value = mock_response
# Should handle None response gracefully
result = client._make_request("GET", "/test")
assert result is None
def test_invalid_device_id_types(self, client):
"""Test handling of invalid device ID types."""
with patch.object(client, '_make_request') as mock_request:
mock_request.side_effect = UnimusValidationError("Invalid device ID")
with pytest.raises(ValueError, match="Invalid device ID"):
client.get_device_by_id("not_a_number")
def test_device_not_found_scenarios(self, client):
"""Test various device not found scenarios."""
with patch.object(client, '_make_request') as mock_request:
# Device by ID not found
mock_request.side_effect = UnimusNotFoundError("Device not found")
with pytest.raises(ValueError, match="Device with ID 999 not found"):
server.unimus_get_device_by_id(999)
# Device by address not found
mock_request.return_value = [] # Empty result array
with pytest.raises(UnimusNotFoundError, match="Device with address .* not found"):
client.get_device_by_address("nonexistent.device.com")
class TestMCPToolsErrorHandling:
"""Test error handling in MCP tools."""
def test_health_check_errors(self, mock_server_globals):
"""Test health check error scenarios."""
# Server unreachable
mock_server_globals['unimus_client'].get_health.side_effect = UnimusError("Connection refused")
with pytest.raises(ValueError, match="Failed to get health status: Connection refused"):
server.unimus_get_health()
# Authentication error
mock_server_globals['unimus_client'].get_health.side_effect = UnimusAuthenticationError("Invalid token")
with pytest.raises(ValueError, match="Failed to get health status: Invalid token"):
server.unimus_get_health()
def test_device_retrieval_errors(self, mock_server_globals):
"""Test device retrieval error scenarios."""
# Invalid filter parameters
mock_server_globals['unimus_client'].get_devices.side_effect = UnimusValidationError("Invalid filter")
with pytest.raises(ValueError, match="Failed to get devices: Invalid filter"):
server.unimus_get_devices({"invalid_filter": "value"})
# Device ID validation errors
mock_server_globals['unimus_client'].get_device_by_id.side_effect = UnimusValidationError("Invalid ID format")
with pytest.raises(ValueError, match="Invalid device ID: Invalid ID format"):
server.unimus_get_device_by_id(-1)
def test_backup_operation_errors(self, mock_server_globals):
"""Test backup operation error scenarios."""
# Device has no backups
mock_server_globals['unimus_client'].get_device_backups.return_value = []
result = server.unimus_get_device_backups(123)
assert result == []
# Backup not found
mock_server_globals['unimus_client'].get_backup_by_id.side_effect = UnimusNotFoundError("Backup not found")
with pytest.raises(ValueError, match="Backup with ID 999 not found"):
server.unimus_get_backup_by_id(999)
# Search pattern errors
mock_server_globals['unimus_client'].search_backup_content.side_effect = UnimusValidationError("Invalid regex pattern")
with pytest.raises(ValueError, match="Failed to search backup content: Invalid regex pattern"):
server.unimus_search_backup_content(123, "[invalid_regex")
def test_advanced_feature_errors(self, mock_server_globals):
"""Test advanced feature error scenarios."""
# Device relationships not available
mock_server_globals['unimus_client'].get_device_relationships.side_effect = UnimusError("Feature not available")
with pytest.raises(ValueError, match="Failed to get device relationships: Feature not available"):
server.unimus_get_device_relationships(123)
# Topology analysis fails
mock_server_globals['unimus_client'].get_network_topology_analysis.side_effect = UnimusError("Analysis failed")
with pytest.raises(ValueError, match="Failed to analyze network topology: Analysis failed"):
server.unimus_get_network_topology_analysis()
class TestEdgeCasesAndBoundaryConditions:
"""Test edge cases and boundary conditions."""
def test_empty_device_lists(self, mock_server_globals):
"""Test handling of empty device lists."""
mock_server_globals['unimus_client'].get_devices.return_value = []
result = server.unimus_get_devices()
assert result == []
assert isinstance(result, list)
def test_large_device_lists(self, mock_server_globals):
"""Test handling of large device lists."""
large_device_list = MockDataFactory.create_device_list(1000)
mock_server_globals['unimus_client'].get_devices.return_value = large_device_list
result = server.unimus_get_devices()
assert len(result) == 1000
assert isinstance(result, list)
def test_unicode_and_special_characters(self, mock_server_globals):
"""Test handling of Unicode and special characters."""
device_with_unicode = MockDataFactory.create_device(
description="Test Device 测试设备 🔧",
address="router-café.example.com"
)
mock_server_globals['unimus_client'].get_device_by_id.return_value = device_with_unicode
result = server.unimus_get_device_by_id(123)
assert "测试设备" in result["description"]
assert "café" in result["address"]
def test_very_long_strings(self, mock_server_globals):
"""Test handling of very long strings."""
long_description = "A" * 10000 # Very long description
device_with_long_data = MockDataFactory.create_device(description=long_description)
mock_server_globals['unimus_client'].get_device_by_id.return_value = device_with_long_data
result = server.unimus_get_device_by_id(123)
assert len(result["description"]) == 10000
def test_null_and_none_values(self, mock_server_globals):
"""Test handling of null and None values."""
device_with_nulls = MockDataFactory.create_device()
device_with_nulls["description"] = None
device_with_nulls["model"] = None
mock_server_globals['unimus_client'].get_device_by_id.return_value = device_with_nulls
result = server.unimus_get_device_by_id(123)
assert result["description"] is None
assert result["model"] is None
def test_extreme_numeric_values(self, clean_environment):
"""Test handling of extreme numeric values."""
# Very large timeout
with pytest.raises(ValueError):
UnimusConfig(url="https://test.com", token="token", timeout=999999999)
# Very large port numbers
with pytest.raises(ValueError, match="Health check port must be between 1 and 65535"):
UnimusConfig(url="https://test.com", token="token", health_check_port=999999)
def test_concurrent_access_simulation(self, mock_server_globals):
"""Test simulation of concurrent access patterns."""
import threading
import time
results = []
errors = []
def make_request():
try:
result = server.unimus_get_health()
results.append(result)
except Exception as e:
errors.append(e)
# Mock response
mock_server_globals['unimus_client'].get_health.return_value = {"status": "OK"}
# Simulate concurrent requests
threads = []
for _ in range(10):
thread = threading.Thread(target=make_request)
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
assert len(results) == 10
assert len(errors) == 0
assert all(result["status"] == "OK" for result in results)
class TestResourceLimitsAndPerformance:
"""Test resource limits and performance edge cases."""
def test_memory_intensive_operations(self, mock_server_globals):
"""Test operations that could be memory intensive."""
# Large backup content
large_content = "".join("line {}\n".format(i) for i in range(1000))
large_backup = MockDataFactory.create_backup(content=large_content)
mock_server_globals['unimus_client'].get_backup_by_id.return_value = large_backup
result = server.unimus_get_backup_by_id(123)
assert "content" in result
assert len(result["content"]) > 5000 # Large content
def test_search_result_limits(self, mock_server_globals):
"""Test search operations with result limits."""
# Large number of search results
large_search_results = [
{
"deviceId": 123,
"backupId": 1,
"lineNumber": i,
"content": f"interface GigabitEthernet1/0/{i}",
"match": "interface"
}
for i in range(10000)
]
mock_server_globals['unimus_client'].search_backup_content.return_value = large_search_results
result = server.unimus_search_backup_content(123, "interface")
assert len(result) == 10000
def test_timeout_scenarios(self, mock_server_globals):
"""Test various timeout scenarios."""
import time
def slow_response():
time.sleep(0.1) # Simulate slow response
return {"status": "OK"}
mock_server_globals['unimus_client'].get_health.side_effect = slow_response
# Should complete successfully (within timeout)
result = server.unimus_get_health()
assert result["status"] == "OK"
class TestDataIntegrityAndValidation:
"""Test data integrity and validation edge cases."""
def test_malformed_device_data(self, mock_server_globals):
"""Test handling of malformed device data."""
malformed_device = {
"id": "not_a_number", # Should be int
"address": 12345, # Should be string
"managed": "yes", # Should be boolean
"createTime": "invalid_timestamp"
}
mock_server_globals['unimus_client'].get_device_by_id.return_value = malformed_device
# Should handle malformed data gracefully
result = server.unimus_get_device_by_id(123)
assert result == malformed_device # Pass through, let client handle
def test_missing_required_fields(self, mock_server_globals):
"""Test handling of missing required fields."""
incomplete_device = {
"id": 123,
# Missing required fields like address, description
}
mock_server_globals['unimus_client'].get_device_by_id.return_value = incomplete_device
result = server.unimus_get_device_by_id(123)
assert result["id"] == 123
# Missing fields should be handled by the client application
def test_unexpected_data_types(self, mock_server_globals):
"""Test handling of unexpected data types in responses."""
unexpected_response = "This should be a dict but it's a string"
mock_server_globals['unimus_client'].get_health.return_value = unexpected_response
result = server.unimus_get_health()
assert result == unexpected_response
@pytest.mark.parametrize("invalid_input", [
None,
"",
0,
-1,
"not_a_number",
[],
{},
float('inf'),
float('nan'),
])
def test_parameter_validation_edge_cases(invalid_input, mock_server_globals):
"""Test parameter validation with various invalid inputs."""
# Mock client to raise validation error for invalid inputs
mock_server_globals['unimus_client'].get_device_by_id.side_effect = UnimusValidationError(f"Invalid input: {invalid_input}")
if invalid_input in [None, "", 0, -1, [], {}, float('inf'), float('nan')]:
with pytest.raises(ValueError):
server.unimus_get_device_by_id(invalid_input)
elif invalid_input == "not_a_number":
with pytest.raises(ValueError):
server.unimus_get_device_by_id(invalid_input)
class TestRobustnessAndRecovery:
"""Test system robustness and recovery capabilities."""
def test_partial_failure_recovery(self, mock_server_globals):
"""Test recovery from partial failures."""
# First call fails, second succeeds
mock_server_globals['unimus_client'].get_health.side_effect = [
UnimusError("Temporary failure"),
{"status": "OK"}
]
# First call should fail
with pytest.raises(ValueError, match="Temporary failure"):
server.unimus_get_health()
# Second call should succeed
result = server.unimus_get_health()
assert result["status"] == "OK"
def test_degraded_service_scenarios(self, mock_server_globals):
"""Test handling of degraded service scenarios."""
# Some features unavailable
mock_server_globals['unimus_client'].get_health.return_value = {"status": "LICENSING_UNREACHABLE"}
mock_server_globals['unimus_client'].get_devices.return_value = MockDataFactory.create_device_list(3)
# Health shows degraded but basic operations still work
health = server.unimus_get_health()
assert health["status"] == "LICENSING_UNREACHABLE"
devices = server.unimus_get_devices()
assert len(devices) == 3
