source/swf-testbed/test_agent_communication.py

0001 #!/usr/bin/env python3
0002 """
0003 Simple test script to verify agent-monitor communication works.
0004 This uses the same approach as the working agents.
0005 """
0006
0007 import os
0008 import sys
0009 import time
0010 import requests
0011 from pathlib import Path
0012
0013 def setup_environment():
0014     """Auto-activate venv and load environment variables."""
0015     script_dir = Path(__file__).resolve().parent
0016
0017     # Auto-activate virtual environment if not already active
0018     if "VIRTUAL_ENV" not in os.environ:
0019         venv_path = script_dir / ".venv"
0020         if venv_path.exists():
0021             print("🔧 Auto-activating virtual environment...")
0022             venv_python = venv_path / "bin" / "python"
0023             if venv_python.exists():
0024                 os.environ["VIRTUAL_ENV"] = str(venv_path)
0025                 os.environ["PATH"] = f"{venv_path}/bin:{os.environ['PATH']}"
0026                 sys.executable = str(venv_python)
0027         else:
0028             print("❌ Error: No Python virtual environment found")
0029             return False
0030
0031     # Load ~/.env environment variables
0032     env_file = Path.home() / ".env"
0033     if env_file.exists():
0034         print("🔧 Loading environment variables from ~/.env...")
0035         with open(env_file) as f:
0036             for line in f:
0037                 line = line.strip()
0038                 if line and not line.startswith('#') and '=' in line:
0039                     if line.startswith('export '):
0040                         line = line[7:]  # Remove 'export '
0041                     key, value = line.split('=', 1)
0042                     value = value.strip('"\'')
0043                     # Skip entries with unexpanded shell variables
0044                     if '$' in value:
0045                         continue
0046                     os.environ[key] = value
0047
0048     # Unset proxy variables to prevent localhost routing through proxy
0049     for proxy_var in ['http_proxy', 'https_proxy', 'HTTP_PROXY', 'HTTPS_PROXY']:
0050         if proxy_var in os.environ:
0051             del os.environ[proxy_var]
0052
0053     return True
0054
0055 def test_monitor_connection():
0056     """Test basic connection to monitor using same approach as agents."""
0057     print("Testing monitor connection...")
0058
0059     monitor_url = os.getenv('SWF_MONITOR_URL', 'https://localhost:8443')
0060     api_token = os.getenv('SWF_API_TOKEN')
0061
0062     if not api_token:
0063         print("❌ No SWF_API_TOKEN found in environment")
0064         return False
0065
0066     print(f"Monitor URL: {monitor_url}")
0067     print(f"API Token: {api_token[:10]}...")
0068
0069     # Configure session exactly like agents do
0070     session = requests.Session()
0071     session.headers.update({'Authorization': f'Token {api_token}'})
0072     session.verify = False  # Allow self-signed certs
0073     session.proxies = {'http': None, 'https': None}
0074
0075     # Suppress SSL warnings
0076     import urllib3
0077     urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
0078
0079     try:
0080         # Test heartbeat endpoint like agents do
0081         heartbeat_data = {
0082             'instance_name': 'test-script-agent',
0083             'agent_type': 'TEST',
0084             'status': 'OK',
0085             'description': 'Test script for agent-monitor communication',
0086             'mq_connected': False
0087         }
0088
0089         print("Sending heartbeat to monitor...")
0090         response = session.post(
0091             f"{monitor_url}/api/systemagents/heartbeat/",
0092             json=heartbeat_data,
0093             timeout=10
0094         )
0095
0096         print(f"Response status: {response.status_code}")
0097         print(f"Response text: {response.text}")
0098
0099         if response.status_code in [200, 201]:
0100             print("✅ SUCCESS: Heartbeat sent successfully!")
0101             result = response.json()
0102             print(f"Agent registered as: {result.get('instance_name')}")
0103             return True
0104         else:
0105             print(f"❌ FAILED: Unexpected response {response.status_code}")
0106             return False
0107
0108     except requests.exceptions.ConnectionError as e:
0109         print(f"❌ CONNECTION ERROR: {e}")
0110         return False
0111     except Exception as e:
0112         print(f"❌ ERROR: {e}")
0113         return False
0114
0115 def main():
0116     print("=== Agent-Monitor Communication Test ===")
0117
0118     if not setup_environment():
0119         sys.exit(1)
0120
0121     if test_monitor_connection():
0122         print("✅ Test PASSED: Agent-monitor communication works!")
0123         sys.exit(0)
0124     else:
0125         print("❌ Test FAILED: Agent-monitor communication not working")
0126         sys.exit(1)
0127
0128 if __name__ == "__main__":
0129     main()