Bug: the app for test client is running in a different event loop

[orsinium]

Description

I've been trying to make rollbacks for database changes for the SQLAlchemy plugin. See the discussion:

https://github.com/orgs/litestar-org/discussions/1919

And I've encountered the issue described in the discussion above and in this issue:

https://github.com/encode/starlette/issues/1315

RuntimeError: Task <Task pending name='anyio.from_thread.BlockingPortal._call_func' coro=<BlockingPortal._call_func() running at /.../anyio/from_thread.py:219> cb=[TaskGroup._spawn.<locals>.task_done() at /.../anyio/_backends/_asyncio.py:726]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop

After diving into the issue deeper, I've discovered that BaseTestClient.portal always creates the portal in a separate thread and runs in this thread a new event loop. This behavior is described in the anyio.from_thread.start_blocking_portal function docstring:

Start a new event loop in a new thread and run a blocking portal in its main task.

That means if I create in tests the database connection before calling AsyncTestClient.__aenter__ (or TestClient.__enter__), it will be running in the main thread pytest-asyncio event loop and so it cannot be used as SQLAlchemyAsyncConfig.session_maker. And if you try, with the power of monkey patching, to create the database connection after entering the client context, you cannot use this database connection in the main thread in tests.

URL to code causing the issue

No response

MCVE

No response

Steps to reproduce

No response

Screenshots

No response

Logs

No response

Litestar Version

2.0.0b2

Platform

• [X] Linux
• [ ] Mac
• [ ] Windows
• [ ] Other (Please specify in the description above)

Funding

• You can sponsor this specific effort via a Polar.sh pledge below
• We receive the pledge once the issue is completed & verified

[Goldziher]

@cofin can you check it out?

[provinzkraut]

I think this might simply be resolved by structuring your tests a bit differently. We also run a whole bunch of tests involving SQLAlchemy and are not encountering that issue. These tests are carefully constructed as to handle the pytest-asyncio pitfalls.

Could you maybe provide a snippet of how you set up your tests so we can help you figure this out?

[orsinium]

Thank you for your response!

I "solved" the issue by migrating to FastAPI. I had the same issue on FastAPI when using their sync TestClient and I solved it by using instead httpx.AsyncClient which does not spawn a new thread to run the service. I can't check anymore if that works with Litestar as well, so I'll close the issue. If anyone faced the same problem, drop a message in this thread, and I'll reopen the issue for you.

[nrbnlulu]

I have this one as well.

[euri10]

I have this one as well.

Can you provide a MVCE @nrbnlulu please so that we can see what's going on ?

[nrbnlulu]

The Issue came from using sqla.AsyncSession.begin() as a scoped dependency. I have now fixed it.

[sylvainOL]

Hello, I've got a similar issue with piccolo orm and litestar

I've got this fixture for creating a fresh db at start:

@pytest.fixture(scope='session', autouse=True)
async def _setup_db() -> None:
    await drop_db_tables(MyModel)
    await create_db_tables(MyModel)

I've got this fixture in order to use transactions in piccolo:

@pytest.fixture(autouse=True)
async def _piccolo_transaction(logging_config: LoggingConfig) -> None:
    """Fixture for Piccolo transactions.

    It should be as simple as::

        async with DB.transaction():
            yield

    However, pytest-asyncio doesn't pass down contextvars, which is how Piccolo
    handles transactions.

    https://github.com/pytest-dev/pytest-asyncio/issues/127

    For now, this is the workaround.
    """
    logging_config.configure()()
    get_running_loop()
    database: PostgresEngine = engine_finder()

    connection = await database.get_new_connection()

    transaction = database.transaction()
    transaction.connection = connection
    transaction.transaction = transaction.connection.transaction()
    await transaction.begin()

    class TransactionProxy:
        def get(self) -> PostgresTransaction:
            return transaction

    database.current_transaction = TransactionProxy()

    yield

    await transaction.rollback()
    await connection.close()

I've got this fixture in order to create my async test client:

@pytest.fixture()
async def test_client(logging_config: LoggingConfig) -> AsyncTestClient:
    app = Litestar(
        route_handlers=[
            admin,
            reserve,
            release,
            test_ok,
        ],
        logging_config=logging_config,
    )
    return AsyncTestClient(app=app)

