'Python async unit tests, how to use async database connection pool?

I'm using Python 3.9.7, databases 0.4.3, asyncpg 0.24.0.

Relevant files/snippets:

tests.py

from unittest import TextTestRunner, TestSuite, TestLoader

runner = TextTestRunner()
test_loader = TestLoader()
suite = TestSuite()
test_suite = add_tests_to_suite(suite)  # <- All my tests are added here

runner.run(test_suite)

db.py

from databases import Database
dal = Database()  # This is the "Databases" lib instance

some_test.py

from unittest import IsolatedAsyncioTestCase
from db import dal

class SomeTest(IsolatedAsyncioTestCase):
    async def some_async_test(self):
        try:
            await dal.connect()
            # Test logic happens here
        finally:
            await dal.disconnect()

The code above works, however, connecting and disconnecting on every unit test is taking around 400ms, which is very slow when dealing with a large amount of unit tests. What is the proper/recommended way of dealing with async database connections in the context of unit tests?

Things I tried:

  • Move dal.connect() to tests.py, but that file is not in the asyncio context, therefore I cannot await the connect() function.
  • Create an asyncio loop in tests.py just so I can await the connect() function, but this approach throws:

RuntimeWarning: coroutine 'IsolatedAsyncioTestCase._asyncioLoopRunner' was never awaited`

  • Run the function dal.connect() only once, rather than on every test, but it throws:

asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress



Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source