"""
Mock data factories for consistent test data across the test suite.
Provides realistic mock data that matches the Unimus API v2 specification.
"""
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
import uuid
class MockDataFactory:
"""Factory for creating consistent mock data for testing."""
@staticmethod
def create_device(
device_id: int = 1,
address: str = "10.0.0.1",
description: str = "Test Device",
vendor: str = "Cisco",
device_type: str = "IOS",
managed: bool = True,
**kwargs
) -> Dict[str, Any]:
"""Create a mock device object."""
base_device = {
"id": device_id,
"uuid": str(uuid.uuid4()),
"address": address,
"description": description,
"managed": managed,
"vendor": vendor,
"type": device_type,
"model": "ISR4321",
"lastJobStatus": "SUCCESSFUL",
"createTime": int((datetime.now() - timedelta(days=30)).timestamp() * 1000),
"zoneId": "1"
}
base_device.update(kwargs)
return base_device
@staticmethod
def create_device_list(count: int = 5) -> List[Dict[str, Any]]:
"""Create a list of mock devices with different characteristics."""
vendors = ["Cisco", "MikroTik", "HP", "Ubiquiti", "Juniper"]
types = ["IOS", "RouterOS", "ArubaOS-CX", "EdgeOS", "JunOS"]
devices = []
for i in range(count):
vendor = vendors[i % len(vendors)]
device_type = types[i % len(types)]
devices.append(MockDataFactory.create_device(
device_id=i + 1,
address=f"10.0.0.{i + 1}",
description=f"Test {vendor} Device {i + 1}",
vendor=vendor,
device_type=device_type,
managed=i % 10 != 0 # Every 10th device is unmanaged
))
return devices
@staticmethod
def create_backup(
backup_id: int = 1,
device_id: int = 1,
content: str = "hostname test-device\n!",
**kwargs
) -> Dict[str, Any]:
"""Create a mock backup object."""
base_backup = {
"id": backup_id,
"deviceId": device_id,
"createTime": int((datetime.now() - timedelta(days=1)).timestamp() * 1000),
"content": content,
"size": len(content.encode('utf-8')),
"successful": True
}
base_backup.update(kwargs)
return base_backup
@staticmethod
def create_backup_list(device_id: int = 1, count: int = 5) -> List[Dict[str, Any]]:
"""Create a list of mock backups for a device."""
backups = []
for i in range(count):
backup_time = datetime.now() - timedelta(days=i)
content = f"hostname test-device\nversion {i + 1}\n!\ninterface GigabitEthernet0/0/0\n ip address 10.0.0.{device_id} 255.255.255.0\n!"
backups.append(MockDataFactory.create_backup(
backup_id=i + 1,
device_id=device_id,
content=content,
createTime=int(backup_time.timestamp() * 1000)
))
return backups
@staticmethod
def create_schedule(
schedule_id: int = 1,
name: str = "Daily Backup",
**kwargs
) -> Dict[str, Any]:
"""Create a mock schedule object."""
base_schedule = {
"id": schedule_id,
"name": name,
"description": f"Automated {name.lower()} schedule",
"enabled": True,
"cronExpression": "0 2 * * *",
"createTime": int((datetime.now() - timedelta(days=7)).timestamp() * 1000),
"lastRunTime": int((datetime.now() - timedelta(hours=22)).timestamp() * 1000),
"nextRunTime": int((datetime.now() + timedelta(hours=2)).timestamp() * 1000)
}
base_schedule.update(kwargs)
return base_schedule
@staticmethod
def create_schedule_list(count: int = 3) -> List[Dict[str, Any]]:
"""Create a list of mock schedules."""
schedule_names = ["Daily Backup", "Weekly Config Check", "Monthly Report"]
cron_expressions = ["0 2 * * *", "0 6 * * 0", "0 8 1 * *"]
schedules = []
for i in range(min(count, len(schedule_names))):
schedules.append(MockDataFactory.create_schedule(
schedule_id=i + 1,
name=schedule_names[i],
cronExpression=cron_expressions[i]
))
return schedules
@staticmethod
def create_health_status(status: str = "OK") -> Dict[str, str]:
"""Create a mock health status response."""
return {"status": status}
@staticmethod
def create_device_with_connections(device_id: int = 1) -> Dict[str, Any]:
"""Create a mock device with connection information."""
device = MockDataFactory.create_device(device_id=device_id)
device["connections"] = [
{
"id": 1,
"name": "SSH Connection",
"type": "SSH",
"port": 22,
"enabled": True
},
{
"id": 2,
"name": "SNMP Connection",
"type": "SNMP",
"port": 161,
"enabled": True
}
]
return device
@staticmethod
def create_device_with_schedule(device_id: int = 1, schedule_id: int = 1) -> Dict[str, Any]:
"""Create a mock device with schedule information."""
device = MockDataFactory.create_device(device_id=device_id)
device["schedule"] = MockDataFactory.create_schedule(schedule_id=schedule_id)
return device
@staticmethod
def create_enriched_metadata(device_id: int = 1) -> Dict[str, Any]:
"""Create mock enriched metadata for a device."""
return {
"backupAge": 1,
"backupFreshness": "fresh",
"deviceHealth": "healthy",
"lastBackupStatus": "successful",
"connectionCount": 2,
"connectionTypes": ["SSH", "SNMP"],
"managementComplexity": "standard",
"configurationStability": "stable",
"deviceLifecycle": "active",
"connectivityRisk": "low",
"backupFrequency": "daily",
"enrichmentTimestamp": datetime.now().isoformat()
}
@staticmethod
def create_device_relationships(device_id: int = 1) -> Dict[str, Any]:
"""Create mock device relationships."""
return {
"deviceId": device_id,
"networkNeighbors": [
{"deviceId": 2, "address": "10.0.0.2", "subnet": "10.0.0.0/24", "distance": 1},
{"deviceId": 3, "address": "10.0.0.3", "subnet": "10.0.0.0/24", "distance": 1}
],
"zoneMembers": [
{"deviceId": 2, "vendor": "Cisco", "type": "IOS"},
{"deviceId": 4, "vendor": "HP", "type": "ArubaOS-CX"}
],
"vendorSiblings": [
{"deviceId": 5, "vendor": "Cisco", "type": "IOS", "similarity": 0.9}
],
"relationshipSummary": {
"totalRelatedDevices": 4,
"networkNeighborCount": 2,
"zoneDeviceCount": 2,
"vendorSiblingCount": 1
}
}
@staticmethod
def create_topology_analysis() -> Dict[str, Any]:
"""Create mock network topology analysis."""
return {
"networkScope": {
"totalDevices": 17,
"managedDevices": 16,
"analyzedSubnets": ["10.0.0.0/24", "192.168.1.0/24"]
},
"deviceClusters": {
"networkBased": [
{
"subnet": "10.0.0.0/24",
"deviceCount": 12,
"devices": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
},
{
"subnet": "192.168.1.0/24",
"deviceCount": 5,
"devices": [13, 14, 15, 16, 17]
}
],
"vendorBased": [
{"vendor": "Cisco", "deviceCount": 8, "devices": [1, 2, 5, 6, 9, 10, 13, 14]},
{"vendor": "HP", "deviceCount": 5, "devices": [3, 7, 11, 15, 16]},
{"vendor": "MikroTik", "deviceCount": 4, "devices": [4, 8, 12, 17]}
]
},
"topologyInsights": {
"deviceRoles": {
"infrastructure_hub": 3,
"network_node": 8,
"endpoint": 4,
"standalone": 2
},
"connectivityPatterns": {
"highlyConnected": [1, 2, 13],
"moderatelyConnected": [3, 4, 5, 6, 7, 8],
"isolatedDevices": [17]
}
},
"securityAssessment": {
"connectionMethods": {
"SSH": {"count": 15, "percentage": 88.2},
"TELNET": {"count": 2, "percentage": 11.8}
},
"riskProfile": {
"low": 15,
"medium": 1,
"high": 1
}
}
}
@staticmethod
def create_api_error(
error_type: str = "UnimusError",
message: str = "API request failed",
status_code: int = 500
) -> Dict[str, Any]:
"""Create mock API error for testing error handling."""
return {
"error": error_type,
"message": message,
"statusCode": status_code,
"timestamp": datetime.now().isoformat()
}
