143 lines
4.9 KiB
Python
143 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify remote integration with Calejo Control Adapter
|
|
Tests discovery and interaction with remote mock services
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import time
|
|
import sys
|
|
|
|
def test_dashboard_health():
|
|
"""Test if dashboard is running"""
|
|
try:
|
|
response = requests.get("http://localhost:8080/health", timeout=5)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(f"✅ Dashboard health: {data}")
|
|
return True
|
|
else:
|
|
print(f"❌ Dashboard health check failed: {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ Dashboard health check error: {e}")
|
|
return False
|
|
|
|
def test_discovery_scan():
|
|
"""Test discovery scan functionality"""
|
|
try:
|
|
response = requests.post("http://localhost:8080/api/v1/dashboard/discovery/scan", timeout=10)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(f"✅ Discovery scan started: {data}")
|
|
return data.get('scan_id')
|
|
else:
|
|
print(f"❌ Discovery scan failed: {response.status_code}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"❌ Discovery scan error: {e}")
|
|
return None
|
|
|
|
def test_discovery_status():
|
|
"""Test discovery status"""
|
|
try:
|
|
response = requests.get("http://localhost:8080/api/v1/dashboard/discovery/status", timeout=5)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(f"✅ Discovery status: {data}")
|
|
return data
|
|
else:
|
|
print(f"❌ Discovery status failed: {response.status_code}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"❌ Discovery status error: {e}")
|
|
return None
|
|
|
|
def test_recent_discoveries():
|
|
"""Test recent discoveries endpoint"""
|
|
try:
|
|
response = requests.get("http://localhost:8080/api/v1/dashboard/discovery/recent", timeout=5)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(f"✅ Recent discoveries: Found {len(data.get('recent_endpoints', []))} endpoints")
|
|
|
|
# Show discovered endpoints
|
|
for endpoint in data.get('recent_endpoints', []):
|
|
print(f" - {endpoint.get('device_name')} ({endpoint.get('address')})")
|
|
|
|
return data
|
|
else:
|
|
print(f"❌ Recent discoveries failed: {response.status_code}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"❌ Recent discoveries error: {e}")
|
|
return None
|
|
|
|
def test_remote_services():
|
|
"""Test direct access to remote services"""
|
|
services = [
|
|
("Mock SCADA", "http://95.111.206.155:8083/health"),
|
|
("Mock Optimizer", "http://95.111.206.155:8084/health"),
|
|
("Existing API", "http://95.111.206.155:8080/health")
|
|
]
|
|
|
|
all_accessible = True
|
|
for name, url in services:
|
|
try:
|
|
response = requests.get(url, timeout=5)
|
|
if response.status_code == 200:
|
|
print(f"✅ {name}: ACCESSIBLE - {response.json()}")
|
|
else:
|
|
print(f"❌ {name}: NOT ACCESSIBLE - Status {response.status_code}")
|
|
all_accessible = False
|
|
except Exception as e:
|
|
print(f"❌ {name}: ERROR - {e}")
|
|
all_accessible = False
|
|
|
|
return all_accessible
|
|
|
|
def main():
|
|
print("🚀 Calejo Control Adapter - Remote Integration Test")
|
|
print("=" * 60)
|
|
|
|
# Test dashboard health
|
|
if not test_dashboard_health():
|
|
print("\n❌ Dashboard not accessible. Please start the dashboard first.")
|
|
sys.exit(1)
|
|
|
|
print("\n🔍 Testing remote service connectivity...")
|
|
remote_ok = test_remote_services()
|
|
|
|
print("\n📊 Testing discovery functionality...")
|
|
|
|
# Start discovery scan
|
|
scan_id = test_discovery_scan()
|
|
|
|
if scan_id:
|
|
print(f"\n⏳ Waiting for discovery scan to complete...")
|
|
time.sleep(5)
|
|
|
|
# Check discovery status
|
|
status = test_discovery_status()
|
|
|
|
# Check recent discoveries
|
|
discoveries = test_recent_discoveries()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("📋 TEST SUMMARY:")
|
|
print(f" Dashboard Health: ✅")
|
|
print(f" Remote Services: {'✅' if remote_ok else '❌'}")
|
|
print(f" Discovery Scan: {'✅' if scan_id else '❌'}")
|
|
print(f" Endpoints Found: {status.get('status', {}).get('total_discovered_endpoints', 0) if status else 0}")
|
|
|
|
if status and status.get('status', {}).get('total_discovered_endpoints', 0) >= 2:
|
|
print("\n🎉 SUCCESS: Remote integration test passed!")
|
|
print(" The system can discover and interact with remote services.")
|
|
else:
|
|
print("\n⚠️ WARNING: Some tests may have issues.")
|
|
else:
|
|
print("\n❌ Discovery scan failed. Check dashboard logs.")
|
|
|
|
if __name__ == "__main__":
|
|
main() |