When I run tests on models / business logic without using Litestar, everything works

but when I try to test my endpoints, I've got the same error as you:

here's a test example:

class TestReserveEndpoint:
    async def test_ok(self, test_client: AsyncTestClient) -> None:
        my_models = await MyModel.objects()
        assert len(my_models) == 0
        async with test_client as client:
            response = await client.get('/api/main?model=my-model')
            assert response.text == 'ok'
            assert response.status_code == HTTP_200_OK
        my_models = await MyModel.objects()
        assert len(my_models) == 1

function is looking like (don't blame the uglyness of the API, I'm taking back the project...):

@get('/api/main')
async def reserve(
    models_list: Annotated[str, Parameter(pattern=r'^([\w-]+,*)+$')]) -> str:
    reserver = await MyModelReserver.from_string(models_list)
    is_reserved = await reserver.reserve()
    if is_reserved:
        return 'ok'
    return 'wait'

I guess I could mock MyModelReserver.from_string (which is tested elsewhere) but I wanted to be basic...

In the test, I've got as error:

RuntimeError: Task <Task pending name='anyio.from_thread.BlockingPortal._call_func' coro=<BlockingPortal._call_func() running at /Users/sylvaindesbureaux/Sources/Sirius/device-concurrency/.venv/lib/python3.12/site-packages/anyio/from_thread.py:217> cb=[TaskGroup._spawn.<locals>.task_done() at /Users/sylvaindesbureaux/Sources/Sirius/device-concurrency/.venv/lib/python3.12/site-packages/anyio/_backends/_asyncio.py:699]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop

any idea how I could make it work?

thanks!

[sylvainOL]

well I found a way to do it, but I don't know what I'm doing ^_^

I've replaced the fixture for test client:

@pytest.fixture()
async def test_httpx(logging_config: LoggingConfig) -> AsyncClient:
    app = Litestar(
        route_handlers=[
            admin,
            reserve,
            release,
            test_ok,
        ],
        logging_config=logging_config,
    )
    transport = ASGITransport(app=app)
    return AsyncClient(transport=transport, base_url='http://testserver')

and changed test for typing:

class TestReserveEndpoint:
    async def test_ok(self, test_client: AsyncClient) -> None:
        my_models = await MyModel.objects()
        assert len(my_models) == 0
        async with test_client as client:
            response = await client.get('/api/main?model=my-model')
            assert response.text == 'ok'
            assert response.status_code == HTTP_200_OK
        my_models = await MyModel.objects()
        assert len(my_models) == 1

and it works!

If that can help somebody!

[gsakkis]

I'm getting the same RuntimeError as @orsinium and @sylvainOL with Beanie/Pymongo when trying to use Litestar's AsyncTestClient. The httpx.AsyncClient solution of @sylvainOL seems to work but on closer inspection it is not quite equivalent to non-test usage; e.g. the app's lifespan and on_startup hooks are not called.

I think this might simply be resolved by structuring your tests a bit differently. We also run a whole bunch of tests involving SQLAlchemy and are not encountering that issue. These tests are carefully constructed as to handle the pytest-asyncio pitfalls.

@provinzkraut could you share a link to some of the tests that use AsyncTestClient with a db (SQLAlchemy or otherwise)? Bonus if the app initializes the db with a lifespan and/or on_startup hooks.

[olzhasar]

I'm building an app with multiple websocket connections interacting with each other, and I am facing the very same issue. Kindly asking to reopen it @orsinium.

Here is a simple code example that demonstrates the issue:

async def test_multiple_clients_event_loop() -> None:
    @get("/")
    def return_loop_id() -> dict:
        return {"loop_id": id(asyncio.get_event_loop())}

    app = Litestar(route_handlers=[return_loop_id])

    async with AsyncTestClient(app) as client_1, AsyncTestClient(app) as client_2:
        response_1 = await client_1.get("/")
        response_2 = await client_2.get("/")

    assert response_1.json() == response_2.json()  # FAILS

The idea of using a portal in a sync test client is understandable, but there probably should be an alternative implementation for the async client, that doesn't delegate anything to a different thread