'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()totests.py, but that file is not in the asyncio context, therefore I cannotawaittheconnect()function. - Create an asyncio loop in
tests.pyjust so I canawaittheconnect()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 |
|---|
