Coverage for mcpgateway / handlers / signal_handlers.py: 100%

1# -*- coding: utf-8 -*- 

2"""Signal handlers for ContextForge Gateway. 

3 

4Provides SIGHUP handling for certificate rotation without restart. 

5""" 

6 

7# Standard 

8import asyncio 

9import logging 

10from typing import Any 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15async def sighup_reload() -> None: 

16 """Clear SSL context cache and drain MCP session pool on SIGHUP for certificate rotation. 

17 

18 Clears the SSL context cache to force recreation of SSL contexts 

19 with potentially updated certificates, and drains the MCP session 

20 pool so pooled connections reconnect with new TLS state. 

21 """ 

22 try: 

23 # First-Party 

24 from mcpgateway.utils.ssl_context_cache import clear_ssl_context_cache # pylint: disable=import-outside-toplevel 

25 

26 clear_ssl_context_cache() 

27 logger.info("SIGHUP: SSL context cache cleared") 

28 except Exception as exc: 

29 logger.error(f"SIGHUP handler failed to clear SSL context cache: {exc}") 

30 

31 try: 

32 # First-Party 

33 from mcpgateway.services.mcp_session_pool import drain_mcp_session_pool # pylint: disable=import-outside-toplevel 

34 

35 await drain_mcp_session_pool() 

36 logger.info("SIGHUP: MCP session pool drained for TLS rotation") 

37 except Exception as exc: 

38 logger.debug(f"SIGHUP: MCP session pool drain skipped: {exc}") 

39 

40 

41def sighup_handler(_signum: int, _frame: Any) -> None: 

42 """Handle SIGHUP signal by scheduling async SSL cache reload. 

43 

44 Signal handler that safely schedules an asynchronous task to clear 

45 the SSL context cache. Uses the running event loop to create a task 

46 for the async reload operation. 

47 

48 Args: 

49 _signum: Signal number (unused but required by signal handler signature) 

50 _frame: Current stack frame (unused but required by signal handler signature) 

51 """ 

52 logger.info("Received SIGHUP signal, scheduling SSL context cache refresh") 

53 try: 

54 event_loop = asyncio.get_running_loop() 

55 event_loop.create_task(sighup_reload()) 

56 except RuntimeError: 

57 logger.warning("SIGHUP received but event loop not running; skipping async reload")