Pytest asyncio event is bound to a different event loop, event loop is closed

Question

I am trying to write some tests for my fastapi app

I am using the prisma-client-py for the database. I don't know if that changes anything

Everything works as they are supposed to apart from every first and last, they both fail with the same error:

RuntimeError: <asyncio.locks.Event object at 0x7f5696832950 [unset]> is bound to a different event loop

This is my conftest.py

import os
import asyncio
import pytest
from typing import Any, AsyncGenerator, Generator, Iterator

from fastapi import FastAPI
from fastapi.testclient import TestClient
from prisma import Prisma, register


from server.database.base import *
from server.config.exceptions import configure_exception_handlers
from server.config.settings import settings
from server.apis import apis


def start_application() -> FastAPI:
    """
    Return a FastAPI app
    """
    _app = FastAPI(
        title=str(settings.TITLE),
        description=str(settings.DESCRIPTION),
        version=str(settings.VERSION),
    )
    configure_exception_handlers(_app)
    _app.include_router(apis)
    return _app


TEST_DB_DSN = "postgresql://postgres:postgres@localhost:5433/postgres"
prisma = Prisma(datasource={"url": TEST_DB_DSN})


async def initialize_db() -> None:
    """
    Initialize the test database
    """
    print("Initializing")
    print("Creating all tables")
    stream = os.popen(f"dotenv -e .env.test prisma db push --skip-generate")
    output = stream.read()
    print(output)


async def teardown_db(client: Prisma) -> None:
    """
    Teardown the test database
    """
    print("Teardown")
    print("Dropping all tables")
    stream = os.popen(
        f'dotenv -e .env.test prisma db execute --url "{TEST_DB_DSN}" --file "./server/tests/utils/reset_db.sql" '
    )
    print("Creating all tables")
    stream = os.popen(f"DB_DSN={TEST_DB_DSN} prisma db push --skip-generate")
    output = stream.read()
    print(output)


@pytest.fixture(scope="session")
def app() -> Generator[FastAPI, Any, None]:
    """
    Initialize the app
    """
    _app = start_application()
    yield _app


@pytest.fixture(scope="module")
def event_loop() -> Iterator[asyncio.AbstractEventLoop]:
    """
    Initialize the event loop
    """
    loop = asyncio.get_event_loop_policy().new_event_loop()

    yield loop
    loop.close()


# Test client
@pytest.fixture(scope="module")
async def client(
    app: FastAPI, event_loop: asyncio.BaseEventLoop
) -> AsyncGenerator[TestClient, None]:
    """
    Initialize the test client
    """
    await initialize_db()
    register(prisma)
    await prisma.connect()
    with TestClient(app) as c:
        yield c
    await prisma.disconnect()
    await teardown_db(client=prisma)

and below are my tests

import asyncio
from fastapi.testclient import TestClient
from jose import jwt
import pytest

from prisma.models import User

from server.config.mail import fm
from server.config.settings import settings
from server.constants.user_types import UserType
from server.helpers.security import create_email_confirmation_token
from server.apis.auth.repositories import AuthRepository

auth_repository = AuthRepository()


def check_mail(data, outbox):
    assert data.get("password") is None
    assert len(outbox) == 1
    assert outbox[0]["subject"] == "Welcome to Mafflle - Verify your email"
    assert outbox[0]["From"] == f"{settings.EMAIL_FROM_NAME} <{settings.EMAIL_FROM}>"
    assert outbox[0]["To"] == data["email"]


@pytest.mark.asyncio
async def test_signup_user(client: TestClient, event_loop: asyncio.AbstractEventLoop):
    """
    Test the /v1/auth/signup/ endpoint.

    This endpoint should return a 201 status code and a
    JSON response with the full user object.
    """
    fm.config.SUPPRESS_SEND = 1
    with fm.record_messages() as outbox:
        payload = {
            "username": "test_user",
            "password": "Password123!",
            "email": "[email protected]",
        }
        response = client.post("/v1/auth/signup/", json=payload)
        data = response.json()

        assert response.status_code == 201
        assert data["username"] == "test_user"
        assert data["email"] == "[email protected]"
        assert data["user_type"] == "USER"
        assert data["email_confirmed"] == False
        assert data["is_active"] == True
        check_mail(data, outbox)

        # confirm email
        confirmation_token = create_email_confirmation_token(data["email"])
        params = {"token": confirmation_token}
        response = client.get("/v1/auth/confirm-email/", params=params)
        assert response.status_code == 200


