From 2ddda38d05bf25ae8b0bb1a68eb0dc300c2b8942 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Sat, 9 May 2026 10:06:10 +0200 Subject: [PATCH] test: Fix flaky `test_emit_system_info_event` by waiting on event Replace the fixed `asyncio.sleep` with an `asyncio.Event` set from the listener and awaited with a generous timeout. Under loaded CI runners, `psutil.cpu_percent(interval=0.1)` running in `asyncio.to_thread` could delay the first emission past the 500ms window, leaving the listener uncalled. --- tests/unit/events/test_local_event_manager.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/unit/events/test_local_event_manager.py b/tests/unit/events/test_local_event_manager.py index 6c42c6c851..f2e998ab18 100644 --- a/tests/unit/events/test_local_event_manager.py +++ b/tests/unit/events/test_local_event_manager.py @@ -11,15 +11,19 @@ async def test_emit_system_info_event() -> None: mocked_listener = AsyncMock() + received = asyncio.Event() async def async_listener(payload: Any) -> None: await mocked_listener(payload) + received.set() system_info_interval = timedelta(milliseconds=50) - test_tolerance_coefficient = 10 async with LocalEventManager(system_info_interval=system_info_interval) as event_manager: event_manager.on(event=Event.SYSTEM_INFO, listener=async_listener) - await asyncio.sleep(system_info_interval.total_seconds() * test_tolerance_coefficient) + # Wait until the listener is invoked at least once. A generous timeout avoids flakiness on + # loaded CI runners, where `psutil.cpu_percent(interval=0.1)` in `asyncio.to_thread` can + # delay the first emission well beyond the configured interval. + await asyncio.wait_for(received.wait(), timeout=10) assert mocked_listener.call_count >= 1 assert isinstance(mocked_listener.call_args[0][0], EventSystemInfoData)