Други, раскажите пожалуйста как победить проблему с варнингом.
/home/master/PycharmProjects/src/kernel/system/interrupt_manager.py:28: DeprecationWarning: There is no current event loop
self.__loop = asyncio.get_event_loop()
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
Причем при простом запуске такова варнинга нет, он появляется при запуске теста.
Вроде как IsolatedAsyncioTestCase при запуске теста должен вызвать asyncio.run() что бы asyncio.get_event_loop() мог получить активный цикл. но по какой то неведомой причине этого не происходит. И я не как не могу нагуглить что с этим делать?
- pytest 8.1.1
- Python 3.11.0rc1
import asyncio
import logging
import unittest
from unittest.mock import MagicMock, patch
from src.kernel.state.state_system import SystemState
from src.kernel.system.interrupt_manager import InterruptManager
class TestInterruptManager(unittest.IsolatedAsyncioTestCase):
"""
Tests for the InterruptManager class, ensuring it properly handles OS interrupt signals
for a graceful shutdown.
File: src/kernel/system/interrupt_manager.py
"""
@classmethod
def setUpClass(cls):
# Disable logging for the duration of these tests.
logging.disable(logging.CRITICAL)
async def asyncSetUp(self):
self.loop = asyncio.new_event_loop()
self.loop.set_debug(False)
asyncio.set_event_loop(self.loop)
@patch('src.kernel.system.interrupt_manager.asyncio')
async def test_handler_sets_exit_event(self, mock_asyncio):
"""
Test that the interrupt handler sets the exit event upon receiving a signal.
"""
# Creating an ioc for SystemState
mock_system_state = MagicMock(SystemState)
# Creating an ioc for asyncio.The event that will be returned when it is called
mock_exit_event = MagicMock()
mock_asyncio.Event.return_value = mock_exit_event
# Creating a mock for the event loop and configuring it so that mock_exit_event.set is called
# when using call_soon_threadsafe
mock_event_loop = MagicMock()
mock_event_loop.call_soon_threadsafe.side_effect = lambda func, *args, **kwargs: func(*args, **kwargs)
mock_asyncio.get_event_loop.return_value = mock_event_loop
# # Initialize the InterruptManager with the locked SystemState
interrupt_manager = InterruptManager(mock_system_state)
# Simulate signal processing
interrupt_manager._InterruptManager__signal_handler()
# Check that exit event has been installed
mock_exit_event.set.assert_called_once()
async def asyncTearDown(self):
self.loop.close()
asyncio.set_event_loop(None)
@classmethod
def tearDownClass(cls):
# Re-enable logging after all tests in this class are done
logging.disable(logging.NOTSET)
if __name__ == '__main__':
unittest.main()
сам тестируемый код:
import asyncio
import logging
import signal
import sys
from src.kernel.core_services.check_types_method import check_types_method
from src.kernel.logger.dictionary.dictionary_system import DictionarySystem as Msg
from src.kernel.state.state_system import SystemState
if sys.platform == "win32":
import win32api
import win32console
class InterruptManager:
"""
Manages OS-level interrupt signals for graceful shutdown of a system.
This class sets up handlers for OS signals (e.g., SIGTERM, SIGINT on Unix/Linux,
CTRL_C_EVENT, CTRL_BREAK_EVENT on Windows) to safely transition the system into
an exit state. It leverages the asyncio event loop for asynchronous operation,
ensuring that the system can respond to shutdown signals even in the middle of
ongoing tasks.
"""
@check_types_method
def __init__(self, system_state: SystemState):
self.__system_state = system_state
self.__loop = asyncio.get_event_loop()
self.__exit_event = asyncio.Event()
if sys.platform != "win32":
self.__loop.add_signal_handler(signal.SIGTERM, self.__signal_handler)
self.__loop.add_signal_handler(signal.SIGINT, self.__signal_handler)
else:
self.__setup_windows_signal_handlers()
def __signal_handler(self):
"""
A handler for OS signals that triggers the shutdown sequence.
This method sets an asyncio event that indicates the system should begin
its shutdown process. It is intended to be called when the application
receives a shutdown signal from the OS.
"""
self.__loop.call_soon_threadsafe(self.__exit_event.set)
async def handler(self):
"""
Waits for a signal from the OS to shut down, after which it changes the state machine to the exit state.
"""
# Running a coroutine to monitor the state of the system in the background
asyncio.create_task(self.__monitor_system_state())
# Waiting for the exit signal
await self.__exit_event.wait()
logging.info(Msg.INF_INTERRUPT_MANAGER_HANDLER_MSG1)
self.__system_state.exit()
async def __monitor_system_state(self):
"""
Periodically checks the system status, and if it matches STATE_EXIT or STATE_ERROR,
sets the exit event.
"""
while not self.__exit_event.is_set():
if self.__system_state.state in (SystemState.STATE_EXIT, SystemState.STATE_ERROR):
self.__exit_event.set()
break
await asyncio.sleep(1)
if sys.platform == "win32":
def __setup_windows_signal_handlers(self):
def console_ctrl_handler(ctrl_type):
if ctrl_type in (win32console.CTRL_C_EVENT, win32console.CTRL_BREAK_EVENT):
self.__signal_handler()
return True
return False
win32api.SetConsoleCtrlHandler(console_ctrl_handler, True)