@pytest.mark.asyncio
async def test_signup_business(
    client: TestClient, event_loop: asyncio.AbstractEventLoop
):
    """
    Test the /v1/auth/signup?user_type=business endpoint.

    This endpoint should return a 201 status code and a
    JSON response with the full user object.
    """
    fm.config.SUPPRESS_SEND = 1
    with fm.record_messages() as outbox:
        payload = {
            "username": "test_business",
            "password": "Password123!",
            "email": "[email protected]",
        }
        params = {"user_type": "BUSINESS"}
        response = client.post("/v1/auth/signup/", params=params, json=payload)
        data = response.json()

        assert response.status_code == 201
        assert data["username"] == "test_business"
        assert data["email"] == "[email protected]"
        assert data["user_type"] == "BUSINESS"
        assert data["email_confirmed"] == False
        assert data["is_active"] == True
        check_mail(data, outbox)


def test_signup_user_with_existing_username(
    client: TestClient, event_loop: asyncio.AbstractEventLoop
):
    """
    Test the /v1/auth/signup endpoint with an existing username.

    This endpoint should return a 400 status code and a
    JSON response with the error message.
    """
    payload = {
        "username": "test_user",
        "password": "Password123!",
        "email": "[email protected]",
    }
    # Create a user first
    client.post("/v1/auth/signup/", json=payload)
    # Then try to create a user with the same username
    response = client.post("/v1/auth/signup/", json=payload)
    data = response.json()
    assert response.status_code == 400
    assert data["detail"] == "User with this username already exists."


def test_signup_user_with_existing_email(
    client: TestClient, event_loop: asyncio.AbstractEventLoop
):
    """
    Test the /v1/auth/signup endpoint with an existing email.

    This endpoint should return a 400 status code and a
    JSON response with the error message.
    """
    payload = {
        "username": "test_user_2",
        "password": "Password123!",
        "email": "[email protected]",
    }
    # Create a user first
    client.post("/v1/auth/signup/", json=payload)
    # Then try to create a user with the same email
    response = client.post("/v1/auth/signup/", json=payload)
    data = response.json()
    assert response.status_code == 400
    assert data["detail"] == "User with this email already exists."


def test_signup_user_with_invalid_password(
    client: TestClient, event_loop: asyncio.AbstractEventLoop
):
    """
    Test the /v1/auth/signup endpoint with an invalid password.

    This endpoint should return a 400 status code and a
    JSON response with the error message.
    """
    payload = {
        "username": "test_user_3",
        "password": "Password",
        "email": "[email protected]",
    }
    response = client.post("/v1/auth/signup/", json=payload)
    data = response.json()
    assert response.status_code == 400
    assert (
        data["detail"]
        == "Password must be at least 8 characters long and contain at least one"
        " number,one uppercase letter and one special character."
    )


@pytest.mark.asyncio
async def test_login_user(client: TestClient, event_loop: asyncio.AbstractEventLoop):
    """
    Test the /v1/auth/login endpoint.

    This endpoint should return a 200 status code and a
    JSON response with the full user object.
    """
    # Create a user first
    payload = {
        "username": "test_user",
        "password": "Password123!",
        "email": "[email protected]",
    }
    client.post("/v1/auth/signup/", json=payload)
    # login
    payload = {"identity": "test_user", "password": "Password123!"}
    response = client.post("/v1/auth/login/", json=payload)
    assert response.status_code == 200
    access_token = response.cookies["access_token"]
    refresh_token = response.cookies["refresh_token"]
    assert access_token is not None
    assert refresh_token is not None
    access_token_data = jwt.decode(
        token=access_token, key=settings.AUTH_SECRET, algorithms=["HS256"]
    )
    assert access_token_data["scope"] == "access_token"
    user = await auth_repository.get_user_by_id(user_id=access_token_data["sub"])
    assert user is not None
    assert user.username == "test_user"
    assert user.email == "[email protected]"
    assert user.user_type == UserType.USER
    assert user.email_confirmed == True
    assert user.is_active == True

    # refresh token
    refresh_token_data = jwt.decode(
        token=refresh_token, key=settings.AUTH_SECRET, algorithms=["HS256"]
    )
    assert refresh_token_data["sub"] == str(user.id)
    assert refresh_token_data["scope"] == "refresh_token"

