pytest event loop is already running

[ryananguiano]

Because the test client calls loop.run_until_complete(connection(receive, send)), I cannot use anything that modifies the event loop in a pytest fixture without getting RuntimeError: This event loop is already running.

I would like to use a package like aresponses to mock aiohttp requests like this:

@pytest.fixture
def mock_endpoint(aresponses):
    aresponses.add('mock.com', '/test/', 'get', 'this is my mock response')

After messing with it for a couple hours, the only way I was able to get successful tests was to wrap the application in middleware:

@pytest.fixture
def app():
    class App:
        def __init__(self, app):
            self.app = app

        def __call__(self, scope):
            return functools.partial(self.asgi, scope=scope)

        async def asgi(self, receive, send, scope):
            async with ResponsesMockServer() as aresponses:
                aresponses.add('mock.com', '/test/', 'get', 'this is my mock response')
                inner = self.app(scope)
                await inner(receive, send)
    return App(application)


@pytest.fixture
def test_client(app):
    with TestClient(app) as test_client:
        yield test_client

I am going to modify this to allow myself to dynamically pass responses through the @pytest.mark decorator, but this workflow is not really convenient at all.

Am I missing something here? Is there a better way to set up my test clients, or should I just keep going down this route?

Thanks

Important

• We're using Polar.sh so you can upvote and help fund this issue.
• We receive the funding once the issue is completed & confirmed by you.
• Thank you in advance for helping prioritize & fund our backlog.

[ryananguiano]

As a follow up, I was able to get to a much cleaner workflow with this code:

import functools
import pytest
from aresponses import ResponsesMockServer


@pytest.fixture
def mocked_responses_app(request):
    mocked_responses = []
    for marker in request.node.iter_markers('mocked_responses'):
        if marker and marker.args:
            if len(marker.args) == 1 and isinstance(marker.args[0], (list, tuple)):
                responses = marker.args[0]
            else:
                responses = marker.args
            mocked_responses.extend(responses)

    class MockedResponsesApp:
        def __init__(self, app):
            self.app = app

        def __call__(self, scope):
            return functools.partial(self.asgi, scope=scope)

        async def asgi(self, receive, send, scope):
            async with ResponsesMockServer() as aresponses:
                for mock in mocked_responses:
                    self.add_mock_response(aresponses, mock)

                inner = self.app(scope)
                await inner(receive, send)

        def add_mock_response(self, aresponses, mock):
            aresponses.add(mock.host,
                           mock.url,
                           mock.method,
                           aresponses.passthrough if mock.response is None else mock.response)

    return MockedResponsesApp(application)


@pytest.fixture
def test_client(mocked_responses_app):
    with TestClient(mocked_responses_app) as test_client:
        yield test_client

###


class TestMock:
    host = 'mock.com'
    url = '/test/'
    method = 'get'
    response = 'this is my mock response'


@pytest.mark.mocked_responses([TestMock])
def test_endpoint(test_client):
    response = test_client.get('/endpoint/')
    assert response.status_code == 200

Now I can just tag any test with @pytest.mark.mocked_responses([list_of_mocks])

[tomchristie]

Okay - that's kinda awkward.

It's a bit of a side effect of the fact that the existing TestClient exposes a sync interface.
See also encode/databases#32 (comment)

Given that I've just released this... https://github.com/encode/requests-async one thing we could look at doing would be moving to having an async test client (requests-async, but with an ASGI adapter) and using async test cases. (pytest-asyncio seems to have that covered.)

[ryananguiano]

@tomchristie thanks! I will check out requests-async. This looks a lot easier to interact with async code than what I was doing.

[tomchristie]

Since the work's progressed, this'll actually end up being http3 now, rather than requests-async - See also ticket #553
Just need to get an ASGI adapter in there so that it can plug straight in as the test client.

[teskje]

Just FYI, I'm also stumbling over this currently trying to use the TestClient in an async test function (with pytest-asyncio). Minimal test case:

import pytest
from starlette.responses import HTMLResponse
from starlette.testclient import TestClient

