How to fetch a record from db asynchronously while unit testing?

user_92153 0
Viewed 451

In this unit test I would like to check if the device has been created and that the expiry date is 7 days in future.

database.py

import databases
database = databases.Database(settings.sqlalchemy_database_uri)

Unit Test:

from database.database import database

def test_successful_register_expiry_set_to_seven_days():
    response = client.post(
        "/register/",
        headers={},
        json={"device_id": "u1"},
    )
    assert response.status_code == 201
    query = device.select(whereclause=device.c.id == "u1")
    d = database.fetch_one(query)
    assert d.expires_at == datetime.utcnow().replace(microsecond=0) + timedelta(days=7)

Because d is a coroutine object it fails with the message:

AttributeError: 'coroutine' object has no attribute 'expires_at'

And I can't use await inside a unit test.

d = await database.fetch_one(query)

What am I missing, please?

fastapi sqlalchemy python
1 Answers

Well it is never awaited, your code returns the coroutine before that semaphore going in to the scheduler.

If you are using an asynchronous driver, you need to await it. Are there any workarounds for this? Yes.

You can use asyncio.run(awaitable) to run the coroutine inside an event loop.

import asyncio

d = asyncio.run(database.fetch_one(query))

If you have an currently running event loop you may want to use that event loop instead. You can achieve that by asyncio.get_event_loop(), which will run the function inside the running loop.

import asyncio

asyncio.get_event_loop().run_until_complete(database.fetch_one(query))

You can also use @pytest.mark.asyncio decorator (see documentation).

@pytest.mark.asyncio
async def dummy():
    await some_awaitable()
user_131611550
Related