Spaces:
Running
Running
File size: 5,564 Bytes
74f2bea |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
#!/usr/bin/env python3
"""
Test NetBox integration
Validates NetBox client and pipeline integration
"""
import logging
from pathlib import Path
from agent.netbox_client import NetBoxClient, NetBoxConfig
from agent.pipeline_engine import OvergrowthPipeline, NetworkIntent
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def test_netbox_mock_mode():
"""Test NetBox client in mock mode (no credentials)"""
print("\n=== Test 1: NetBox Mock Mode ===")
client = NetBoxClient()
assert client.mock_mode, "Should be in mock mode without credentials"
# Test site operations
site = client.create_site("test-site", "Test Site")
assert site["name"] == "test-site"
logger.info(f"✓ Created mock site: {site}")
# Test VLAN operations
vlan = client.create_vlan(vid=10, name="Management", description="Mgmt network")
assert vlan["vid"] == 10
logger.info(f"✓ Created mock VLAN: {vlan}")
# Test prefix operations
prefix = client.create_prefix("10.0.10.0/24", description="Mgmt subnet")
assert prefix["prefix"] == "10.0.10.0/24"
logger.info(f"✓ Created mock prefix: {prefix}")
print("✓ All mock mode tests passed\n")
def test_pipeline_with_netbox():
"""Test pipeline with NetBox integration"""
print("\n=== Test 2: Pipeline with NetBox ===")
# Create pipeline with NetBox enabled
pipeline = OvergrowthPipeline(use_netbox=True)
# Should be in mock mode without credentials
assert pipeline.netbox.mock_mode, "Should be in mock mode"
logger.info("✓ Pipeline created with NetBox client")
# Run through consultation and SoT generation
intent = NetworkIntent(
description="Small office network with 3 VLANs",
business_requirements=["Secure guest WiFi", "Isolated IoT devices"],
constraints=["Budget under $2000", "Easy management"],
budget="$2000",
timeline="2 weeks"
)
# Generate source of truth
model = pipeline.stage2_generate_sot(intent)
assert model.name is not None
assert len(model.vlans) > 0 or len(model.subnets) > 0
logger.info(f"✓ Generated network model: {model.name}")
logger.info(f" - VLANs: {len(model.vlans)}")
logger.info(f" - Subnets: {len(model.subnets)}")
logger.info(f" - Services: {model.services}")
print("✓ Pipeline integration test passed\n")
def test_network_sync():
"""Test syncing a complete network model"""
print("\n=== Test 3: Network Model Sync ===")
client = NetBoxClient()
# Create a sample network model
network_model = {
"name": "demo-network",
"description": "Demo network for testing",
"vlans": [
{"id": 10, "name": "Management", "purpose": "Network management"},
{"id": 20, "name": "Users", "purpose": "Employee workstations"},
{"id": 30, "name": "Guest", "purpose": "Guest WiFi"}
],
"subnets": [
{"network": "10.0.10.0/24", "vlan": 10, "purpose": "Management"},
{"network": "10.0.20.0/24", "vlan": 20, "purpose": "Users"},
{"network": "10.0.30.0/24", "vlan": 30, "purpose": "Guest WiFi"}
],
"devices": [
{"name": "core-sw-01", "model": "Cisco 9300", "role": "core"},
{"name": "access-sw-01", "model": "Cisco 2960", "role": "access"}
]
}
summary = client.sync_network_model(network_model)
logger.info(f"✓ Sync summary:")
logger.info(f" - Sites: {summary['sites']}")
logger.info(f" - Devices: {summary['devices']}")
logger.info(f" - VLANs: {summary['vlans']}")
logger.info(f" - Prefixes: {summary['prefixes']}")
if summary['errors']:
logger.warning(f" - Errors: {summary['errors']}")
print("✓ Network sync test passed\n")
def test_real_netbox_connection():
"""Test connection to a real NetBox instance (if configured)"""
print("\n=== Test 4: Real NetBox Connection (Optional) ===")
import os
if not os.getenv("NETBOX_URL") and not os.getenv("NAUTOBOT_URL"):
print("⊘ Skipping - No NetBox credentials configured")
print(" To test with real NetBox, set:")
print(" export NETBOX_URL=http://netbox.example.com")
print(" export NETBOX_TOKEN=your_api_token")
return
try:
client = NetBoxClient()
if client.mock_mode:
print("⊘ NetBox configured but connection failed - check credentials")
return
# Test basic operations
site = client.get_site("test-site")
logger.info(f"✓ Connected to NetBox at {client.config.url}")
if site:
logger.info(f" Found existing site: {site['name']}")
else:
logger.info(" No test site found")
print("✓ Real NetBox connection test passed\n")
except Exception as e:
logger.error(f"✗ NetBox connection failed: {e}")
if __name__ == "__main__":
print("\n" + "="*60)
print("NetBox Integration Test Suite")
print("="*60)
try:
test_netbox_mock_mode()
test_pipeline_with_netbox()
test_network_sync()
test_real_netbox_connection()
print("\n" + "="*60)
print("✓ All tests completed successfully!")
print("="*60 + "\n")
except Exception as e:
print(f"\n✗ Test failed: {e}")
import traceback
traceback.print_exc()
exit(1)
|