This is my error/failure in detail

========================================================================== test session starts ==========================================================================
platform linux -- Python 3.10.4, pytest-7.1.2, pluggy-1.0.0 -- /home/sheyzi/code/mafflle/mafflle_backend/venv/bin/python3
cachedir: .pytest_cache
rootdir: /home/sheyzi/code/mafflle/mafflle_backend, configfile: pyproject.toml
plugins: asyncio-0.18.3, anyio-3.6.1
asyncio: mode=auto
collected 6 items                                                                                                                                                       

server/tests/test_authentication/test_users.py::test_signup_user FAILED                                                                                           [ 16%]
server/tests/test_authentication/test_users.py::test_signup_business PASSED                                                                                       [ 33%]
server/tests/test_authentication/test_users.py::test_signup_user_with_existing_username PASSED                                                                    [ 50%]
server/tests/test_authentication/test_users.py::test_signup_user_with_existing_email PASSED                                                                       [ 66%]
server/tests/test_authentication/test_users.py::test_signup_user_with_invalid_password PASSED                                                                     [ 83%]
server/tests/test_authentication/test_users.py::test_login_user FAILED                                                                                            [100%]

=============================================================================== FAILURES ================================================================================
___________________________________________________________________________ test_signup_user ____________________________________________________________________________

client = <starlette.testclient.TestClient object at 0x7fb442d59870>, event_loop = <_UnixSelectorEventLoop running=False closed=False debug=False>

    @pytest.mark.asyncio
    async def test_signup_user(client: TestClient, event_loop: asyncio.AbstractEventLoop):
        """
        Test the /v1/auth/signup/ endpoint.
    
        This endpoint should return a 201 status code and a
        JSON response with the full user object.
        """
        fm.config.SUPPRESS_SEND = 1
        with fm.record_messages() as outbox:
            payload = {
                "username": "test_user",
                "password": "Password123!",
                "email": "[email protected]",
            }
>           response = client.post("/v1/auth/signup/", json=payload)

server/tests/test_authentication/test_users.py:40: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
venv/lib/python3.10/site-packages/requests/sessions.py:635: in post
    return self.request("POST", url, data=data, json=json, **kwargs)
