Make loop accessible outside of test function

[bj0]

I am using pytest_asyncio for my asyncio tests and am trying to get it to work with another fixture I wrote. They work together fine as long as I only access the loop inside the test function by calling functions on the passed in fixture. I was trying to figure out if it was possible to access the loop in the fixture function itself (or even pytest_runtest_setup).

I'm not that familiar with py.test internals, but as far as I can tell from reading examples and source code, it looks like pytest_asyncio creates the loop right before calling the test function (after *_setup and fixture functions are called) and then closes the loop right after execution. Is this right?

Specifically, what I'm trying to do is kick off an asyncio.ensure_future() right before the test runs using the loop that the test will be running on.

[Tinche]

So, if you're using the @pytest.mark.asyncio marker on a coroutine, this is what generally happens:

• find the event_loop fixture (this is so it can be overridden)
• get the event loop from it (the event loop is what the fixture is)
• use the event loop to run the marked coroutine
• close the event loop

If you have a more complex scenario, you don't actually need to use the marker at all. Just make an ordinary pytest test function that receives the event_loop fixture, do whatever you like, and then use the given event loop to run something (event_loop.run_until_complete()). Does this help?

Furthermore, pytest fixtures can depend on other fixtures. There's nothing stopping you from doing something like this:

def my_fixture(event_loop):
    # Do whatever, including asyncio.ensure_future()

@pytest.mark.asyncio
@pytest.mark.usefixtures('my_fixture')
async def test_something():
    await asyncio.sleep(1)

I think this should work (can't test it out currently). It's more complex but if it fits your use case, a little cleaner.

[bj0]

Ok, I had seen that option but thought that the fixture would receive a different instance of event_loop than the test function (I guess it figures out what order the fixtures need to be created in?). I just checked and it works for setting up, thanks.

The only issue I'm having now is that the event_loop is closed before any teardown code is called. For instance:

@pytest.yield_fixture
def my_fixture(event_loop):
    # do some setup that starts a background task with ensure_future(*, loop=event_loop)

    yield fixture

    # do some teardown that cleans up the background task

When the teardown code runs the loop is closed, which throws RuntimeError: Event loop is closed.

[Tinche]

Yeah, the problem is we're kinda bypassing the normal pytest fixture mechanism with the event loop. Fixtures are supposed to handle their own teardown, like in your snippet, but the event loop teardown is handled by the plugin (and before your fixture teardown).

My idea was to do it this way to make writing your event_loop fixture trivial for beginners (just create an instance and return it) but now I'm not sure it was the best course of action.

If you need a quick and dirty fix until this is sorted here, try something like this:

@pytest.yield_fixture
def event_loop():
    """Create an instance of the default event loop for each test case."""
    policy = asyncio.get_event_loop_policy()
    res = policy.new_event_loop()
    res._close = res.close
    res.close = lambda: None

    yield res

    res._close()

Yes, this is incredibly dirty, but should work for your case while I think of a good way of handling this.

[bj0]

That makes a lot of sense. I agree that moving the close to the event_loop fixture would probably be the easiest to understand. Fortunately for my case it's not a show-stopper.

Thanks for looking into it.

[lars-tiede]

As mentioned here I had to add set_event_loop() to the dirty hack mentioned above to get it to work for me:

@pytest.yield_fixture
def event_loop():
    """Create an instance of the default event loop for each test case."""
    policy = asyncio.get_event_loop_policy()
    res = policy.new_event_loop()
    asyncio.set_event_loop(res)
    res._close = res.close
    res.close = lambda: None

    yield res

    res._close()

[Tinche]

Hi, this issue should be solved by the latest release (0.5.0). Please close if that's the case!

[thomasst]

This is still an issue with 0.5.0. The event loop is not properly restored, causing tests that expect an open event loop and that follow an asyncio-marked-test to fail.

import asyncio
import pytest

def test_1():
    loop = asyncio.get_event_loop()
    assert not loop.is_closed()

@pytest.mark.asyncio
async def test_2():
    pass

def test_3():
    loop = asyncio.get_event_loop()
    assert not loop.is_closed()

============================= test session starts ==============================
platform darwin -- Python 3.5.0, pytest-3.0.7, py-1.4.33, pluggy-0.4.0 -- /private/tmp/venv/bin/python3.5
cachedir: .cache
rootdir: /private/tmp/venv, inifile:
plugins: asyncio-0.5.0
collected 3 items 

test_event_loop.py::test_1 PASSED
test_event_loop.py::test_2 PASSED
test_event_loop.py::test_3 FAILED

=================================== FAILURES ===================================
____________________________________ test_3 ____________________________________

    def test_3():
        loop = asyncio.get_event_loop()
>       assert not loop.is_closed()
E       assert not True
E        +  where True = <bound method BaseEventLoop.is_closed of <_UnixSelectorEventLoop running=False closed=True debug=False>>()
E        +    where <bound method BaseEventLoop.is_closed of <_UnixSelectorEventLoop running=False closed=True debug=False>> = <_UnixSelectorEventLoop running=False closed=True debug=False>.is_closed

test_event_loop.py:14: AssertionError
!!!!!!!!!!!!!!!!!!!! Interrupted: stopping after 1 failures !!!!!!!!!!!!!!!!!!!!
====================== 1 failed, 2 passed in 0.15 seconds ======================

[Tinche]

@thomasst Should be fixed in master, could you give it a try from git and report back?

[thomasst]

Thanks, looks good!

[AIGeneratedUsername]

pytest-asyncio 0.15.1 the problem is still here.

I can not use asyncio.get_event_loop().run_until_complete(function(*args, **kwargs)) in my tests.

I've tried @lars-tiede workaround, but I get "RuntimeError" "attached to a different loop" exception. And I tried @Tinche workaround, but I get RuntimeError: Event loop is closed

[asvetlov]

asyncio.get_event_loop().run_until_complete(function(*args, **kwargs)) doesn't work, sure.

The test should depend on event_loop fixture explicitly and use the given loop instance.
BTW, using asyncio.get_event_loop() on the top-level is deprecated starting from Python 3.10