198 lines
5.7 KiB
Python
Executable File
198 lines
5.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Host IP Detection Test Script
|
|
|
|
Tests the dynamic host IP detection functionality across different Docker environments.
|
|
This script can be run both inside and outside containers to validate the detection logic.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import asyncio
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
# Add session-manager to path for imports
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from host_ip_detector import HostIPDetector, get_host_ip, reset_host_ip_cache
|
|
|
|
# Set up logging
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def test_host_ip_detection():
|
|
"""Test host IP detection with various methods."""
|
|
print("🧪 Testing Docker Host IP Detection")
|
|
print("=" * 50)
|
|
|
|
detector = HostIPDetector()
|
|
|
|
# Test each detection method individually
|
|
methods = [
|
|
(
|
|
"Docker Internal (host.docker.internal)",
|
|
detector._detect_via_docker_internal,
|
|
),
|
|
("Gateway Environment Variables", detector._detect_via_gateway_env),
|
|
("Route Table (/proc/net/route)", detector._detect_via_route_table),
|
|
("Network Connection", detector._detect_via_network_connect),
|
|
("Common Gateways", detector._detect_via_common_gateways),
|
|
]
|
|
|
|
print("\n📋 Testing Individual Detection Methods:")
|
|
print("-" * 40)
|
|
|
|
working_methods = []
|
|
for name, method in methods:
|
|
try:
|
|
result = method()
|
|
if result:
|
|
print(f"✅ {name}: {result}")
|
|
working_methods.append((name, result))
|
|
else:
|
|
print(f"❌ {name}: No result")
|
|
except Exception as e:
|
|
print(f"⚠️ {name}: Error - {e}")
|
|
|
|
# Test the main detection function
|
|
print("\n🎯 Testing Main Detection Function:")
|
|
print("-" * 40)
|
|
|
|
try:
|
|
detected_ip = detector.detect_host_ip()
|
|
print(f"✅ Successfully detected host IP: {detected_ip}")
|
|
|
|
# Validate the IP
|
|
if detector._validate_ip(detected_ip):
|
|
print("✅ IP address validation passed")
|
|
else:
|
|
print("❌ IP address validation failed")
|
|
|
|
# Test connectivity to the detected IP
|
|
if detector._test_ip_connectivity(detected_ip):
|
|
print("✅ Connectivity test to detected IP passed")
|
|
else:
|
|
print("⚠️ Connectivity test to detected IP failed (may be normal)")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Main detection failed: {e}")
|
|
return False
|
|
|
|
# Test caching
|
|
print("\n📦 Testing Caching Behavior:")
|
|
print("-" * 40)
|
|
|
|
import time
|
|
|
|
start_time = time.time()
|
|
cached_ip = detector.detect_host_ip() # Should use cache
|
|
cache_time = time.time() - start_time
|
|
|
|
print(f"✅ Cached result: {cached_ip} (took {cache_time:.4f}s)")
|
|
|
|
# Test cache reset
|
|
reset_host_ip_cache()
|
|
new_detector = HostIPDetector()
|
|
fresh_ip = new_detector.detect_host_ip()
|
|
print(f"✅ Fresh detection after reset: {fresh_ip}")
|
|
|
|
# Test async version
|
|
print("\n⚡ Testing Async Version:")
|
|
print("-" * 40)
|
|
|
|
try:
|
|
async_ip = await detector.async_detect_host_ip()
|
|
print(f"✅ Async detection: {async_ip}")
|
|
except Exception as e:
|
|
print(f"❌ Async detection failed: {e}")
|
|
|
|
print("\n🎉 Host IP Detection Test Complete!")
|
|
return True
|
|
|
|
|
|
def test_environment_info():
|
|
"""Display environment information for debugging."""
|
|
print("🌍 Environment Information:")
|
|
print("-" * 30)
|
|
|
|
# Check if running in Docker
|
|
in_docker = Path("/.dockerenv").exists()
|
|
print(f"Docker container: {'Yes' if in_docker else 'No'}")
|
|
|
|
# Display relevant environment variables
|
|
env_vars = [
|
|
"DOCKER_HOST",
|
|
"HOST_IP",
|
|
"DOCKER_HOST_GATEWAY",
|
|
"GATEWAY",
|
|
]
|
|
|
|
print("Environment variables:")
|
|
for var in env_vars:
|
|
value = os.getenv(var, "Not set")
|
|
print(f" {var}: {value}")
|
|
|
|
# Check network interfaces (if available)
|
|
try:
|
|
import socket
|
|
|
|
hostname = socket.gethostname()
|
|
print(f"Hostname: {hostname}")
|
|
|
|
try:
|
|
host_ip = socket.gethostbyname(hostname)
|
|
print(f"Hostname IP: {host_ip}")
|
|
except socket.gaierror:
|
|
print("Hostname IP: Unable to resolve")
|
|
|
|
try:
|
|
docker_internal = socket.gethostbyname("host.docker.internal")
|
|
print(f"host.docker.internal: {docker_internal}")
|
|
except socket.gaierror:
|
|
print("host.docker.internal: Not resolvable")
|
|
|
|
except Exception as e:
|
|
print(f"Network info error: {e}")
|
|
|
|
# Check route table if available
|
|
if Path("/proc/net/route").exists():
|
|
print("Route table (first few lines):")
|
|
try:
|
|
with open("/proc/net/route", "r") as f:
|
|
lines = f.readlines()[:3] # First 3 lines
|
|
for line in lines:
|
|
print(f" {line.strip()}")
|
|
except Exception as e:
|
|
print(f"Route table read error: {e}")
|
|
|
|
|
|
async def main():
|
|
"""Main test function."""
|
|
print("🚀 Docker Host IP Detection Test Suite")
|
|
print("=" * 50)
|
|
|
|
# Display environment info
|
|
test_environment_info()
|
|
print()
|
|
|
|
# Run detection tests
|
|
success = await test_host_ip_detection()
|
|
|
|
print("\n" + "=" * 50)
|
|
if success:
|
|
print("✅ All tests completed successfully!")
|
|
detected_ip = get_host_ip()
|
|
print(f"📍 Detected Docker host IP: {detected_ip}")
|
|
print("💡 This IP should be used for container-to-host communication")
|
|
else:
|
|
print("❌ Some tests failed. Check the output above for details.")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|