async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = HTMLResponse('<html><body>Hello, world!</body></html>')
    await response(scope, receive, send)

@pytest.mark.asyncio
async def test_app():
    client = TestClient(app)
    response = client.get('/')
    assert response.status_code == 200

Fails with "RuntimeError: This event loop is already running" too. I guess there is currently no workaround, aside from just not using async tests, which I think wouldn't work in my specific case.

Thank you for working on this!

[eddebc]

Found a workaround from some other issue.

import nest_asyncio
nest_asyncio.apply()

at the top of the code.

[jacopofar]

I had a similar problem running pytest and the ASGI server in the same event loop and found this solution don't know if it's useful for you. Still gives some warning about files still open when closing the server

[taybin]

I have a similar problem where I'm also trying to embed a database connection with a transaction active for the current test in the request.state. I haven't quite found the magic configuration of pytest, pytest-asyncio, and TestClient to make this work yet.

Basically, I'd love to see an example of how to use TestClient along with databases/asyncpg.

[tomplex]

@taybin - I just found this library, after much wailing and gnashing of teeth with errors like the one in this issue. It worked for me; hope it helps you, too.

EDIT: Just realized I responded in a different issue than I meant to about a similar problem, but I'll leave it in case it's helpful.

[dmontagu]

@tomplex thanks for sharing, that looks useful.

I got excited about the idea of trying to merge it upstream, but noticed it was GPL licensed 😕.

[adsko]

@dmontagu Hey, this library is on MIT license now :)

[otsuka]

I used the starlette TestClient with nest_asyncio as a workaround for this problem, but I changed to use async-asgi-testclient.
It works very well so far.

[dmig-alarstudios]

async-asgi-testclient depends on requests. Such a shame!
I expected to get rid of requests in my project with help of this library.

[dmig-alarstudios]

Still best solution for me: #652 (comment)

[euri10]

Still best solution for me: #652 (comment)

asgi-lifespan simplifies this a lot, it adds another dependency though

[dmig-alarstudios]

@euri10 I tried to use just httpx.AsyncClient instead of starlette.testclient.TestClient -- works fine, what's the point of using asgi-lifespan?

[euri10]

Well it handles the lifespan events you may have declared in your app

[dmig-alarstudios]

@euri10 sorry, was too lazy to read specs :)
I don't use any of lifespan events, so httpx.AsyncClient is just enough.

[adriangb]

I tested this on 0.18.0. Modifying the example in #440 (comment):

import pytest
from starlette.responses import Response
from starlette.testclient import TestClient

async def app(scope, receive, send):
    await Response()(scope, receive, send)


@pytest.mark.anyio
async def test_app():
    client = TestClient(app)
    response = client.get('/')
    assert response.status_code == 200

This seems to work on the asyncio backend, but not Trio. On Trio it just gets stuck, presumably in a deadlock somewhere.
So I guess the short term solution is:

import pytest
from starlette.responses import Response
from starlette.testclient import TestClient

async def app(scope, receive, send):
    await Response()(scope, receive, send)


@pytest.mark.parametrize('anyio_backend', ['trio'])
@pytest.mark.anyio
async def test_app():
    client = TestClient(app)
    response = client.get('/')
    assert response.status_code == 200

Long term, I think @Kludex 's work in #1376 has the potential to fix this.
@Kludex , do you think this will be supported by the new test client / worth adding a test case for?

@ryananguiano (if you are still able to recall, I realize this is a couple years old now), would your original use case have been satisfied if there was an async version of TestClient available with the same API?

[Kludex]

I don't think so.

[cglacet]

Did anyone found a solution that doesn't involve hacking (without nesting loops)?

[chbndrhnns]

Did anyone found a solution that doesn't involve hacking (without nesting loops)?

I only use the httpx-based client as described in #652 (comment) and it works very well.
Together with asgi-lifespan, also startup and shutdown tasks are executed: https://github.com/florimondmanca/asgi-lifespan

[Kludex]

I don't know exactly when it was fixed, but I can't reproduce this on the latest version of pytest-asyncio/anyio and starlette.