conftest.py•14.3 kB
"""Pytest configuration and shared fixtures for NetBox MCP Server tests.
This module provides comprehensive test fixtures and configuration
for testing the NetBox MCP Server. It demonstrates best practices
for testing MCP servers and infrastructure applications.
"""
import asyncio
import json
import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from unittest.mock import AsyncMock, Mock, patch
import pytest
from mcp.types import TextContent
from src.vault_client import VaultClient
from src.netbox_client import NetBoxClient
from src.state_confidence import StateConfidenceClient
# Configure test logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
@pytest.fixture(scope="session")
def event_loop():
"""Create an instance of the default event loop for the test session."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
def sample_netbox_device():
"""Sample NetBox device data for testing."""
return {
"id": 1,
"name": "web-server-01",
"display": "web-server-01",
"device_type": {"slug": "server", "display": "Server"},
"device_role": {"slug": "web-server", "display": "Web Server"},
"site": {"slug": "datacenter-1", "display": "Datacenter 1"},
"rack": {"name": "R01", "display": "R01"},
"position": 10,
"face": "front",
"status": {"value": "active", "label": "Active"},
"primary_ip4": {
"id": 1,
"address": "192.168.1.10/24",
"display": "192.168.1.10/24"
},
"primary_ip6": None,
"serial": "ABC123456789",
"asset_tag": "WS-001",
"created": "2025-01-01T00:00:00Z",
"last_updated": "2025-01-01T12:00:00Z",
"custom_fields": {
"environment": "production",
"owner": "platform-team"
},
"tags": [
{"name": "production", "slug": "production"},
{"name": "web-tier", "slug": "web-tier"}
]
}
@pytest.fixture
def sample_netbox_vm():
"""Sample NetBox virtual machine data for testing."""
return {
"id": 2,
"name": "app-server-01",
"display": "app-server-01",
"role": {"slug": "app-server", "display": "Application Server"},
"status": {"value": "active", "label": "Active"},
"cluster": {"name": "k8s-cluster-1", "display": "K8s Cluster 1"},
"site": {"slug": "datacenter-1", "display": "Datacenter 1"},
"primary_ip4": {
"id": 2,
"address": "192.168.1.20/24",
"display": "192.168.1.20/24"
},
"vcpus": 4,
"memory": 8192,
"disk": 100,
"created": "2025-01-01T00:00:00Z",
"last_updated": "2025-01-01T12:00:00Z",
"custom_fields": {
"environment": "staging",
"owner": "app-team"
},
"tags": [
{"name": "staging", "slug": "staging"},
{"name": "app-tier", "slug": "app-tier"}
]
}
@pytest.fixture
def sample_netbox_ip():
"""Sample NetBox IP address data for testing."""
return {
"id": 3,
"address": "192.168.1.30/24",
"display": "192.168.1.30/24",
"family": {"value": 4, "label": "IPv4"},
"status": {"value": "active", "label": "Active"},
"role": {"value": "secondary", "label": "Secondary"},
"assigned_object_type": "dcim.interface",
"assigned_object": {
"id": 1,
"name": "eth0",
"device": {"name": "web-server-01", "display": "web-server-01"}
},
"nat_inside": None,
"nat_outside": None,
"created": "2025-01-01T00:00:00Z",
"last_updated": "2025-01-01T12:00:00Z",
"custom_fields": {
"purpose": "application",
"monitoring": "enabled"
},
"tags": [
{"name": "application", "slug": "application"},
{"name": "monitored", "slug": "monitored"}
]
}
@pytest.fixture
def sample_netbox_vlan():
"""Sample NetBox VLAN data for testing."""
return {
"id": 4,
"vid": 100,
"name": "web-tier",
"display": "VLAN 100 (web-tier)",
"status": {"value": "active", "label": "Active"},
"site": {"slug": "datacenter-1", "display": "Datacenter 1"},
"group": {"name": "production", "display": "Production"},
"description": "Web tier VLAN for production servers",
"created": "2025-01-01T00:00:00Z",
"last_updated": "2025-01-01T12:00:00Z",
"custom_fields": {
"purpose": "web-tier",
"security_level": "dmz"
},
"tags": [
{"name": "production", "slug": "production"},
{"name": "web-tier", "slug": "web-tier"}
]
}
@pytest.fixture
def sample_certainty_score():
"""Sample state confidence certainty score for testing."""
return {
"certainty_score": 0.95,
"data_age_seconds": 300,
"last_verified": "2025-01-01T12:00:00Z",
"verification_method": "api_poll",
"source_reliability": 0.98,
"state_type": "device",
"state_id": "web-server-01"
}
@pytest.fixture
def mock_vault_client():
"""Mock Vault client with comprehensive test data."""
client = Mock(spec=VaultClient)
client._token = "test-vault-token-12345"
client._token_expiry = datetime.now(timezone.utc).timestamp() + 3600
client.authenticate.return_value = True
client.mint_netbox_token.return_value = "netbox-api-token-67890"
client.is_authenticated.return_value = True
# Add error scenarios
client.authenticate.side_effect = None # Can be overridden in tests
client.mint_netbox_token.side_effect = None # Can be overridden in tests
return client
@pytest.fixture
def mock_netbox_client(mock_vault_client, sample_netbox_device, sample_netbox_vm,
sample_netbox_ip, sample_netbox_vlan):
"""Mock NetBox client with comprehensive test data."""
client = Mock(spec=NetBoxClient)
client.vault_client = mock_vault_client
client.netbox_url = "http://test-netbox.example.com"
# Device methods
client.list_devices.return_value = [sample_netbox_device]
client.get_device.return_value = sample_netbox_device
client.search_devices.return_value = [sample_netbox_device]
# VM methods
client.list_virtual_machines.return_value = [sample_netbox_vm]
client.get_virtual_machine.return_value = sample_netbox_vm
client.list_vm_interfaces.return_value = [
{
"id": 1,
"name": "eth0",
"type": {"value": "virtual", "label": "Virtual"},
"enabled": True,
"mtu": 1500,
"mac_address": "00:11:22:33:44:55",
"ip_addresses": [sample_netbox_ip]
}
]
# IP methods
client.list_ip_addresses.return_value = [sample_netbox_ip]
client.get_ip_address.return_value = sample_netbox_ip
client.search_ip_addresses.return_value = [sample_netbox_ip]
# VLAN methods
client.list_vlans.return_value = [sample_netbox_vlan]
client.get_vlan.return_value = sample_netbox_vlan
client.list_vlan_ip_addresses.return_value = [sample_netbox_ip]
return client
@pytest.fixture
def mock_state_client(sample_certainty_score):
"""Mock state confidence client with comprehensive test data."""
client = Mock(spec=StateConfidenceClient)
client.get_certainty_score.return_value = sample_certainty_score
# Add error scenarios
client.get_certainty_score.side_effect = None # Can be overridden in tests
return client
@pytest.fixture
def mock_netbox_client_empty():
"""Mock NetBox client that returns empty results."""
client = Mock(spec=NetBoxClient)
client.list_devices.return_value = []
client.get_device.return_value = None
client.search_devices.return_value = []
client.list_virtual_machines.return_value = []
client.get_virtual_machine.return_value = None
client.list_vm_interfaces.return_value = []
client.list_ip_addresses.return_value = []
client.get_ip_address.return_value = None
client.search_ip_addresses.return_value = []
client.list_vlans.return_value = []
client.get_vlan.return_value = None
client.list_vlan_ip_addresses.return_value = []
return client
@pytest.fixture
def mock_netbox_client_error():
"""Mock NetBox client that raises exceptions."""
client = Mock(spec=NetBoxClient)
client.list_devices.side_effect = Exception("NetBox API error")
client.get_device.side_effect = Exception("NetBox API error")
client.search_devices.side_effect = Exception("NetBox API error")
client.list_virtual_machines.side_effect = Exception("NetBox API error")
client.get_virtual_machine.side_effect = Exception("NetBox API error")
client.list_vm_interfaces.side_effect = Exception("NetBox API error")
client.list_ip_addresses.side_effect = Exception("NetBox API error")
client.get_ip_address.side_effect = Exception("NetBox API error")
client.search_ip_addresses.side_effect = Exception("NetBox API error")
client.list_vlans.side_effect = Exception("NetBox API error")
client.get_vlan.side_effect = Exception("NetBox API error")
client.list_vlan_ip_addresses.side_effect = Exception("NetBox API error")
return client
@pytest.fixture
def mock_state_client_error():
"""Mock state confidence client that raises exceptions."""
client = Mock(spec=StateConfidenceClient)
client.get_certainty_score.side_effect = Exception("PostgreSQL connection error")
return client
@pytest.fixture
def sample_tool_arguments():
"""Sample tool arguments for testing."""
return {
"list_hosts": {"limit": 10, "include_certainty": True},
"get_host": {"hostname": "web-server-01", "include_certainty": True},
"search_hosts": {"query": "web", "limit": 5, "include_certainty": True},
"list_vms": {"limit": 10, "include_certainty": True},
"get_vm": {"hostname": "app-server-01", "include_certainty": True},
"list_vm_interfaces": {"vm_name": "app-server-01", "include_certainty": True},
"list_ips": {"limit": 10, "include_certainty": True},
"get_ip": {"ip_address": "192.168.1.30/24", "include_certainty": True},
"search_ips": {"query": "192.168.1", "limit": 5, "include_certainty": True},
"list_vlans": {"limit": 10, "include_certainty": True},
"get_vlan": {"vlan_id": 100, "include_certainty": True},
"list_vlan_ips": {"vlan_id": 100, "include_certainty": True}
}
@pytest.fixture
def invalid_tool_arguments():
"""Invalid tool arguments for testing error handling."""
return {
"get_host": {}, # Missing required hostname
"get_vm": {}, # Missing required hostname
"get_ip": {}, # Missing required ip_address
"get_vlan": {}, # Missing required vlan_id
"list_vm_interfaces": {}, # Missing required vm_name
"list_vlan_ips": {} # Missing required vlan_id
}
# Test data factories for property-based testing
@pytest.fixture
def device_factory():
"""Factory for creating test device data."""
def _create_device(**overrides):
base_device = {
"id": 1,
"name": "test-device",
"display": "test-device",
"device_type": {"slug": "server", "display": "Server"},
"device_role": {"slug": "server", "display": "Server"},
"site": {"slug": "test-site", "display": "Test Site"},
"status": {"value": "active", "label": "Active"},
"primary_ip4": {"address": "192.168.1.1/24", "display": "192.168.1.1/24"},
"created": "2025-01-01T00:00:00Z",
"last_updated": "2025-01-01T12:00:00Z"
}
base_device.update(overrides)
return base_device
return _create_device
@pytest.fixture
def vm_factory():
"""Factory for creating test VM data."""
def _create_vm(**overrides):
base_vm = {
"id": 1,
"name": "test-vm",
"display": "test-vm",
"role": {"slug": "server", "display": "Server"},
"status": {"value": "active", "label": "Active"},
"primary_ip4": {"address": "192.168.1.1/24", "display": "192.168.1.1/24"},
"vcpus": 2,
"memory": 4096,
"disk": 50,
"created": "2025-01-01T00:00:00Z",
"last_updated": "2025-01-01T12:00:00Z"
}
base_vm.update(overrides)
return base_vm
return _create_vm
# Performance testing fixtures
@pytest.fixture
def large_dataset():
"""Large dataset for performance testing."""
devices = []
for i in range(1000):
devices.append({
"id": i,
"name": f"device-{i:04d}",
"display": f"device-{i:04d}",
"device_type": {"slug": "server", "display": "Server"},
"device_role": {"slug": "server", "display": "Server"},
"status": {"value": "active", "label": "Active"},
"primary_ip4": {"address": f"192.168.1.{i % 254 + 1}/24"},
"created": "2025-01-01T00:00:00Z",
"last_updated": "2025-01-01T12:00:00Z"
})
return devices
# Async test helpers
@pytest.fixture
def async_mock():
"""Helper for creating async mocks."""
def _create_async_mock(*args, **kwargs):
mock = AsyncMock(*args, **kwargs)
return mock
return _create_async_mock
# Test markers for different test categories
def pytest_configure(config):
"""Configure pytest with custom markers."""
config.addinivalue_line("markers", "unit: Unit tests")
config.addinivalue_line("markers", "integration: Integration tests")
config.addinivalue_line("markers", "performance: Performance tests")
config.addinivalue_line("markers", "error_handling: Error handling tests")
config.addinivalue_line("markers", "edge_cases: Edge case tests")
config.addinivalue_line("markers", "slow: Slow running tests")