Laravel MCP Companion

"""Unit tests for graceful shutdown handler.""" import pytest import signal from unittest.mock import patch, MagicMock, call from shutdown_handler import GracefulShutdown class TestGracefulShutdown: """Test graceful shutdown functionality.""" def test_init(self): """Test initialization with default and custom logger.""" # Default logger shutdown = GracefulShutdown() assert shutdown.logger is not None assert shutdown.shutdown_callbacks == [] assert shutdown.is_shutting_down is False # Custom logger mock_logger = MagicMock() shutdown_custom = GracefulShutdown(logger=mock_logger) assert shutdown_custom.logger == mock_logger @patch('shutdown_handler.signal.signal') def test_signal_handlers_registered(self, mock_signal): """Test that signal handlers are properly registered.""" shutdown = GracefulShutdown() # Check that signal handlers were registered expected_calls = [ call(signal.SIGINT, shutdown._handle_shutdown), call(signal.SIGTERM, shutdown._handle_shutdown), ] # SIGBREAK is Windows-specific if hasattr(signal, 'SIGBREAK'): expected_calls.append(call(signal.SIGBREAK, shutdown._handle_shutdown)) mock_signal.assert_has_calls(expected_calls, any_order=True) def test_register_callback(self): """Test registering shutdown callbacks.""" shutdown = GracefulShutdown() def test_callback(arg1, arg2, kwarg1=None): pass shutdown.register(test_callback, "arg1_value", "arg2_value", kwarg1="kwarg1_value") assert len(shutdown.shutdown_callbacks) == 1 callback, args, kwargs = shutdown.shutdown_callbacks[0] assert callback == test_callback assert args == ("arg1_value", "arg2_value") assert kwargs == {"kwarg1": "kwarg1_value"} def test_register_multiple_callbacks(self): """Test registering multiple shutdown callbacks.""" shutdown = GracefulShutdown() def callback1(): pass def callback2(): pass shutdown.register(callback1) shutdown.register(callback2) assert len(shutdown.shutdown_callbacks) == 2 @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_handle_shutdown_first_signal(self, mock_print, mock_exit): """Test handling first shutdown signal.""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) callback_called = [] def test_callback(): callback_called.append(True) shutdown.register(test_callback) # Simulate SIGINT shutdown._handle_shutdown(signal.SIGINT, None) # Check that callback was called assert len(callback_called) == 1 assert shutdown.is_shutting_down is True # Check logging mock_logger.info.assert_called() mock_logger.debug.assert_called() # Check exit mock_exit.assert_called_once_with(0) @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_handle_shutdown_second_signal(self, mock_print, mock_exit): """Test handling second shutdown signal (force exit).""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) shutdown.is_shutting_down = True # Already shutting down # Simulate second signal shutdown._handle_shutdown(signal.SIGTERM, None) # Should force exit with code 1 mock_exit.assert_any_call(1) mock_logger.warning.assert_called() @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_handle_shutdown_callback_error(self, mock_print, mock_exit): """Test handling shutdown when callback raises exception.""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) def failing_callback(): raise Exception("Callback failed") def working_callback(): pass shutdown.register(failing_callback) shutdown.register(working_callback) # Should handle the error and continue shutdown._handle_shutdown(signal.SIGINT, None) # Should have logged the error mock_logger.error.assert_called() # Should still exit normally mock_exit.assert_called_once_with(0) @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_handle_shutdown_callbacks_executed_in_reverse_order(self, mock_print, mock_exit): """Test that callbacks are executed in reverse registration order.""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) execution_order = [] def callback1(): execution_order.append(1) def callback2(): execution_order.append(2) def callback3(): execution_order.append(3) # Register in order 1, 2, 3 shutdown.register(callback1) shutdown.register(callback2) shutdown.register(callback3) shutdown._handle_shutdown(signal.SIGINT, None) # Should execute in reverse order: 3, 2, 1 assert execution_order == [3, 2, 1] @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_handle_shutdown_with_callback_arguments(self, mock_print, mock_exit): """Test shutdown handling with callback arguments.""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) callback_args = [] def test_callback(arg1, arg2, kwarg1=None, kwarg2=None): callback_args.extend([arg1, arg2, kwarg1, kwarg2]) shutdown.register(test_callback, "pos1", "pos2", kwarg1="kw1", kwarg2="kw2") shutdown._handle_shutdown(signal.SIGINT, None) assert callback_args == ["pos1", "pos2", "kw1", "kw2"] @patch('shutdown_handler.signal.Signals') @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_signal_name_handling(self, mock_print, mock_exit, mock_signals): """Test signal name extraction for logging.""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) # Mock signal name mock_signal = MagicMock() mock_signal.name = "SIGINT" mock_signals.return_value = mock_signal shutdown._handle_shutdown(signal.SIGINT, None) # Should log with signal name log_calls = mock_logger.info.call_args_list assert any("SIGINT" in str(call) for call in log_calls) @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_signal_name_fallback(self, mock_print, mock_exit): """Test signal name fallback when signal.Signals not available.""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) # Temporarily remove Signals attribute from signal module original_signals = getattr(signal, 'Signals', None) if hasattr(signal, 'Signals'): delattr(signal, 'Signals') try: shutdown._handle_shutdown(signal.SIGINT, None) finally: # Restore Signals attribute if it existed if original_signals is not None: signal.Signals = original_signals # Should still work and log with signal number mock_logger.info.assert_called() log_calls = mock_logger.info.call_args_list assert any(str(signal.SIGINT) in str(call) for call in log_calls) def test_callback_registration_with_no_args(self): """Test callback registration with no arguments.""" shutdown = GracefulShutdown() def simple_callback(): pass shutdown.register(simple_callback) callback, args, kwargs = shutdown.shutdown_callbacks[0] assert callback == simple_callback assert args == () assert kwargs == {} def test_empty_shutdown(self): """Test shutdown with no registered callbacks.""" with patch('shutdown_handler.os._exit') as mock_exit, \ patch('builtins.print'): mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) shutdown._handle_shutdown(signal.SIGINT, None) # Should still work and exit normally mock_exit.assert_called_once_with(0) mock_logger.info.assert_called() @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_callback_with_side_effects(self, mock_print, mock_exit): """Test callback that has side effects.""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) side_effects = [] def callback_with_side_effects(): side_effects.append("cleanup_performed") # Simulate file cleanup, connection closing, etc. return "cleanup_result" shutdown.register(callback_with_side_effects) shutdown._handle_shutdown(signal.SIGINT, None) assert "cleanup_performed" in side_effects @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_multiple_callback_errors(self, mock_print, mock_exit): """Test handling multiple callback errors.""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) def failing_callback_1(): raise ValueError("Error 1") def failing_callback_2(): raise RuntimeError("Error 2") def working_callback(): pass shutdown.register(failing_callback_1) shutdown.register(working_callback) shutdown.register(failing_callback_2) shutdown._handle_shutdown(signal.SIGINT, None) # Should have logged both errors error_calls = mock_logger.error.call_args_list assert len(error_calls) == 2 # Should still exit normally mock_exit.assert_called_once_with(0) def test_shutdown_state_management(self): """Test shutdown state management.""" shutdown = GracefulShutdown() assert shutdown.is_shutting_down is False with patch('shutdown_handler.os._exit'), \ patch('builtins.print'): shutdown._handle_shutdown(signal.SIGINT, None) assert shutdown.is_shutting_down is True @pytest.mark.integration class TestShutdownIntegration: """Integration tests for shutdown handler.""" def test_shutdown_handler_in_context(self): """Test shutdown handler usage in realistic context.""" cleanup_performed = [] def cleanup_database(): cleanup_performed.append("database") def cleanup_files(): cleanup_performed.append("files") def cleanup_network(): cleanup_performed.append("network") shutdown_handler = GracefulShutdown() shutdown_handler.register(cleanup_database) shutdown_handler.register(cleanup_files) shutdown_handler.register(cleanup_network) with patch('shutdown_handler.os._exit'), \ patch('builtins.print'): # Simulate shutdown shutdown_handler._handle_shutdown(signal.SIGTERM, None) # All cleanup should have been performed in reverse order assert cleanup_performed == ["network", "files", "database"] def test_resource_cleanup_scenario(self): """Test realistic resource cleanup scenario.""" resources = { "database_connection": True, "file_handles": True, "network_sockets": True, "temp_files": True } def cleanup_resources(): resources["database_connection"] = False resources["file_handles"] = False resources["network_sockets"] = False resources["temp_files"] = False shutdown_handler = GracefulShutdown() shutdown_handler.register(cleanup_resources) with patch('shutdown_handler.os._exit'), \ patch('builtins.print'): shutdown_handler._handle_shutdown(signal.SIGINT, None) # All resources should be cleaned up assert all(not active for active in resources.values()) def test_shutdown_with_failing_and_succeeding_cleanup(self): """Test mixed success/failure cleanup scenario.""" cleanup_status = {} def critical_cleanup(): cleanup_status["critical"] = "success" def failing_cleanup(): cleanup_status["failing"] = "attempted" raise Exception("Cleanup failed") def optional_cleanup(): cleanup_status["optional"] = "success" shutdown_handler = GracefulShutdown() shutdown_handler.register(critical_cleanup) shutdown_handler.register(failing_cleanup) shutdown_handler.register(optional_cleanup) with patch('shutdown_handler.os._exit'), \ patch('builtins.print'): shutdown_handler._handle_shutdown(signal.SIGINT, None) # Critical and optional should succeed, failing should be attempted assert cleanup_status["critical"] == "success" assert cleanup_status["failing"] == "attempted" assert cleanup_status["optional"] == "success" class TestShutdownHandlerEdgeCases: """Test edge cases and error scenarios.""" def test_callback_modifying_callback_list(self): """Test callback that tries to modify the callback list.""" shutdown = GracefulShutdown() def modifying_callback(): # Try to add another callback during shutdown def new_callback(): pass shutdown.register(new_callback) shutdown.register(modifying_callback) with patch('shutdown_handler.os._exit'), \ patch('builtins.print'): # Should not crash shutdown._handle_shutdown(signal.SIGINT, None) def test_callback_infinite_loop_protection(self): """Test protection against callback infinite loops.""" shutdown = GracefulShutdown() call_count = [0] def recursive_callback(): call_count[0] += 1 if call_count[0] < 5: # Prevent actual infinite loop in test # This would be bad in real code pass shutdown.register(recursive_callback) with patch('shutdown_handler.os._exit'), \ patch('builtins.print'): shutdown._handle_shutdown(signal.SIGINT, None) # Should complete without hanging assert call_count[0] > 0 @patch('shutdown_handler.os._exit') @patch('builtins.print') def test_callback_with_complex_exception(self, mock_print, mock_exit): """Test callback with complex exception handling.""" mock_logger = MagicMock() shutdown = GracefulShutdown(logger=mock_logger) def complex_failing_callback(): try: raise ValueError("Inner error") except ValueError as e: raise RuntimeError("Outer error") from e shutdown.register(complex_failing_callback) shutdown._handle_shutdown(signal.SIGINT, None) # Should handle complex exception chains mock_logger.error.assert_called() mock_exit.assert_called_once_with(0) def test_very_long_callback_list(self): """Test shutdown with many callbacks.""" shutdown = GracefulShutdown() execution_count = [0] def counting_callback(): execution_count[0] += 1 # Register many callbacks for _ in range(100): shutdown.register(counting_callback) with patch('shutdown_handler.os._exit'), \ patch('builtins.print'): shutdown._handle_shutdown(signal.SIGINT, None) # All callbacks should execute assert execution_count[0] == 100