monitor_app/tests/test_agent_monitor_integration.py

0001 """
0002 Integration tests for agent-monitor communication.
0003 Uses the exact same approach as the working test script.
0004 """
0005
0006 import os
0007 import time
0008 import requests
0009 from pathlib import Path
0010
0011 from django.test import TestCase
0012 import urllib3
0013
0014 # Disable SSL warnings for self-signed certificates in tests
0015 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
0016
0017
0018 class TestAgentMonitorIntegration(TestCase):
0019     """Test integration between agents and monitor API using exact working approach."""
0020
0021     def setUp(self):
0022         """Set up test environment exactly like working agents."""
0023         # Load environment exactly like working agents do
0024         env_file = Path.home() / ".env"
0025         if env_file.exists():
0026             with open(env_file) as f:
0027                 for line in f:
0028                     line = line.strip()
0029                     if line and not line.startswith('#') and '=' in line:
0030                         if line.startswith('export '):
0031                             line = line[7:]  # Remove 'export '
0032                         key, value = line.split('=', 1)
0033                         os.environ[key] = value.strip('"\'')
0034
0035         # Unset proxy variables exactly like working agents
0036         for proxy_var in ['http_proxy', 'https_proxy', 'HTTP_PROXY', 'HTTPS_PROXY']:
0037             if proxy_var in os.environ:
0038                 del os.environ[proxy_var]
0039
0040         # Get configuration exactly like working agents
0041         self.monitor_url = os.getenv('SWF_MONITOR_URL', 'https://localhost:8443')
0042         self.api_token = os.getenv('SWF_API_TOKEN')
0043
0044         if not self.api_token:
0045             self.skipTest("SWF_API_TOKEN environment variable not set")
0046
0047         # Configure session exactly like working agents
0048         self.session = requests.Session()
0049         self.session.headers.update({'Authorization': f'Token {self.api_token}'})
0050         self.session.verify = False  # Allow self-signed certs
0051         self.session.proxies = {'http': None, 'https': None}
0052
0053     def test_agent_heartbeat_exactly_like_working_script(self):
0054         """Test heartbeat using the exact same approach as the working script."""
0055         # Use exact same heartbeat data as working script
0056         heartbeat_data = {
0057             'instance_name': 'django-test-agent',
0058             'agent_type': 'TEST',
0059             'status': 'OK',
0060             'description': 'Django test using exact working approach',
0061             'mq_connected': False
0062         }
0063
0064         try:
0065             # Make exact same API call as working script
0066             response = self.session.post(
0067                 f"{self.monitor_url}/api/systemagents/heartbeat/",
0068                 json=heartbeat_data,
0069                 timeout=10
0070             )
0071
0072             # Check results exactly like working script
0073             print(f"Response status: {response.status_code}")
0074             print(f"Response text: {response.text}")
0075
0076             if response.status_code in [200, 201]:
0077                 result = response.json()
0078                 self.assertIn('instance_name', result)
0079                 self.assertEqual(result['instance_name'], 'django-test-agent')
0080                 print("✅ Django test SUCCESS: Heartbeat sent successfully!")
0081             else:
0082                 self.fail(f"Unexpected response {response.status_code}: {response.text}")
0083
0084         except requests.exceptions.ConnectionError as e:
0085             self.skipTest(f"Monitor not running: {e}")
0086         except Exception as e:
0087             self.fail(f"Test error: {e}")
0088
0089     def test_agent_list_api_exactly_like_working_approach(self):
0090         """Test system agents list using exact working approach."""
0091         try:
0092             response = self.session.get(
0093                 f"{self.monitor_url}/api/systemagents/",
0094                 timeout=10
0095             )
0096
0097             print(f"List agents response status: {response.status_code}")
0098
0099             if response.status_code == 200:
0100                 agents = response.json()
0101                 self.assertIsInstance(agents, list)
0102                 print(f"✅ Found {len(agents)} agents in monitor")
0103
0104                 # Look for our test agents
0105                 test_agents = [a for a in agents if 'test' in a['instance_name'].lower()]
0106                 if test_agents:
0107                     print(f"Found test agents: {[a['instance_name'] for a in test_agents]}")
0108             else:
0109                 self.fail(f"Failed to get agents list: {response.status_code}")
0110
0111         except requests.exceptions.ConnectionError as e:
0112             self.skipTest(f"Monitor not running: {e}")
0113         except Exception as e:
0114             self.fail(f"Test error: {e}")
0115
0116
0117 if __name__ == "__main__":
0118     import pytest
0119     pytest.main([__file__, "-v", "-s"])