venv/lib/python3.10/site-packages/starlette/testclient.py:468: in request
    return super().request(
venv/lib/python3.10/site-packages/requests/sessions.py:587: in request
    resp = self.send(prep, **send_kwargs)
venv/lib/python3.10/site-packages/requests/sessions.py:701: in send
    r = adapter.send(request, **kwargs)
venv/lib/python3.10/site-packages/starlette/testclient.py:266: in send
    raise exc
venv/lib/python3.10/site-packages/starlette/testclient.py:263: in send
    portal.call(self.app, scope, receive, send)
venv/lib/python3.10/site-packages/anyio/from_thread.py:283: in call
    return cast(T_Retval, self.start_task_soon(func, *args).result())
/usr/lib/python3.10/concurrent/futures/_base.py:446: in result
    return self.__get_result()
/usr/lib/python3.10/concurrent/futures/_base.py:391: in __get_result
    raise self._exception
venv/lib/python3.10/site-packages/anyio/from_thread.py:219: in _call_func
    retval = await retval
venv/lib/python3.10/site-packages/fastapi/applications.py:261: in __call__
    await super().__call__(scope, receive, send)
venv/lib/python3.10/site-packages/starlette/applications.py:112: in __call__
    await self.middleware_stack(scope, receive, send)
venv/lib/python3.10/site-packages/starlette/middleware/errors.py:181: in __call__
    raise exc
venv/lib/python3.10/site-packages/starlette/middleware/errors.py:159: in __call__
    await self.app(scope, receive, _send)
venv/lib/python3.10/site-packages/starlette/exceptions.py:82: in __call__
    raise exc
venv/lib/python3.10/site-packages/starlette/exceptions.py:71: in __call__
    await self.app(scope, receive, sender)
venv/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py:21: in __call__
    raise e
venv/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py:18: in __call__
    await self.app(scope, receive, send)
venv/lib/python3.10/site-packages/starlette/routing.py:656: in __call__
    await route.handle(scope, receive, send)
venv/lib/python3.10/site-packages/starlette/routing.py:259: in handle
    await self.app(scope, receive, send)
venv/lib/python3.10/site-packages/starlette/routing.py:61: in app
    response = await func(request)
venv/lib/python3.10/site-packages/fastapi/routing.py:227: in app
    raw_response = await run_endpoint_function(
venv/lib/python3.10/site-packages/fastapi/routing.py:160: in run_endpoint_function
    return await dependant.call(**values)
server/apis/auth/router.py:30: in signup
    return await self.auth_service.signup(
server/apis/auth/services.py:82: in signup
    if await self.auth_repository.get_user_by_username_or_email(user.username):
server/apis/auth/repositories.py:66: in get_user_by_username_or_email
    user = await User.prisma().find_first(
venv/lib/python3.10/site-packages/prisma/actions.py:1389: in find_first
    resp = await self._client._execute(
venv/lib/python3.10/site-packages/prisma/client.py:353: in _execute
    return await self._engine.query(builder.build())
venv/lib/python3.10/site-packages/prisma/engine/query.py:185: in query
    return await self.request('POST', '/', content=content)
venv/lib/python3.10/site-packages/prisma/engine/http.py:96: in request
    resp = await self.session.request(method, url, **kwargs)
venv/lib/python3.10/site-packages/prisma/_async_http.py:28: in request
    return Response(await self.session.request(method, url, **kwargs))
venv/lib/python3.10/site-packages/httpx/_client.py:1506: in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
venv/lib/python3.10/site-packages/httpx/_client.py:1593: in send
    response = await self._send_handling_auth(
venv/lib/python3.10/site-packages/httpx/_client.py:1621: in _send_handling_auth
    response = await self._send_handling_redirects(
venv/lib/python3.10/site-packages/httpx/_client.py:1658: in _send_handling_redirects
    response = await self._send_single_request(request)
venv/lib/python3.10/site-packages/httpx/_client.py:1695: in _send_single_request
    response = await transport.handle_async_request(request)
venv/lib/python3.10/site-packages/httpx/_transports/default.py:353: in handle_async_request
    resp = await self._pool.handle_async_request(req)
venv/lib/python3.10/site-packages/httpcore/_async/connection_pool.py:253: in handle_async_request
    raise exc
venv/lib/python3.10/site-packages/httpcore/_async/connection_pool.py:237: in handle_async_request
    response = await connection.handle_async_request(request)
venv/lib/python3.10/site-packages/httpcore/_async/connection.py:90: in handle_async_request
    return await self._connection.handle_async_request(request)
venv/lib/python3.10/site-packages/httpcore/_async/http11.py:102: in handle_async_request
    raise exc
venv/lib/python3.10/site-packages/httpcore/_async/http11.py:81: in handle_async_request
    ) = await self._receive_response_headers(**kwargs)
venv/lib/python3.10/site-packages/httpcore/_async/http11.py:143: in _receive_response_headers
    event = await self._receive_event(timeout=timeout)
venv/lib/python3.10/site-packages/httpcore/_async/http11.py:172: in _receive_event
    data = await self._network_stream.read(
venv/lib/python3.10/site-packages/httpcore/backends/asyncio.py:31: in read
    return await self._stream.receive(max_bytes=max_bytes)
venv/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:1265: in receive
    await self._protocol.read_event.wait()
/usr/lib/python3.10/asyncio/locks.py:211: in wait
    fut = self._get_loop().create_future()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <asyncio.locks.Event object at 0x7fb442dda980 [set]>

    def _get_loop(self):
        loop = events._get_running_loop()
    
        if self._loop is None:
            with _global_lock:
                if self._loop is None:
                    self._loop = loop
        if loop is not self._loop:
>           raise RuntimeError(f'{self!r} is bound to a different event loop')
E           RuntimeError: <asyncio.locks.Event object at 0x7fb442dda980 [unset]> is bound to a different event loop

/usr/lib/python3.10/asyncio/mixins.py:30: RuntimeError
------------------------------------------------------------------------- Captured stdout setup -------------------------------------------------------------------------
Initializing
Creating all tables
Environment variables loaded from .env
Prisma schema loaded from prisma/schema.prisma
Datasource "db": PostgreSQL database "postgres", schema "public" at "localhost:5433"

🚀  Your database is now in sync with your schema. Done in 1.89s


____________________________________________________________________________ test_login_user ____________________________________________________________________________

client = <starlette.testclient.TestClient object at 0x7fb442d59870>, event_loop = <_UnixSelectorEventLoop running=False closed=False debug=False>

    @pytest.mark.asyncio
    async def test_login_user(client: TestClient, event_loop: asyncio.AbstractEventLoop):
        """
        Test the /v1/auth/login endpoint.
    
        This endpoint should return a 200 status code and a
        JSON response with the full user object.
        """
        # Create a user first
        payload = {
            "username": "test_user",
            "password": "Password123!",
            "email": "[email protected]",
        }
        client.post("/v1/auth/signup/", json=payload)
        # login
        payload = {"identity": "test_user", "password": "Password123!"}
        response = client.post("/v1/auth/login/", json=payload)
        assert response.status_code == 200
        access_token = response.cookies["access_token"]
        refresh_token = response.cookies["refresh_token"]
        assert access_token is not None
        assert refresh_token is not None
        access_token_data = jwt.decode(
            token=access_token, key=settings.AUTH_SECRET, algorithms=["HS256"]
        )
        assert access_token_data["scope"] == "access_token"
>       user = await auth_repository.get_user_by_id(user_id=access_token_data["sub"])

server/tests/test_authentication/test_users.py:185: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
server/apis/auth/repositories.py:33: in get_user_by_id
    user = await User.prisma().find_first(
venv/lib/python3.10/site-packages/prisma/actions.py:1389: in find_first
    resp = await self._client._execute(
venv/lib/python3.10/site-packages/prisma/client.py:353: in _execute
    return await self._engine.query(builder.build())
venv/lib/python3.10/site-packages/prisma/engine/query.py:185: in query
    return await self.request('POST', '/', content=content)
venv/lib/python3.10/site-packages/prisma/engine/http.py:96: in request
    resp = await self.session.request(method, url, **kwargs)
venv/lib/python3.10/site-packages/prisma/_async_http.py:28: in request
    return Response(await self.session.request(method, url, **kwargs))
venv/lib/python3.10/site-packages/httpx/_client.py:1506: in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
venv/lib/python3.10/site-packages/httpx/_client.py:1593: in send
    response = await self._send_handling_auth(
venv/lib/python3.10/site-packages/httpx/_client.py:1621: in _send_handling_auth
    response = await self._send_handling_redirects(
venv/lib/python3.10/site-packages/httpx/_client.py:1658: in _send_handling_redirects
    response = await self._send_single_request(request)
venv/lib/python3.10/site-packages/httpx/_client.py:1695: in _send_single_request
    response = await transport.handle_async_request(request)
venv/lib/python3.10/site-packages/httpx/_transports/default.py:353: in handle_async_request
    resp = await self._pool.handle_async_request(req)
venv/lib/python3.10/site-packages/httpcore/_async/connection_pool.py:253: in handle_async_request
    raise exc
venv/lib/python3.10/site-packages/httpcore/_async/connection_pool.py:237: in handle_async_request
    response = await connection.handle_async_request(request)
venv/lib/python3.10/site-packages/httpcore/_async/connection.py:90: in handle_async_request
    return await self._connection.handle_async_request(request)
venv/lib/python3.10/site-packages/httpcore/_async/http11.py:102: in handle_async_request
    raise exc
venv/lib/python3.10/site-packages/httpcore/_async/http11.py:81: in handle_async_request
    ) = await self._receive_response_headers(**kwargs)
venv/lib/python3.10/site-packages/httpcore/_async/http11.py:143: in _receive_response_headers
    event = await self._receive_event(timeout=timeout)
venv/lib/python3.10/site-packages/httpcore/_async/http11.py:172: in _receive_event
    data = await self._network_stream.read(
venv/lib/python3.10/site-packages/httpcore/backends/asyncio.py:31: in read
    return await self._stream.receive(max_bytes=max_bytes)
venv/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:1265: in receive
    await self._protocol.read_event.wait()
/usr/lib/python3.10/asyncio/locks.py:211: in wait
    fut = self._get_loop().create_future()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <asyncio.locks.Event object at 0x7fb44038b7c0 [unset]>

    def _get_loop(self):
        loop = events._get_running_loop()
    
        if self._loop is None:
            with _global_lock:
                if self._loop is None:
                    self._loop = loop
        if loop is not self._loop:
>           raise RuntimeError(f'{self!r} is bound to a different event loop')
E           RuntimeError: <asyncio.locks.Event object at 0x7fb44038b7c0 [unset]> is bound to a different event loop

/usr/lib/python3.10/asyncio/mixins.py:30: RuntimeError
----------------------------------------------------------------------- Captured stdout teardown ------------------------------------------------------------------------
Teardown
Dropping all tables
Creating all tables
Environment variables loaded from .env
Prisma schema loaded from prisma/schema.prisma
Datasource "db": PostgreSQL database "mafflle_auth", schema "public" at "localhost:5432"

⚠️  There might be data loss when applying the changes:

  • The values [ADMIN] on the enum `UserType` will be removed. If these variants are still used in the database, this will fail.


----------------------------------------------------------------------- Captured stderr teardown ------------------------------------------------------------------------
Error: Use the --accept-data-loss flag to ignore the data loss warnings like prisma db push --accept-data-loss
=========================================================================== warnings summary ============================================================================
venv/lib/python3.10/site-packages/aioredis/connection.py:11
  /home/sheyzi/code/mafflle/mafflle_backend/venv/lib/python3.10/site-packages/aioredis/connection.py:11: DeprecationWarning: The distutils package is deprecated and slated for removal in Python 3.12. Use setuptools or check PEP 632 for potential alternatives
    from distutils.version import StrictVersion

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
======================================================================== short test summary info ========================================================================
FAILED server/tests/test_authentication/test_users.py::test_signup_user - RuntimeError: <asyncio.locks.Event object at 0x7fb442dda980 [unset]> is bound to a different...
FAILED server/tests/test_authentication/test_users.py::test_login_user - RuntimeError: <asyncio.locks.Event object at 0x7fb44038b7c0 [unset]> is bound to a different ...
================================================================ 2 failed, 4 passed, 1 warning in 9.72s =================================================================

The first test always take way more time than the others

Hallo

Answer 1

The event_loop and app fixtures have different scopes (respectively "module" and "session". The FastAPI app may be created using a different loop. I suggest to use the "session" scope for the event_loop fixture and to declare it on top of your test module.

Moreover, I am not sure what is the behavior of using a coroutine when the loop reference is required by synchronous fixtures. If it still doesn't work with the "session" scope, try with a synchronous fixture:

# declare this before any other fixtures with the "session" scope
# that may reference the event loop
@pytest.fixture(scope="session")
def event_loop():
    return asyncio.get_event_loop()

Fusibility

Just saw your answer after the page refreshed when I submitted mine... What a coincidence, a year and 5 months without an answer, and then two within a day! I think your suggestion is a good one, although the fastapi.testclient.TestClient will also be creating a separate event loop which was not immediately obvious to me. – Cairo

event_loop function is deprecated now in fastapi, but the rest of the answer is still relevant. I had to make sure that all my dependencies are using the same scope so that they share the same event loop. – Billbug

Answer 2

For anyone else arriving at this with similar problems (...bound to a different event loop) when testing a FastAPI app. I'm not 100% sure my solution solves exactly the issue in the question, but is very likely closely related, and likely useful to anyone else coming across this thread.

After trying few different approaches, the solution I came to in the end was simply to use the TestClient from the async-asgi-testclient package. It can almost directly be used as a drop-in replacement of the fastapi.testclient.TestClient, but it runs in the same event loop as the tests rather than creating a standalone event loop like the included TestClient, avoiding the ...different event loop errors I was seeing at least.

For me it was as simple as replacing:

@pytest.fixture
def sync_test_client(app):
    from fastapi.testclient import TestClient
    with TestClient(app) as client:
        yield client

with

@pytest_asyncio.fixture
async def async_asgi_test_client(app):
    from async_asgi_testclient import TestClient
    async with TestClient(app) as client:
        yield client

Then, using the async test client instead of the sync test client in my tests.

Another option I came across was to use the httpx AsyncClient, however, that is a more bare-bones approach, and does not include handling app lifespan or testing websocket connections out of the box (as far as I could tell).

Cairo

Thank you, it works amazing! Do note that you need to await requests made from the async client. You can use it when you really need to work in the same event loop as the FastAPI application – Rowen