Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

How to use event_loop_policy without triggering warnings? #799

Open
jgarvin opened this issue Mar 9, 2024 · 1 comment
Open

How to use event_loop_policy without triggering warnings? #799

jgarvin opened this issue Mar 9, 2024 · 1 comment

Comments

@jgarvin
Copy link

jgarvin commented Mar 9, 2024

I was using a event_loop fixture to use a custom event loop class, and noticed the warning from pytest to not do that and use an event loop policy instead. However, no matter how I define my policy fixture pytest warns that I didn't close the event loop. The policy usually isn't responsible for closing the event loop though, so I assume this is a pytest bug? Any tips on what I should be doing appreciated.

class CustomEventLoop(asyncio.SelectorEventLoop):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)

class CustomEventLoopPolicy(asyncio.AbstractEventLoopPolicy):
    def __init__(self) -> None:
        self._loop: asyncio.AbstractEventLoop | None = None
        # Use the default policy for child watcher methods
        self._default_policy = asyncio.get_event_loop_policy()

    def get_event_loop(self) -> asyncio.AbstractEventLoop:
        if self._loop is None or self._loop.is_closed():
            self._loop = self.new_event_loop()
        return self._loop

    def new_event_loop(self) -> asyncio.AbstractEventLoop:
        return CustomEventLoop()

    def set_event_loop(self, loop: asyncio.AbstractEventLoop | None) -> None:
        assert isinstance(loop, TaskTrackingEventLoop)
        self._loop = loop

    def get_child_watcher(self) -> asyncio.AbstractChildWatcher:
        # Delegate to the default policy's child watcher
        return self._default_policy.get_child_watcher()

    def set_child_watcher(self, watcher: asyncio.AbstractChildWatcher) -> None:
        # Delegate to the default policy's method
        self._default_policy.set_child_watcher(watcher)

@pytest.fixture(scope="module")
def event_loop_policy() -> Generator[TaskTrackingEventLoopPolicy | None, None, None]:
    policy = CustomEventLoopPolicy()
    yield policy
    policy.get_event_loop().close() # makes no difference
@jgarvin
Copy link
Author

jgarvin commented Mar 12, 2024

Some further evidence this may be a pytest-asyncio issue is using this decorator instead of pytest.mark.asyncio works with no warnings/errors:

import inspect
import asyncio
from typing import Callable, Awaitable

def async_test(f: Callable[[], None] | Callable[[], Awaitable[None]]) -> Callable[[], None]:
    def new_function() -> None:
        nonlocal f
        if inspect.iscoroutinefunction(f):
            with asyncio.Runner(loop_factory=CustomEventLoop) as runner:
                runner.run(f())
        else:
            f()

    return new_function

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant