pytest-dev
"RuntimeError: Thread 'MainThread' already has a main context" when used with glib event loop
glib event loop has its own event loop policy, and implicitly sets the event loop for the main thread. See https://pygobject.gnome.org/guide/asynchronous.html Note this needs gobject >= 3.50.0, to have the native asyncio support.
This doesn't work with pytest-asyncio and results in exception as in title when pytest_fixture_setup calls policy.set_event_loop(loop).
Minimal reproducer:
import asyncio
import pytest
from gi.events import GLibEventLoopPolicy
asyncio.set_event_loop_policy(GLibEventLoopPolicy())
@pytest.mark.asyncio
async def test_foo():
pass
The output:
======================================= ERRORS =======================================
_____________________________ ERROR at setup of test_foo _____________________________
policy = <gi.events.GLibEventLoopPolicy object at 0x7f81069ecc20>
@contextlib.contextmanager
def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[None]:
old_loop_policy = _get_event_loop_policy()
try:
old_loop = _get_event_loop_no_warn()
except RuntimeError:
old_loop = None
_set_event_loop_policy(policy)
try:
> yield
usr/lib/python3.14/site-packages/pytest_asyncio/plugin.py:543:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
usr/lib/python3.14/site-packages/pytest_asyncio/plugin.py:757: in _scoped_runner
runner = Runner().__enter__()
usr/lib64/python3.14/asyncio/runners.py:59: in __enter__
self._lazy_init()
usr/lib64/python3.14/asyncio/runners.py:150: in _lazy_init
events.set_event_loop(self._loop)
usr/lib64/python3.14/asyncio/events.py:839: in set_event_loop
_get_event_loop_policy().set_event_loop(loop)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <gi.events.GLibEventLoopPolicy object at 0x7f81069ecc20>
loop = <GLibEventLoop running=False closed=False debug=False ctx=0x5FB5B1886F60 loop=0x5FB5B1B968A0>
def set_event_loop(self, loop):
"""Set the event loop for the current context (python thread) to loop.
This is only permitted if the thread has no thread default main context
with the main thread using the default main context.
"""
# Only accept glib event loops, otherwise things will just mess up
assert loop is None or isinstance(loop, GLibEventLoop)
ctx = ctx_td = GLib.MainContext.get_thread_default()
if ctx is None and threading.current_thread() is threading.main_thread():
ctx = GLib.MainContext.default()
if loop is None:
# We do permit unsetting the current loop/context
old = self._loops.pop(hash(ctx), None)
if old:
if hash(old._context) != hash(ctx):
warnings.warn(
"GMainContext was changed unknowingly by asyncio integration!",
RuntimeWarning,
)
if ctx_td:
GLib.MainContext.pop_thread_default(ctx_td)
else:
# Only allow attaching if the thread has no main context yet
if ctx:
> raise RuntimeError(
f"Thread {threading.current_thread().name!r} already has a main context, "
"get_event_loop() will create a new loop if needed"
)
E RuntimeError: Thread 'MainThread' already has a main context, get_event_loop() will create a new loop if needed
usr/lib64/python3.14/site-packages/gi/events.py:813: RuntimeError
During handling of the above exception, another exception occurred:
fixturedef = <FixtureDef argname='_function_scoped_runner' scope='function' baseid=''>
request = <SubRequest '_function_scoped_runner' for <Coroutine test_foo>>
@pytest.hookimpl(wrapper=True)
def pytest_fixture_setup(fixturedef: FixtureDef, request) -> object | None:
asyncio_mode = _get_asyncio_mode(request.config)
if not _is_asyncio_fixture_function(fixturedef.func):
if asyncio_mode == Mode.STRICT:
# Ignore async fixtures without explicit asyncio mark in strict mode
# This applies to pytest_trio fixtures, for example
> return (yield)
usr/lib/python3.14/site-packages/pytest_asyncio/plugin.py:681:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
usr/lib/python3.14/site-packages/pytest_asyncio/plugin.py:756: in _scoped_runner
with _temporary_event_loop_policy(new_loop_policy):
usr/lib64/python3.14/contextlib.py:162: in __exit__
self.gen.throw(value)
usr/lib/python3.14/site-packages/pytest_asyncio/plugin.py:546: in _temporary_event_loop_policy
_set_event_loop(old_loop)
usr/lib/python3.14/site-packages/pytest_asyncio/plugin.py:575: in _set_event_loop
asyncio.set_event_loop(loop)
usr/lib64/python3.14/asyncio/events.py:839: in set_event_loop
_get_event_loop_policy().set_event_loop(loop)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <gi.events.GLibEventLoopPolicy object at 0x7f81069ecc20>
loop = <GLibEventLoop running=False closed=False debug=False ctx=0x5FB5B1762B90 loop=0x5FB5B19B7110>
def set_event_loop(self, loop):
"""Set the event loop for the current context (python thread) to loop.
This is only permitted if the thread has no thread default main context
with the main thread using the default main context.
"""
# Only accept glib event loops, otherwise things will just mess up
assert loop is None or isinstance(loop, GLibEventLoop)
ctx = ctx_td = GLib.MainContext.get_thread_default()
if ctx is None and threading.current_thread() is threading.main_thread():
ctx = GLib.MainContext.default()
if loop is None:
# We do permit unsetting the current loop/context
old = self._loops.pop(hash(ctx), None)
if old:
if hash(old._context) != hash(ctx):
warnings.warn(
"GMainContext was changed unknowingly by asyncio integration!",
RuntimeWarning,
)
if ctx_td:
GLib.MainContext.pop_thread_default(ctx_td)
else:
# Only allow attaching if the thread has no main context yet
if ctx:
> raise RuntimeError(
f"Thread {threading.current_thread().name!r} already has a main context, "
"get_event_loop() will create a new loop if needed"
)
E RuntimeError: Thread 'MainThread' already has a main context, get_event_loop() will create a new loop if needed
usr/lib64/python3.14/site-packages/gi/events.py:813: RuntimeError
================================== warnings summary ==================================
usr/lib64/python3.14/site-packages/gi/events.py:718
/usr/lib64/python3.14/site-packages/gi/events.py:718: DeprecationWarning: 'asyncio.AbstractEventLoopPolicy' is deprecated and slated for removal in Python 3.16
class GLibEventLoopPolicy(asyncio.AbstractEventLoopPolicy):
testcase.py:5
/testcase.py:5: DeprecationWarning: 'asyncio.set_event_loop_policy' is deprecated and slated for removal in Python 3.16
asyncio.set_event_loop_policy(GLibEventLoopPolicy())
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
============================== short test summary info ===============================
ERROR testcase.py::test_foo - RuntimeError: Thread 'MainThread' already has a main context, get_event_loop() wi...
Non-minimal failure: https://gitlab.com/QubesOS/qubes-desktop-linux-menu/-/jobs/11350892441
Thanks for reporting this and for the great reproducer! To be honest, I never even heard of GLibEventLoopPolicy until now.
NB: Pytest-asyncio wants to move away from the asyncio policy system, because it's deprecated in slated for removal. #1164 is an attempt to replace it in favor of a loop_factory argument to @pytest.mark.asyncio, but it's very complex and the policy system is still around, so we need to support it for some time to come.
I can reproduce the error, but from the logs, I cannot identify what the underlying issue is. The code fails when opening a context manager with an asyncio.Runner.
Can you elaborate a bit on the idea of a GLibContext (as used by the GLibEventLoop? Currently, I don't see the conection between this and with asyncio.Runner ☹️
The GLibEventLoopPolicy is to combine glib (gtk) event loop with asyncio - basically to use asyncio in a pygtk application. The native glib support is pretty new thing, previously it required 3rd-party module like https://github.com/beeware/gbulb.
I guess the issue here is that glib creates event loop implicitly and only exposes it via asyncio API, but I think you can't really replace it for the main thread due to how glib works. TBH, I don't know how this can be fixed... Not creating new loop for each test has several downsides.
I can manage to do most tests bypassing parts related to asyncio, but naturally, when the application uses asyncio (and pygtk), I do need to use them in tests at some point...
Pytest-asyncio gives the option to run more than one test in a single event loop via the loop_scope argument to pytest.mark.asyncio. In your reproducer, there's just a single test that fails.
Hopefully, the (very pressing) rework of the asyncio policy support in pytest-asyncio will shine some light on this.
If that can help, there is some interesting stuff in the PyGObject issue to deal with the deprecation of the loop factory: https://gitlab.gnome.org/GNOME/pygobject/-/issues/697
Necrobump. I'm cruelly hijacking this. For anyone who cares, you can ignore the above DeprecationWarning errors now unconditionally emitted by pytest-asyncio under newer Python versions with pytest options like:
pytest -W ignore::DeprecationWarning:pytest_asyncio.plugin
This matters for @beartype projects like pytest-beartype, which coerce otherwise non-fatal warnings into fatal test failures with -W error. I always enable -W error across all of my projects. It keeps me honest. Few things do. You are now thinking: "...I don't care." But wait! There's more. You might even care.
@beartype Fixed This Months Ago. So Can You! Maybe.
Personally, I'd make resolving these deprecations this plugin's highest priority. You shouldn't trust anything I say, but @beartype internally resolved most (...maybe even all?) of these deprecations with plugin logic resembling:
from asyncio import (
get_event_loop,
new_event_loop,
set_event_loop,
)
from functools import wraps
from inspect import iscoroutinefunction
from pytest import hookimpl
from sys import version_info
from warnings import (
catch_warnings,
simplefilter,
)
@hookimpl(hookwrapper=True, tryfirst=True)
def pytest_pyfunc_call(pyfuncitem: 'pytest.Function') -> None:
# Test function to be called by this hook.
test_func = pyfuncitem.obj
# If this test function is an asynchronous coroutine function (i.e.,
# callable declared with "async def" whose body contains *NO* "yield"
# expressions)...
if iscoroutinefunction(test_func):
@wraps(test_func)
def test_func_synchronous(*args, **kwargs):
'''
Closure synchronously calling the current asynchronous test
coroutine function under a new event loop uniquely isolated to this
coroutine.
'''
# If the active Python interpreter targets Python >= 3.14, avoid
# calling the deprecated get_event_loop_policy() getter preferred
# under Python <= 3.13. Instead...
if version_info >= (3, 14):
# Attempt to...
try:
# Current event loop for the current threading context if
# any *OR* raise a "RuntimeError" otherwise.
event_loop_old = get_event_loop()
# Close this loop.
event_loop_old.close()
# If attempting to retrieve the current event loop raised a
# "RuntimeError", there is *NO* current event loop to be closed.
# In this case, silently reduce to a noop.
except RuntimeError:
pass
# Else, the active Python interpreter targets Python <= 3.13. In
# this case, prefer calling the get_event_loop_policy() getter
# deprecated under Python >= 3.14. Specifically...
else:
# Defer version-specific imports.
from asyncio import get_event_loop_policy
# With a warning context manager...
with catch_warnings():
# Ignore *ALL* deprecating warnings emitted by the
# get_event_loop() function called below. For unknown
# reasons, CPython 3.11 devs thought that emitting a "There
# is no current event loop" warning (erroneously classified
# as a "deprecation") was a wonderful idea. "asyncio" is
# arduous enough to portably support as it is.
simplefilter('ignore', DeprecationWarning)
# Current event loop for the current threading context if
# any *OR* create a new event loop otherwise. Note that the
# higher-level asyncio.get_event_loop() getter is
# intentionally *NOT* called here, as Python 3.10 broke
# backward compatibility by refactoring that getter to be an
# alias for the wildly different asyncio.get_running_loop()
# getter, which *MUST* be called only from within either an
# asynchronous callable or running event loop. In either
# case, asyncio.get_running_loop() and thus
# asyncio.get_event_loop() is useless in this context.
# Instead, we call the lower-level
# get_event_loop_policy().get_event_loop() getter -- which
# asyncio.get_event_loop() used to wrap. *facepalm*
#
# This getter should ideally return "None" rather than
# creating a new event loop without our permission if no
# loop has been set. This getter instead does the latter,
# implying that this closure will typically instantiate two
# event loops per asynchronous coroutine test function:
# * The first useless event loop implicitly created by this
# get_event_loop() call.
# * The second useful event loop explicitly created by the
# subsequent new_event_loop() call.
#
# Since there exists *NO* other means of querying the
# current event loop, we reluctantly bite the bullet and pay
# the piper. Work with me here, guys!
event_loop_old = get_event_loop_policy().get_event_loop()
# Close this loop, regardless of whether the prior
# get_event_loop() call just implicitly created this loop,
# because the "asyncio" API offers *NO* means of
# differentiating these two common edge cases. *facepalm*
event_loop_old.close()
# New event loop isolated to this coroutine.
#
# Note that this event loop has yet to be set as the current event
# loop for the current threading context. Explicit is better than
# implicit.
event_loop = new_event_loop()
# Set this as the current event loop for this threading context.
set_event_loop(event_loop)
# Coroutine object produced by this asynchronous coroutine test
# function. Technically, coroutine functions are *NOT* actually
# coroutines; they're just syntactic sugar implemented as standard
# synchronous functions dynamically creating and returning
# asynchronous coroutine objects on each call.
test_func_coroutine = test_func(*args, **kwargs)
# Synchronously run a new asynchronous task implicitly scheduled to
# run this coroutine, ignoring the value returned by this coroutine
# (if any) while reraising any exception raised by this coroutine
# up the call stack to pytest.
event_loop.run_until_complete(test_func_coroutine)
# Close this event loop.
event_loop.close()
# Replace this asynchronous coroutine test function with this
# synchronous closure wrapping this test function.
pyfuncitem.obj = test_func_synchronous
# Perform this test by calling this test function.
yield
Easy-peasy. In theory, pytest-asyncio should be behaving similarly. In practice, pytest-asyncio is a beast. Of course, @beartype is a beast too. All of our packages are beasts around here! Roaaaar.
