test: restore default event loop for sync tests
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
"""Shared fixtures for the hermes-agent test suite."""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
@@ -59,6 +60,39 @@ def mock_config():
|
||||
def _timeout_handler(signum, frame):
|
||||
raise TimeoutError("Test exceeded 30 second timeout")
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _ensure_current_event_loop(request):
|
||||
"""Provide a default event loop for sync tests that call get_event_loop().
|
||||
|
||||
Python 3.11+ no longer guarantees a current loop for plain synchronous tests.
|
||||
A number of gateway tests still use asyncio.get_event_loop().run_until_complete(...).
|
||||
Ensure they always have a usable loop without interfering with pytest-asyncio's
|
||||
own loop management for @pytest.mark.asyncio tests.
|
||||
"""
|
||||
if request.node.get_closest_marker("asyncio") is not None:
|
||||
yield
|
||||
return
|
||||
|
||||
try:
|
||||
loop = asyncio.get_event_loop_policy().get_event_loop()
|
||||
except RuntimeError:
|
||||
loop = None
|
||||
|
||||
created = loop is None or loop.is_closed()
|
||||
if created:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
if created and loop is not None:
|
||||
try:
|
||||
loop.close()
|
||||
finally:
|
||||
asyncio.set_event_loop(None)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _enforce_test_timeout():
|
||||
"""Kill any individual test that takes longer than 30 seconds."""
|
||||
|
||||
Reference in New Issue
Block a user