"""
Unit tests for all MCP tools in server.py
Tests each MCP tool function with mocked dependencies to ensure proper
error handling, parameter validation, and response formatting.
"""
import pytest
from unittest.mock import Mock, patch, MagicMock
from typing import Dict, List, Any
# Import MCP tools from server.py
from server import (
unimus_get_health,
unimus_get_devices,
unimus_get_device_by_id,
unimus_get_device_by_address,
unimus_get_device_backups,
unimus_get_backup_by_id,
unimus_get_schedules,
unimus_get_schedule_by_id,
unimus_search_backup_content,
unimus_get_devices_with_changed_backups,
unimus_get_backup_diff,
unimus_get_device_relationships,
unimus_get_network_topology_analysis
)
from unimus_client import UnimusError, UnimusNotFoundError, UnimusValidationError, UnimusAuthenticationError
from tests.fixtures.mock_data import MockDataFactory
class TestUnimusGetHealth:
"""Test unimus_get_health MCP tool."""
def test_get_health_success(self, mock_server_globals):
"""Test successful health status retrieval."""
# Setup
expected_response = {"status": "OK"}
mock_server_globals['unimus_client'].get_health.return_value = expected_response
# Execute
result = unimus_get_health()
# Assert
assert result == expected_response
mock_server_globals['unimus_client'].get_health.assert_called_once()
def test_get_health_licensing_unreachable(self, mock_server_globals):
"""Test health status with licensing issues."""
# Setup
expected_response = {"status": "LICENSING_UNREACHABLE"}
mock_server_globals['unimus_client'].get_health.return_value = expected_response
# Execute
result = unimus_get_health()
# Assert
assert result == expected_response
assert result["status"] == "LICENSING_UNREACHABLE"
def test_get_health_error_status(self, mock_server_globals):
"""Test health status with error state."""
# Setup
expected_response = {"status": "ERROR"}
mock_server_globals['unimus_client'].get_health.return_value = expected_response
# Execute
result = unimus_get_health()
# Assert
assert result == expected_response
assert result["status"] == "ERROR"
def test_get_health_client_error(self, mock_server_globals):
"""Test health check with client error."""
# Setup
mock_server_globals['unimus_client'].get_health.side_effect = UnimusError("Connection failed")
# Execute & Assert
with pytest.raises(ValueError, match="Failed to get health status: Connection failed"):
unimus_get_health()
class TestUnimusGetDevices:
"""Test unimus_get_devices MCP tool."""
def test_get_devices_no_filters(self, mock_server_globals):
"""Test getting all devices without filters."""
# Setup
expected_devices = MockDataFactory.create_device_list(3)
mock_server_globals['unimus_client'].get_devices.return_value = expected_devices
# Execute
result = unimus_get_devices()
# Assert
assert result == expected_devices
assert len(result) == 3
mock_server_globals['unimus_client'].get_devices.assert_called_once_with({})
def test_get_devices_with_filters(self, mock_server_globals):
"""Test getting devices with filters."""
# Setup
filters = {"vendor": "Cisco", "managed": True}
cisco_devices = [MockDataFactory.create_device(vendor="Cisco")]
mock_server_globals['unimus_client'].get_devices.return_value = cisco_devices
# Execute
result = unimus_get_devices(filters)
# Assert
assert result == cisco_devices
assert len(result) == 1
assert result[0]["vendor"] == "Cisco"
mock_server_globals['unimus_client'].get_devices.assert_called_once_with(filters)
def test_get_devices_empty_filters(self, mock_server_globals):
"""Test getting devices with empty filters dict."""
# Setup
expected_devices = MockDataFactory.create_device_list(2)
mock_server_globals['unimus_client'].get_devices.return_value = expected_devices
# Execute
result = unimus_get_devices({})
# Assert
assert result == expected_devices
mock_server_globals['unimus_client'].get_devices.assert_called_once_with({})
def test_get_devices_client_error(self, mock_server_globals):
"""Test get devices with client error."""
# Setup
mock_server_globals['unimus_client'].get_devices.side_effect = UnimusError("API error")
# Execute & Assert
with pytest.raises(ValueError, match="Failed to get devices: API error"):
unimus_get_devices()
class TestUnimusGetDeviceById:
"""Test unimus_get_device_by_id MCP tool."""
def test_get_device_by_id_basic(self, mock_server_globals):
"""Test getting device by ID with default parameters."""
# Setup
device_id = 123
expected_device = MockDataFactory.create_device(device_id=device_id)
mock_server_globals['unimus_client'].get_device_by_id.return_value = expected_device
# Execute
result = unimus_get_device_by_id(device_id)
# Assert
assert result == expected_device
assert result["id"] == device_id
mock_server_globals['unimus_client'].get_device_by_id.assert_called_once_with(
device_id, None, None, False
)
def test_get_device_by_id_with_connections(self, mock_server_globals):
"""Test getting device by ID with connections."""
# Setup
device_id = 123
expected_device = MockDataFactory.create_device_with_connections(device_id=device_id)
mock_server_globals['unimus_client'].get_device_by_id.return_value = expected_device
# Execute
result = unimus_get_device_by_id(device_id, include_connections=True)
# Assert
assert result == expected_device
assert "connections" in result
mock_server_globals['unimus_client'].get_device_by_id.assert_called_once_with(
device_id, True, None, False
)
def test_get_device_by_id_with_attributes(self, mock_server_globals):
"""Test getting device by ID with specific attributes."""
# Setup
device_id = 123
attributes = ["schedule", "connections"]
expected_device = MockDataFactory.create_device_with_schedule(device_id=device_id)
mock_server_globals['unimus_client'].get_device_by_id.return_value = expected_device
# Execute
result = unimus_get_device_by_id(device_id, include_attributes=attributes)
# Assert
assert result == expected_device
mock_server_globals['unimus_client'].get_device_by_id.assert_called_once_with(
device_id, None, attributes, False
)
def test_get_device_by_id_with_metadata_enrichment(self, mock_server_globals):
"""Test getting device by ID with metadata enrichment."""
# Setup
device_id = 123
expected_device = MockDataFactory.create_device(device_id=device_id)
expected_device.update(MockDataFactory.create_enriched_metadata(device_id=device_id))
mock_server_globals['unimus_client'].get_device_by_id.return_value = expected_device
# Execute
result = unimus_get_device_by_id(device_id, enrich_metadata=True)
# Assert
assert result == expected_device
assert "backupAge" in result
assert "deviceHealth" in result
mock_server_globals['unimus_client'].get_device_by_id.assert_called_once_with(
device_id, None, None, True
)
def test_get_device_by_id_not_found(self, mock_server_globals):
"""Test getting device by ID when device not found."""
# Setup
device_id = 999
mock_server_globals['unimus_client'].get_device_by_id.side_effect = UnimusNotFoundError("Device not found")
# Execute & Assert
with pytest.raises(ValueError, match="Device with ID 999 not found"):
unimus_get_device_by_id(device_id)
def test_get_device_by_id_validation_error(self, mock_server_globals):
"""Test getting device by ID with validation error."""
# Setup
device_id = -1
mock_server_globals['unimus_client'].get_device_by_id.side_effect = UnimusValidationError("Invalid ID")
# Execute & Assert
with pytest.raises(ValueError, match="Invalid device ID: Invalid ID"):
unimus_get_device_by_id(device_id)
def test_get_device_by_id_client_error(self, mock_server_globals):
"""Test getting device by ID with general client error."""
# Setup
device_id = 123
mock_server_globals['unimus_client'].get_device_by_id.side_effect = UnimusError("Server error")
# Execute & Assert
with pytest.raises(ValueError, match="Failed to get device: Server error"):
unimus_get_device_by_id(device_id)
class TestUnimusGetDeviceByAddress:
"""Test unimus_get_device_by_address MCP tool."""
def test_get_device_by_address_basic(self, mock_server_globals):
"""Test getting device by address."""
# Setup
address = "192.168.1.1"
expected_device = MockDataFactory.create_device(address=address)
mock_server_globals['unimus_client'].get_device_by_address.return_value = expected_device
# Execute
result = unimus_get_device_by_address(address)
# Assert
assert result == expected_device
assert result["address"] == address
mock_server_globals['unimus_client'].get_device_by_address.assert_called_once_with(address, None)
def test_get_device_by_address_with_zone(self, mock_server_globals):
"""Test getting device by address with zone ID."""
# Setup
address = "router.example.com"
zone_id = "2"
expected_device = MockDataFactory.create_device(address=address, zoneId=zone_id)
mock_server_globals['unimus_client'].get_device_by_address.return_value = expected_device
# Execute
result = unimus_get_device_by_address(address, zone_id)
# Assert
assert result == expected_device
assert result["address"] == address
assert result["zoneId"] == zone_id
mock_server_globals['unimus_client'].get_device_by_address.assert_called_once_with(address, zone_id)
def test_get_device_by_address_not_found(self, mock_server_globals):
"""Test getting device by address when not found."""
# Setup
address = "nonexistent.device.com"
mock_server_globals['unimus_client'].get_device_by_address.side_effect = UnimusNotFoundError("Device not found")
# Execute & Assert
with pytest.raises(ValueError, match="Device with address 'nonexistent.device.com' not found"):
unimus_get_device_by_address(address)
def test_get_device_by_address_not_found_with_zone(self, mock_server_globals):
"""Test getting device by address with zone when not found."""
# Setup
address = "192.168.1.100"
zone_id = "3"
mock_server_globals['unimus_client'].get_device_by_address.side_effect = UnimusNotFoundError("Device not found")
# Execute & Assert
with pytest.raises(ValueError, match="Device with address '192.168.1.100' in zone 3 not found"):
unimus_get_device_by_address(address, zone_id)
def test_get_device_by_address_client_error(self, mock_server_globals):
"""Test getting device by address with client error."""
# Setup
address = "192.168.1.1"
mock_server_globals['unimus_client'].get_device_by_address.side_effect = UnimusError("API error")
# Execute & Assert
with pytest.raises(ValueError, match="Failed to get device: API error"):
unimus_get_device_by_address(address)
class TestUnimusGetDeviceBackups:
"""Test unimus_get_device_backups MCP tool."""
def test_get_device_backups_unlimited(self, mock_server_globals):
"""Test getting all device backups without limit."""
# Setup
device_id = 123
expected_backups = MockDataFactory.create_backup_list(device_id=device_id, count=5)
mock_server_globals['unimus_client'].get_device_backups.return_value = expected_backups
# Execute
result = unimus_get_device_backups(device_id)
# Assert
assert result == expected_backups
assert len(result) == 5
assert all(backup["deviceId"] == device_id for backup in result)
mock_server_globals['unimus_client'].get_device_backups.assert_called_once_with(device_id, None)
def test_get_device_backups_with_limit(self, mock_server_globals):
"""Test getting device backups with limit."""
# Setup
device_id = 123
limit = 3
expected_backups = MockDataFactory.create_backup_list(device_id=device_id, count=limit)
mock_server_globals['unimus_client'].get_device_backups.return_value = expected_backups
# Execute
result = unimus_get_device_backups(device_id, limit)
# Assert
assert result == expected_backups
assert len(result) == limit
mock_server_globals['unimus_client'].get_device_backups.assert_called_once_with(device_id, limit)
def test_get_device_backups_not_found(self, mock_server_globals):
"""Test getting backups for non-existent device."""
# Setup
device_id = 999
mock_server_globals['unimus_client'].get_device_backups.side_effect = UnimusNotFoundError("Device not found")
# Execute & Assert
with pytest.raises(ValueError, match="Device with ID 999 not found"):
unimus_get_device_backups(device_id)
def test_get_device_backups_client_error(self, mock_server_globals):
"""Test getting device backups with client error."""
# Setup
device_id = 123
mock_server_globals['unimus_client'].get_device_backups.side_effect = UnimusError("API error")
# Execute & Assert
with pytest.raises(ValueError, match="Failed to get device backups: API error"):
unimus_get_device_backups(device_id)
class TestUnimusGetBackupById:
"""Test unimus_get_backup_by_id MCP tool."""
def test_get_backup_by_id_success(self, mock_server_globals):
"""Test getting backup by ID successfully."""
# Setup
backup_id = 456
expected_backup = MockDataFactory.create_backup(backup_id=backup_id)
mock_server_globals['unimus_client'].get_backup_by_id.return_value = expected_backup
# Execute
result = unimus_get_backup_by_id(backup_id)
# Assert
assert result == expected_backup
assert result["id"] == backup_id
mock_server_globals['unimus_client'].get_backup_by_id.assert_called_once_with(backup_id)
def test_get_backup_by_id_not_found(self, mock_server_globals):
"""Test getting backup by ID when not found."""
# Setup
backup_id = 999
mock_server_globals['unimus_client'].get_backup_by_id.side_effect = UnimusNotFoundError("Backup not found")
# Execute & Assert
with pytest.raises(ValueError, match="Backup with ID 999 not found"):
unimus_get_backup_by_id(backup_id)
def test_get_backup_by_id_client_error(self, mock_server_globals):
"""Test getting backup by ID with client error."""
# Setup
backup_id = 456
mock_server_globals['unimus_client'].get_backup_by_id.side_effect = UnimusError("API error")
# Execute & Assert
with pytest.raises(ValueError, match="Failed to get backup: API error"):
unimus_get_backup_by_id(backup_id)
class TestUnimusSearchBackupContent:
"""Test unimus_search_backup_content MCP tool."""
def test_search_backup_content_basic(self, mock_server_globals):
"""Test basic backup content search."""
# Setup
pattern = "interface"
expected_results = [
{
"deviceId": 123,
"backupId": 1,
"lineNumber": 5,
"content": "interface GigabitEthernet0/0",
"match": "interface"
}
]
mock_server_globals['unimus_client'].search_backup_content.return_value = expected_results
# Execute
result = unimus_search_backup_content(pattern)
# Assert
assert result == expected_results
assert len(result) == 1
assert result[0]["match"] == "interface"
mock_server_globals['unimus_client'].search_backup_content.assert_called_once_with(
pattern=pattern, device_filters=None, context_lines=2, limit=None, since=None, until=None
)
def test_search_backup_content_with_options(self, mock_server_globals):
"""Test backup content search with additional options."""
# Setup
pattern = r"ip\s+address"
device_filters = {"type": "IOS"}
context_lines = 3
limit = 10
expected_results = []
mock_server_globals['unimus_client'].search_backup_content.return_value = expected_results
# Execute
result = unimus_search_backup_content(
pattern, device_filters=device_filters, context_lines=context_lines, limit=limit
)
# Assert
assert result == expected_results
mock_server_globals['unimus_client'].search_backup_content.assert_called_once_with(
pattern=pattern, device_filters=device_filters, context_lines=context_lines, limit=limit, since=None, until=None
)
def test_search_backup_content_client_error(self, mock_server_globals):
"""Test backup content search with client error."""
# Setup
pattern = "test"
mock_server_globals['unimus_client'].search_backup_content.side_effect = UnimusError("Search failed")
# Execute & Assert
with pytest.raises(ValueError, match="Failed to search backup content: Search failed"):
unimus_search_backup_content(pattern)
# Continue with remaining MCP tools...
class TestUnimusGetDeviceRelationships:
"""Test unimus_get_device_relationships MCP tool."""
def test_get_device_relationships_success(self, mock_server_globals):
"""Test getting device relationships successfully."""
# Setup
device_id = 123
expected_relationships = MockDataFactory.create_device_relationships(device_id=device_id)
mock_server_globals['unimus_client'].get_device_relationships.return_value = expected_relationships
# Execute
result = unimus_get_device_relationships(device_id)
# Assert
assert result == expected_relationships
assert result["deviceId"] == device_id
assert "networkNeighbors" in result
assert "relationshipSummary" in result
mock_server_globals['unimus_client'].get_device_relationships.assert_called_once_with(
device_id=device_id, include_network_neighbors=True, include_zone_peers=True, include_connection_analysis=True
)
def test_get_device_relationships_with_scope(self, mock_server_globals):
"""Test getting device relationships with specific options."""
# Setup
device_id = 123
expected_relationships = MockDataFactory.create_device_relationships(device_id=device_id)
mock_server_globals['unimus_client'].get_device_relationships.return_value = expected_relationships
# Execute
result = unimus_get_device_relationships(device_id, include_network_neighbors=False, include_zone_peers=True)
# Assert
assert result == expected_relationships
mock_server_globals['unimus_client'].get_device_relationships.assert_called_once_with(
device_id=device_id, include_network_neighbors=False, include_zone_peers=True, include_connection_analysis=True
)
def test_get_device_relationships_not_found(self, mock_server_globals):
"""Test getting device relationships when device not found."""
# Setup
device_id = 999
mock_server_globals['unimus_client'].get_device_relationships.side_effect = UnimusNotFoundError("Device not found")
# Execute & Assert
with pytest.raises(ValueError, match="Device with ID 999 not found"):
unimus_get_device_relationships(device_id)
class TestUnimusGetNetworkTopologyAnalysis:
"""Test unimus_get_network_topology_analysis MCP tool."""
def test_get_network_topology_analysis_success(self, mock_server_globals):
"""Test getting network topology analysis successfully."""
# Setup
expected_analysis = MockDataFactory.create_topology_analysis()
mock_server_globals['unimus_client'].get_network_topology_analysis.return_value = expected_analysis
# Execute
result = unimus_get_network_topology_analysis()
# Assert
assert result == expected_analysis
assert "networkScope" in result
assert "deviceClusters" in result
assert "topologyInsights" in result
mock_server_globals['unimus_client'].get_network_topology_analysis.assert_called_once_with(
zone_id=None, include_clusters=True, include_security_analysis=True
)
def test_get_network_topology_analysis_with_filters(self, mock_server_globals):
"""Test getting network topology analysis with specific options."""
# Setup
zone_id = "zone-1"
expected_analysis = MockDataFactory.create_topology_analysis()
mock_server_globals['unimus_client'].get_network_topology_analysis.return_value = expected_analysis
# Execute
result = unimus_get_network_topology_analysis(zone_id=zone_id, include_clusters=False)
# Assert
assert result == expected_analysis
mock_server_globals['unimus_client'].get_network_topology_analysis.assert_called_once_with(
zone_id=zone_id, include_clusters=False, include_security_analysis=True
)
def test_get_network_topology_analysis_client_error(self, mock_server_globals):
"""Test getting network topology analysis with client error."""
# Setup
mock_server_globals['unimus_client'].get_network_topology_analysis.side_effect = UnimusError("Analysis failed")
# Execute & Assert
with pytest.raises(ValueError, match="Failed to analyze network topology: Analysis failed"):
unimus_get_network_topology_analysis()
# Additional test classes for remaining MCP tools would follow the same pattern...
# For brevity, I'm including the main structure. The remaining tools would be:
# - TestUnimusGetDeviceSchedules
# - TestUnimusGetScheduleById
# - TestUnimusFindDevicesWithChanges
# - TestUnimusGetBackupDiff
@pytest.mark.parametrize("tool_name,client_method,expected_exception", [
("unimus_get_health", "get_health", UnimusAuthenticationError),
("unimus_get_devices", "get_devices", UnimusAuthenticationError),
("unimus_get_device_by_id", "get_device_by_id", UnimusAuthenticationError),
])
def test_authentication_errors(tool_name, client_method, expected_exception, mock_server_globals):
"""Test that authentication errors are properly handled across all tools."""
# Setup
tool_func = globals()[tool_name]
getattr(mock_server_globals['unimus_client'], client_method).side_effect = expected_exception("Auth failed")
# Execute & Assert
with pytest.raises(ValueError, match="Auth failed|Failed to"):
if tool_name == "unimus_get_device_by_id":
tool_func(123)
else:
tool_func()
@pytest.mark.integration
class TestMCPToolsIntegration:
"""Integration tests for MCP tools with realistic scenarios."""
def test_device_workflow_integration(self, mock_server_globals):
"""Test complete device workflow: get devices -> get specific device -> get backups."""
# Setup devices list
devices = MockDataFactory.create_device_list(3)
mock_server_globals['unimus_client'].get_devices.return_value = devices
# Setup specific device
device_id = devices[0]["id"]
device_detail = MockDataFactory.create_device_with_connections(device_id=device_id)
mock_server_globals['unimus_client'].get_device_by_id.return_value = device_detail
# Setup device backups
backups = MockDataFactory.create_backup_list(device_id=device_id)
mock_server_globals['unimus_client'].get_device_backups.return_value = backups
# Execute workflow
all_devices = unimus_get_devices()
first_device = unimus_get_device_by_id(device_id, include_connections=True)
device_backups = unimus_get_device_backups(device_id)
# Assert workflow
assert len(all_devices) == 3
assert first_device["id"] == device_id
assert "connections" in first_device
assert len(device_backups) == 5
assert all(backup["deviceId"] == device_id for backup in device_backups)
def test_search_and_analysis_workflow(self, mock_server_globals):
"""Test search and analysis workflow."""
# Setup
device_id = 123
search_results = [{"deviceId": device_id, "lineNumber": 10, "content": "interface GigabitEthernet0/0"}]
relationships = MockDataFactory.create_device_relationships(device_id=device_id)
topology = MockDataFactory.create_topology_analysis()
mock_server_globals['unimus_client'].search_backup_content.return_value = search_results
mock_server_globals['unimus_client'].get_device_relationships.return_value = relationships
mock_server_globals['unimus_client'].get_network_topology_analysis.return_value = topology
# Execute workflow
search_result = unimus_search_backup_content(device_id, "interface")
device_relations = unimus_get_device_relationships(device_id)
network_topology = unimus_get_network_topology_analysis()
# Assert workflow
assert len(search_result) == 1
assert search_result[0]["deviceId"] == device_id
assert device_relations["deviceId"] == device_id
assert "networkScope" in network_topology
assert "deviceClusters" in network_topology
