monitor_app/tests/test_activemq_ssl_connection.py

0001 from django.test import TestCase
0002 from django.urls import reverse
0003 from django.contrib.auth.models import User
0004 from django.utils import timezone
0005 from rest_framework.test import APITestCase, APIClient
0006 from rest_framework import status
0007 from monitor_app.models import SystemAgent, AppLog, Run, StfFile, Subscriber
0008 from monitor_app.serializers import AppLogSerializer
0009 from django.core.management import call_command
0010 from io import StringIO
0011 import logging
0012 import uuid
0013 import re
0014
0015
0016 class ActiveMQSSLConnectionTests(TestCase):
0017     """
0018     Tests for ActiveMQ SSL connection functionality.
0019     These tests verify that the SSL configuration works correctly with the certificate.
0020     """
0021
0022     def setUp(self):
0023         """Set up test environment with ActiveMQ settings"""
0024         # Import here to avoid issues if stomp is not available
0025         try:
0026             import stomp
0027             import ssl
0028             self.stomp_available = True
0029         except ImportError:
0030             self.stomp_available = False
0031
0032     def test_activemq_ssl_configuration(self):
0033         """Test that ActiveMQ SSL settings are properly configured"""
0034         from django.conf import settings
0035
0036         # Test that all required SSL settings are available
0037         self.assertTrue(hasattr(settings, 'ACTIVEMQ_USE_SSL'))
0038         self.assertTrue(hasattr(settings, 'ACTIVEMQ_SSL_CA_CERTS'))
0039         self.assertTrue(hasattr(settings, 'ACTIVEMQ_HOST'))
0040         self.assertTrue(hasattr(settings, 'ACTIVEMQ_PORT'))
0041         self.assertTrue(hasattr(settings, 'ACTIVEMQ_USER'))
0042         self.assertTrue(hasattr(settings, 'ACTIVEMQ_PASSWORD'))
0043
0044         # Test that SSL is enabled and certificate path is set
0045         if settings.ACTIVEMQ_USE_SSL:
0046             self.assertIsNotNone(settings.ACTIVEMQ_SSL_CA_CERTS)
0047             self.assertNotEqual(settings.ACTIVEMQ_SSL_CA_CERTS, '')
0048
0049     def test_certificate_file_exists(self):
0050         """Test that the SSL certificate file exists and is readable"""
0051         from django.conf import settings
0052         import os
0053
0054         if getattr(settings, 'ACTIVEMQ_USE_SSL', False):
0055             cert_file = settings.ACTIVEMQ_SSL_CA_CERTS
0056             if cert_file:
0057                 self.assertTrue(os.path.exists(cert_file),
0058                                f"Certificate file not found: {cert_file}")
0059                 self.assertTrue(os.path.isfile(cert_file),
0060                                f"Certificate path is not a file: {cert_file}")
0061                 # Test that file is readable
0062                 with open(cert_file, 'r') as f:
0063                     content = f.read()
0064                     self.assertIn('-----BEGIN CERTIFICATE-----', content)
0065                     self.assertIn('-----END CERTIFICATE-----', content)
0066
0067     def test_activemq_connection_mock(self):
0068         """Test ActiveMQ connection setup with mocked connection"""
0069         if not self.stomp_available:
0070             self.skipTest("stomp.py not available")
0071
0072         from django.conf import settings
0073         from unittest.mock import patch, MagicMock
0074         import stomp
0075         import ssl
0076
0077         # Mock the stomp connection
0078         with patch('stomp.Connection') as mock_connection_class:
0079             mock_connection = MagicMock()
0080             mock_connection_class.return_value = mock_connection
0081
0082             # Test connection setup
0083             host = getattr(settings, 'ACTIVEMQ_HOST', 'localhost')
0084             port = getattr(settings, 'ACTIVEMQ_PORT', 61612)
0085             use_ssl = getattr(settings, 'ACTIVEMQ_USE_SSL', False)
0086
0087             # Create connection
0088             conn = stomp.Connection(host_and_ports=[(host, port)], vhost=host, try_loopback_connect=False)
0089
0090             # Verify connection was created with correct parameters
0091             mock_connection_class.assert_called_once_with(
0092                 host_and_ports=[(host, port)],
0093                 vhost=host,
0094                 try_loopback_connect=False
0095             )
0096
0097             # Test SSL configuration if enabled
0098             if use_ssl:
0099                 ssl_ca_certs = getattr(settings, 'ACTIVEMQ_SSL_CA_CERTS', '')
0100                 if ssl_ca_certs:
0101                     # Verify SSL would be configured
0102                     self.assertTrue(hasattr(conn, 'transport'))
0103
0104     def test_activemq_listener_import(self):
0105         """Test that the ActiveMQ listener components import correctly"""
0106         try:
0107             from monitor_app.activemq_connection import ActiveMQConnectionManager
0108             from monitor_app.activemq_processor import WorkflowMessageProcessor
0109             self.assertTrue(ActiveMQConnectionManager is not None)
0110             self.assertTrue(WorkflowMessageProcessor is not None)
0111         except ImportError as e:
0112             self.fail(f"Failed to import ActiveMQ components: {e}")
0113
0114     def test_activemq_ssl_connection_attempt(self):
0115         """Test actual SSL connection attempt (will fail if service not available)"""
0116         if not self.stomp_available:
0117             self.skipTest("stomp.py not available")
0118
0119         from django.conf import settings
0120         import stomp
0121         import ssl
0122         import socket
0123
0124         # Only run if SSL is configured
0125         if not getattr(settings, 'ACTIVEMQ_USE_SSL', False):
0126             self.skipTest("SSL not configured for ActiveMQ")
0127
0128         host = settings.ACTIVEMQ_HOST
0129         port = settings.ACTIVEMQ_PORT
0130         ca_certs = settings.ACTIVEMQ_SSL_CA_CERTS
0131
0132         # First check if the port is reachable
0133         try:
0134             sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
0135             sock.settimeout(2)  # 2 second timeout
0136             result = sock.connect_ex((host, port))
0137             sock.close()
0138
0139             if result != 0:
0140                 self.skipTest(f"ActiveMQ service not reachable at {host}:{port}")
0141
0142         except Exception as e:
0143             self.skipTest(f"Cannot test network connectivity: {e}")
0144
0145         # Test SSL connection setup
0146         conn = stomp.Connection(host_and_ports=[(host, port)], vhost=host, try_loopback_connect=False)
0147
0148         # Configure SSL
0149         if ca_certs:
0150             try:
0151                 conn.transport.set_ssl(
0152                     for_hosts=[(host, port)],
0153                     ca_certs=ca_certs,
0154                     ssl_version=ssl.PROTOCOL_TLS_CLIENT
0155                 )
0156
0157                 # Try to connect (will timeout or fail if service not available)
0158                 user = settings.ACTIVEMQ_USER
0159                 password = settings.ACTIVEMQ_PASSWORD
0160
0161                 # Use a short timeout for testing
0162                 try:
0163                     conn.connect(login=user, passcode=password, wait=True, version='1.2', timeout=5)
0164                     conn.disconnect()
0165                     # If we get here, connection succeeded
0166                     self.assertTrue(True, "ActiveMQ SSL connection successful")
0167                 except Exception as e:
0168                     # Connection failed - this is expected if service is not available
0169                     # We just verify the SSL setup didn't crash
0170                     self.assertIsInstance(e, (stomp.exception.ConnectFailedException,
0171                                              ConnectionError, OSError, socket.error))
0172
0173             except Exception as e:
0174                 self.fail(f"SSL configuration failed: {e}")