CODEWITHSUNDEEP
pythonsqlalchemypytestpython-asyncio
alphaomega
alphaomega

Reputation: 187

How to call multiple SQLAlchemy `asyncio.run()` functions from within the same test in Pytest?

Using Pytest 6.2.5 and SQLAlchemy 1.4 async engine with Postgres backend.

I have two tests that test simple update and read functions. They are SQLAlchemy asyncio methods. The update test updates a value in the database and the read function asserts that the newly written value is present. Currently these are two separate tests which both pass. I want to move them into a single test, since they are interdependent.

Placing these two functions into a single test causes the test to fail. I can see that it is related to the async functionality, but I thought the @pytest.mark.asyncio decorator and the asyncio.run() async-specific functionality would account for that.

These two tests pass:

@pytest.mark.asyncio
def test_update():
    return asyncio.run(update('appuser', {'name': 'Alice'}, {'org': 'Wonderland'}))


@pytest.mark.asyncio
def test_read():
    read = asyncio.run(read('appuser', query={'name': 'Alice'}, mods={'fields': ['org']}))
    assert read == [('Wonderland',)]

This fails

@pytest.mark.asyncio
def test_read():
    asyncio.run(update('appuser', {'name': 'Alice'}, {'org': 'Wonderland'}))
    read = asyncio.run(read('appuser', query={'name': 'Alice'}, mods={'fields': ['org']}))
    assert read == [('Wonderland',)]

With this error:

>   ???
E   RuntimeError: Task <Task pending coro=<PostgresDb.read() running at file.py:262> cb=[_run_until_complete_cb() at /usr/lib/python3.7/asyncio/base_events.py:158]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop

asyncpg/protocol/protocol.pyx:338: RuntimeError

I tried placing the test_update as a function within the test_read function and calling it from within there like this, but the error is the same:

@pytest.mark.asyncio
def test_read():
    def test_update():
        asyncio.run(DB.update('appuser', {'name': 'Alice'}, {'org': 'Wonderland'}))
    test_update()
    read = asyncio.run(DB.read('appuser', query={'name': 'Alice'}, mods={'fields': ['org']}))
    assert read == [('Wonderland',)]

How can we run multiple asyncio.run() functions inside of a single Pytest?

Upvotes: 1

Views: 568

Answers (1)

Bharel
Bharel

Reputation: 26900

You're mixing regular functions together with async ones.

Using @pytest.mark.asyncio requires an async function (async def).

Also, instead of using asyncio.run() which creates a new loop and can only be used once, try awaiting like so:

@pytest.mark.asyncio
async def test_read():
    await update('appuser', {'name': 'Alice'}, {'org': 'Wonderland'})
    read = await read('appuser', query={'name': 'Alice'}, mods={'fields': ['org']})
    assert read == [('Wonderland',)]

Upvotes: 2

Related Questions