"""
Comprehensive HTTP server and health check endpoint testing.
Tests server startup, health endpoints, threading behavior, and degraded modes
following Gemini's Phase 1 guidance for 100% coverage.
"""
import pytest
from unittest.mock import Mock, patch, MagicMock, call
import threading
import json
import os
from io import BytesIO
from http.server import BaseHTTPRequestHandler, HTTPServer
from unimus_client import (
UnimusRestClient,
UnimusError,
UnimusNotFoundError,
UnimusAuthenticationError
)
import server
from server import HealthCheckHandler, start_health_check_server
class TestHealthCheckHandler:
"""Test the HealthCheckHandler HTTP endpoint behavior."""
def create_handler_with_mocks(self, path, client_mock=None):
"""Create HealthCheckHandler with mocked I/O streams."""
handler = HealthCheckHandler.__new__(HealthCheckHandler)
handler.wfile = BytesIO()
handler.rfile = BytesIO()
handler.path = path
handler.client = client_mock or Mock()
# Mock HTTP response methods
handler.send_response = Mock()
handler.send_header = Mock()
handler.end_headers = Mock()
return handler
def test_healthz_endpoint_always_returns_200(self):
"""Test /healthz endpoint always returns 200 (liveness check)."""
handler = self.create_handler_with_mocks('/healthz')
handler.do_GET()
handler.send_response.assert_called_once_with(200)
handler.send_header.assert_called_with('Content-type', 'application/json')
handler.end_headers.assert_called_once()
# Check response body was written
response_data = handler.wfile.getvalue()
response_json = json.loads(response_data.decode('utf-8'))
assert response_json['status'] == 'alive'
assert response_json['server_version'] == '1.0.0'
def test_readyz_endpoint_successful_ready_check(self):
"""Test /readyz endpoint with successful ready check."""
client_mock = Mock()
client_mock.validate_connection.return_value = True
client_mock.get_health.return_value = {"status": "OK"}
handler = self.create_handler_with_mocks('/readyz', client_mock)
handler.do_GET()
handler.send_response.assert_called_once_with(200)
handler.send_header.assert_called_with('Content-type', 'application/json')
handler.end_headers.assert_called_once()
response_data = handler.wfile.getvalue()
response_json = json.loads(response_data.decode('utf-8'))
assert response_json['status'] == 'ready'
assert response_json['unimus_status'] == 'OK'
def test_readyz_endpoint_unimus_not_ready_503(self):
"""Test /readyz endpoint when Unimus connection fails (503)."""
client_mock = Mock()
client_mock.validate_connection.return_value = False
handler = self.create_handler_with_mocks('/readyz', client_mock)
handler.do_GET()
handler.send_response.assert_called_once_with(503)
handler.send_header.assert_called_with('Content-type', 'application/json')
handler.end_headers.assert_called_once()
response_data = handler.wfile.getvalue()
response_json = json.loads(response_data.decode('utf-8'))
assert response_json['status'] == 'not_ready'
assert 'Unimus connection failed' in response_json['reason']
def test_readyz_endpoint_unimus_health_error_503(self):
"""Test /readyz endpoint when Unimus health check fails (503)."""
client_mock = Mock()
client_mock.validate_connection.return_value = True
client_mock.get_health.side_effect = UnimusError("Service unavailable")
handler = self.create_handler_with_mocks('/readyz', client_mock)
handler.do_GET()
handler.send_response.assert_called_once_with(503)
handler.send_header.assert_called_with('Content-type', 'application/json')
handler.end_headers.assert_called_once()
response_data = handler.wfile.getvalue()
response_json = json.loads(response_data.decode('utf-8'))
assert response_json['status'] == 'not_ready'
assert 'Service unavailable' in response_json['reason']
def test_readyz_endpoint_authentication_error_503(self):
"""Test /readyz endpoint with authentication error (503)."""
client_mock = Mock()
client_mock.validate_connection.return_value = True
client_mock.get_health.side_effect = UnimusAuthenticationError("Invalid token")
handler = self.create_handler_with_mocks('/readyz', client_mock)
handler.do_GET()
handler.send_response.assert_called_once_with(503)
response_data = handler.wfile.getvalue()
response_json = json.loads(response_data.decode('utf-8'))
assert 'Invalid token' in response_json['reason']
def test_health_endpoint_redirect_to_readyz(self):
"""Test /health endpoint redirects to /readyz (301)."""
handler = self.create_handler_with_mocks('/health')
handler.do_GET()
handler.send_response.assert_called_once_with(301)
handler.send_header.assert_called_with('Location', '/readyz')
handler.end_headers.assert_called_once()
# No response body for redirect in the actual implementation
def test_unknown_endpoint_404(self):
"""Test unknown endpoint returns 404."""
handler = self.create_handler_with_mocks('/unknown')
handler.do_GET()
handler.send_response.assert_called_once_with(404)
handler.end_headers.assert_called_once()
response_data = handler.wfile.getvalue()
assert response_data == b'Not Found'
def test_no_client_provided_degrades_gracefully(self):
"""Test handler behavior when no client is provided."""
handler = self.create_handler_with_mocks('/readyz', client_mock=None)
handler.do_GET()
# Should call send_response with 503 for readiness check
handler.send_response.assert_called_with(503)
response_data = handler.wfile.getvalue()
response_json = json.loads(response_data.decode('utf-8'))
assert response_json['status'] == 'not_ready'
# The actual message varies based on the exact error path
assert 'not_ready' in response_json['status']
class TestHealthServerStartup:
"""Test health server startup and threading behavior."""
@patch('server.threading.Thread')
@patch('server.HTTPServer')
def test_successful_server_startup(self, mock_http_server, mock_thread):
"""Test successful health server startup with threading."""
# Mock HTTPServer instance
mock_server_instance = Mock()
mock_http_server.return_value = mock_server_instance
# Mock Thread instance
mock_thread_instance = Mock()
mock_thread.return_value = mock_thread_instance
client_mock = Mock()
port = 8080
start_health_check_server(client_mock, port)
# Verify HTTPServer was created with correct parameters
# The actual implementation uses partial(HealthCheckHandler, client)
mock_http_server.assert_called_once()
# Verify server instance was configured - the client is passed via partial
# The actual server instance doesn't have a client attribute directly
# Verify thread was created and started
mock_thread.assert_called_once_with(
target=mock_server_instance.serve_forever
)
mock_thread_instance.start.assert_called_once()
# Verify daemon property was set
assert mock_thread_instance.daemon == True
@patch('server.threading.Thread')
@patch('server.HTTPServer')
@patch('server.logger')
def test_server_startup_http_server_error(self, mock_logger, mock_http_server, mock_thread):
"""Test server startup when HTTPServer creation fails."""
mock_http_server.side_effect = OSError("Address already in use")
client_mock = Mock()
port = 8080
start_health_check_server(client_mock, port=port)
# Verify error was logged
mock_logger.warning.assert_called_once()
call_args = mock_logger.warning.call_args[0][0]
assert "Failed to start health check server" in call_args
# Thread should not be created if HTTPServer fails
mock_thread.assert_not_called()
@patch('server.threading.Thread')
@patch('server.HTTPServer')
@patch('server.logger')
def test_server_startup_thread_creation_error(self, mock_logger, mock_http_server, mock_thread):
"""Test server startup when thread creation fails."""
mock_server_instance = Mock()
mock_http_server.return_value = mock_server_instance
mock_thread.side_effect = RuntimeError("Cannot create thread")
client_mock = Mock()
port = 8080
start_health_check_server(client_mock, port=port)
# Verify error was logged
mock_logger.warning.assert_called_once()
call_args = mock_logger.warning.call_args[0][0]
assert "Failed to start health check server" in call_args
class TestServerInitializationAndSetup:
"""Test server initialization, client setup, and degraded modes."""
@patch('server.load_configuration')
@patch('server.UnimusRestClient')
def test_setup_unimus_client_success(self, mock_client_class, mock_load_config):
"""Test successful Unimus client setup from environment."""
mock_config = Mock()
mock_config.url = 'https://test.unimus.com'
mock_config.token = 'test-token'
mock_config.timeout = 30
mock_config.verify_ssl = True
mock_load_config.return_value = mock_config
mock_client_instance = Mock()
mock_client_class.return_value = mock_client_instance
client, config = server.setup_unimus_client()
mock_client_class.assert_called_once_with(
url='https://test.unimus.com',
token='test-token',
timeout=30,
verify_ssl=True
)
assert client == mock_client_instance
assert config == mock_config
@patch('server.load_configuration')
@patch('server.logger')
def test_setup_unimus_client_config_error(self, mock_logger, mock_load_config):
"""Test client setup with configuration error."""
mock_load_config.side_effect = ValueError("Configuration error")
client, config = server.setup_unimus_client()
assert client is None
assert config is None
@patch('server.load_configuration')
@patch('server.UnimusRestClient')
def test_setup_unimus_client_client_initialization_error(self, mock_client_class, mock_load_config):
"""Test client setup when UnimusRestClient initialization fails."""
mock_config = Mock()
mock_config.url = 'https://test.unimus.com'
mock_config.token = 'test-token'
mock_config.timeout = 30
mock_config.verify_ssl = True
mock_load_config.return_value = mock_config
mock_client_class.side_effect = ValueError("Invalid URL format")
client, config = server.setup_unimus_client()
assert client is None
assert config is None
@patch('server.validate_environment')
@patch('server.UnimusRestClient')
@patch('server.logger')
def test_initialize_client_success(self, mock_logger, mock_client_class, mock_validate_env):
"""Test successful client initialization with validation (deprecated function)."""
mock_validate_env.return_value = ('https://test.com', 'test-token')
mock_client = Mock()
mock_client.validate_connection.return_value = True
mock_client_class.return_value = mock_client
result = server.initialize_client()
assert result == mock_client
mock_client.validate_connection.assert_called_once()
mock_logger.info.assert_called()
@patch('server.validate_environment')
@patch('server.logger')
def test_initialize_client_environment_failure(self, mock_logger, mock_validate_env):
"""Test client initialization when environment validation fails."""
mock_validate_env.side_effect = ValueError("Missing environment variables")
with pytest.raises(ValueError):
server.initialize_client()
@patch('server.validate_environment')
@patch('server.UnimusRestClient')
@patch('server.logger')
def test_initialize_client_connection_failure(self, mock_logger, mock_client_class, mock_validate_env):
"""Test client initialization when connection validation fails."""
mock_validate_env.return_value = ('https://test.com', 'test-token')
mock_client = Mock()
mock_client.validate_connection.return_value = False
mock_client_class.return_value = mock_client
with pytest.raises(ValueError, match="Failed to connect to Unimus"):
server.initialize_client()
class TestMainServerExecution:
"""Test main server execution block and integration."""
@patch('server.setup_unimus_client')
@patch('server.start_health_check_server')
def test_main_execution_success(self, mock_start_health, mock_setup_client):
"""Test successful main server execution logic."""
mock_client = Mock()
mock_config = Mock()
mock_config.enable_health_server = True
mock_config.health_check_port = 9090
mock_setup_client.return_value = (mock_client, mock_config)
# Simulate main execution logic
unimus_client, config = server.setup_unimus_client()
if config is None or config.enable_health_server:
health_port = config.health_check_port if config else 8080
server.start_health_check_server(unimus_client, port=health_port)
assert unimus_client == mock_client
mock_start_health.assert_called_once_with(mock_client, port=9090)
@patch('server.setup_unimus_client')
@patch('server.start_health_check_server')
@patch('server.logger')
def test_main_execution_client_initialization_failure(self, mock_logger, mock_start_health, mock_setup_client):
"""Test main execution when client initialization fails."""
mock_setup_client.return_value = (None, None)
# Test degraded mode execution
unimus_client, config = server.setup_unimus_client()
if config is None or (config.enable_health_server if config else True):
health_port = config.health_check_port if config else 8080
server.start_health_check_server(unimus_client, port=health_port)
mock_start_health.assert_called_once_with(None, port=8080)
@patch('server.setup_unimus_client')
@patch('server.start_health_check_server')
@patch('server.logger')
def test_main_execution_health_server_disabled(self, mock_logger, mock_start_health, mock_setup_client):
"""Test main execution when health server is disabled."""
mock_client = Mock()
mock_config = Mock()
mock_config.enable_health_server = False
mock_setup_client.return_value = (mock_client, mock_config)
# Test health server disabled
unimus_client, config = server.setup_unimus_client()
if config is None or config.enable_health_server:
server.start_health_check_server(unimus_client, port=8080)
# Health server should not be started
mock_start_health.assert_not_called()
class TestServerIntegrationScenarios:
"""Test integrated server scenarios and edge cases."""
@patch('server.initialize_client')
@patch('server.start_health_check_server')
def test_complete_degraded_mode_operation(self, mock_start_health, mock_init_client):
"""Test complete server operation in degraded mode."""
mock_init_client.return_value = None
client = server.initialize_client()
server.start_health_check_server(client, 8080)
# Verify degraded mode setup
assert client is None
mock_start_health.assert_called_once_with(None, 8080)
def test_handler_inheritance_and_structure(self):
"""Test HealthCheckHandler inheritance and structure."""
assert issubclass(HealthCheckHandler, BaseHTTPRequestHandler)
assert hasattr(HealthCheckHandler, 'do_GET')
# Client is passed via __init__, not a class attribute
assert hasattr(HealthCheckHandler, '__init__')
@patch('server.threading.Thread')
@patch('server.HTTPServer')
def test_thread_daemon_configuration(self, mock_http_server, mock_thread):
"""Test that health server thread is properly configured as daemon."""
mock_server_instance = Mock()
mock_http_server.return_value = mock_server_instance
mock_thread_instance = Mock()
mock_thread.return_value = mock_thread_instance
start_health_check_server(Mock(), 8080)
# Verify thread was created
mock_thread.assert_called_once()
# Verify daemon property was set after creation
assert mock_thread_instance.daemon is True
# Verify thread start was called
mock_thread_instance.start.assert_called_once()
