monitor_app/tests/test_django_dual_server_integration.py

0001 #!/usr/bin/env python3
0002 """
0003 Integration test for Django dual server configuration (Django HTTP on 8002, Django HTTPS on 8443).
0004 This tests the actual running Django servers, not just Django's test framework.
0005
0006 Run this test while the Django dual server is running:
0007 ./start_django_dual.sh
0008
0009 Then run:
0010 python manage.py test monitor_app.tests.test_django_dual_server_integration --keepdb
0011 """
0012
0013 import requests
0014 import urllib3
0015 import os
0016 import sys
0017 from datetime import datetime
0018 from django.test import TestCase
0019 from django.conf import settings
0020
0021 # Disable SSL warnings for self-signed certificates
0022 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
0023
0024
0025 class DjangoDualServerIntegrationTest(TestCase):
0026     """Integration tests for the actual running Django dual server configuration."""
0027
0028     @classmethod
0029     def setUpClass(cls):
0030         """Set up for Django dual server integration tests."""
0031         super().setUpClass()
0032
0033         # URLs for the actual running Django servers
0034         cls.django_http_url = os.getenv('SWF_MONITOR_HTTP_URL', 'http://localhost:8002')
0035         cls.django_https_url = os.getenv('SWF_MONITOR_URL', 'https://localhost:8443')
0036         cls.api_token = os.getenv('SWF_API_TOKEN', '')
0037
0038         # Check if Django servers are running
0039         cls.django_http_running = cls._check_django_server_running(cls.django_http_url)
0040         cls.django_https_running = cls._check_django_server_running(cls.django_https_url, verify_ssl=False)
0041
0042         if not cls.django_http_running:
0043             print(f"⚠️  Django HTTP server not running at {cls.django_http_url}")
0044         if not cls.django_https_running:
0045             print(f"⚠️  Django HTTPS server not running at {cls.django_https_url}")
0046
0047     @classmethod
0048     def _check_django_server_running(cls, url, verify_ssl=True):
0049         """Check if a Django server is running at the given URL."""
0050         try:
0051             response = requests.get(f"{url}/admin/login/",
0052                                   timeout=2,
0053                                   verify=verify_ssl)
0054             return response.status_code in [200, 302, 404]  # Any valid HTTP response
0055         except:
0056             return False
0057
0058     def test_django_http_server_logging_endpoint(self):
0059         """Test Django HTTP server (port 8002) for REST logging without authentication."""
0060         if not self.django_http_running:
0061             self.skipTest(f"Django HTTP server not running at {self.django_http_url}")
0062
0063         log_data = {
0064             'app_name': 'django_integration_test',
0065             'instance_name': 'test_django_dual_server',
0066             'timestamp': datetime.now().isoformat(),
0067             'level': 20,  # INFO
0068             'levelname': 'INFO',
0069             'message': 'Integration test log via Django HTTP server',
0070             'module': 'test_django_dual_server_integration',
0071             'funcname': 'test_django_http_server_logging_endpoint',
0072             'lineno': 55,
0073             'process': os.getpid(),
0074             'thread': 0,
0075             'extra_data': {'test_type': 'integration', 'server': 'Django_HTTP'}
0076         }
0077
0078         response = requests.post(
0079             f"{self.django_http_url}/api/logs/",
0080             json=log_data,
0081             timeout=5
0082         )
0083
0084         # Should succeed without authentication
0085         self.assertEqual(response.status_code, 201,
0086                         f"Django HTTP logging failed: {response.text}")
0087
0088         # Verify response contains the created log
0089         response_data = response.json()
0090         self.assertEqual(response_data['app_name'], 'django_integration_test')
0091         self.assertEqual(response_data['message'], 'Integration test log via Django HTTP server')
0092
0093     def test_django_https_server_unauthenticated_request(self):
0094         """Test Django HTTPS server (port 8443) returns 403 for unauthenticated requests."""
0095         if not self.django_https_running:
0096             self.skipTest(f"Django HTTPS server not running at {self.django_https_url}")
0097
0098         response = requests.get(
0099             f"{self.django_https_url}/api/systemagents/",
0100             verify=False,  # Skip SSL verification for self-signed cert
0101             timeout=5
0102         )
0103
0104         # Should return 403 Forbidden for unauthenticated request
0105         self.assertEqual(response.status_code, 403,
0106                         f"Expected 403 for unauthenticated Django HTTPS request, got {response.status_code}")
0107
0108     def test_django_https_server_authenticated_request(self):
0109         """Test Django HTTPS server (port 8443) with proper authentication."""
0110         if not self.django_https_running:
0111             self.skipTest(f"Django HTTPS server not running at {self.django_https_url}")
0112
0113         if not self.api_token:
0114             self.skipTest("No API token available in SWF_API_TOKEN environment variable")
0115
0116         headers = {'Authorization': f'Token {self.api_token}'}
0117
0118         response = requests.get(
0119             f"{self.django_https_url}/api/systemagents/",
0120             headers=headers,
0121             verify=False,  # Skip SSL verification for self-signed cert
0122             timeout=5
0123         )
0124
0125         # Should return 200 OK with authentication
0126         self.assertEqual(response.status_code, 200,
0127                         f"Django HTTPS authenticated request failed: {response.text}")
0128
0129         # Should return a list of agents
0130         data = response.json()
0131         self.assertIsInstance(data, list, "Response should be a list of agents")
0132
0133     def test_django_https_server_heartbeat_endpoint(self):
0134         """Test Django HTTPS server heartbeat endpoint with authentication."""
0135         if not self.django_https_running:
0136             self.skipTest(f"Django HTTPS server not running at {self.django_https_url}")
0137
0138         if not self.api_token:
0139             self.skipTest("No API token available in SWF_API_TOKEN environment variable")
0140
0141         headers = {'Authorization': f'Token {self.api_token}'}
0142         heartbeat_data = {
0143             'instance_name': 'django_integration_test_agent',
0144             'agent_type': 'test',
0145             'status': 'OK',
0146             'last_heartbeat': datetime.now().isoformat()
0147         }
0148
0149         response = requests.post(
0150             f"{self.django_https_url}/api/systemagents/heartbeat/",
0151             json=heartbeat_data,
0152             headers=headers,
0153             verify=False,
0154             timeout=5
0155         )
0156
0157         # Should return 200 OK or 201 Created (depending on whether agent exists)
0158         self.assertIn(response.status_code, [200, 201],
0159                      f"Django HTTPS heartbeat failed: {response.text}")
0160
0161         # Verify response
0162         response_data = response.json()
0163         self.assertEqual(response_data['instance_name'], 'django_integration_test_agent')
0164
0165     def test_django_dual_server_configuration_summary(self):
0166         """Test that both Django servers are properly configured for their intended purposes."""
0167         results = {
0168             'django_http_server': {
0169                 'url': self.django_http_url,
0170                 'running': self.django_http_running,
0171                 'purpose': 'Django REST logging (no auth required)'
0172             },
0173             'django_https_server': {
0174                 'url': self.django_https_url,
0175                 'running': self.django_https_running,
0176                 'purpose': 'Django authenticated API calls'
0177             }
0178         }
0179
0180         # Print configuration summary
0181         print("\n" + "="*60)
0182         print("DJANGO DUAL SERVER CONFIGURATION TEST SUMMARY")
0183         print("="*60)
0184
0185         for server_type, config in results.items():
0186             status = "✅ RUNNING" if config['running'] else "❌ NOT RUNNING"
0187             print(f"{server_type.upper()}: {status}")
0188             print(f"  URL: {config['url']}")
0189             print(f"  Purpose: {config['purpose']}")
0190
0191         print("="*60)
0192
0193         # Both Django servers should be running for full functionality
0194         if self.django_http_running and self.django_https_running:
0195             print("✅ DJANGO DUAL SERVER CONFIGURATION: FULLY OPERATIONAL")
0196         else:
0197             print("⚠️  DJANGO DUAL SERVER CONFIGURATION: INCOMPLETE")
0198             if not self.django_http_running and not self.django_https_running:
0199                 self.fail("Neither Django HTTP nor Django HTTPS server is running. Start with: ./start_django_dual.sh")
0200             elif not self.django_http_running:
0201                 self.fail(f"Django HTTP server not running at {self.django_http_url}")
0202             elif not self.django_https_running:
0203                 self.fail(f"Django HTTPS server not running at {self.django_https_url}")
0204
0205         print("")
0206
0207     @classmethod
0208     def tearDownClass(cls):
0209         """Clean up after Django dual server integration tests."""
0210         super().tearDownClass()
0211         # Integration tests don't need special cleanup
0212         pass