I'm creating small educational app using fastapi and aiopg for connectioon to db. It remains to write only the tests. I try to use pytest-asyncio and dependency overriding for connection to my test_db.
@pytest.fixture(scope="session", autouse=True)
def setup_test_db():
create_db()
sql_script_path = "./db.sql"
try:
conn = ps.connect(DSN_TEST)
cursor = conn.cursor()
with open(sql_script_path, "r") as script_file:
sql_script = script_file.read()
cursor.execute(sql_script)
conn.commit()
conn.close()
except ps.Error as e:
raise Exception(f"Error initializing the test database: {e}")
app.dependency_overrides[get_database_pool] = get_test_database_pool
app.dependency_overrides[get_db] = get_test_db
yield
delete_db()
@pytest.fixture
async def ac():
async with AsyncClient(app=app, base_url="http://test") as ac:
yield ac
and test itself
async def test_root(ac: AsyncClient):
response = await ac.get("/orders/")
assert response.status_code == 200
I specified in pytest.ini asyncio_mode = True, run pytest and get next error.
pool = await aiopg.create_pool(DSN_TEST) # type: ignore
....\\Lib\\site-packages\\aiopg\\pool.py:300: in from_pool_fill
await self.\_fill_free_pool(False)
....\\Lib\\site-packages\\aiopg\\pool.py:336: in \_fill_free_pool
conn = await connect(
....\\Lib\\site-packages\\aiopg\\connection.py:65: in connect
connection = Connection(
....\\Lib\\site-packages\\aiopg\\connection.py:772: in __init__
self.\_loop.add_reader(
_______________________________________________________________________
self = \<ProactorEventLoop running=False closed=False debug=False\>, fd = 884, callback = \<function Connection.\_ready at 0x0000011210F49080\>
args = (\<weakref at 0x000001121331D080; to 'Connection' at 0x0000011213327990\>,)
def add_reader(self, fd, callback, *args):
> raise NotImplementedError
E NotImplementedError
........\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\asyncio\\events.py:530: NotImplementedError\
Maybe someone know correct way of doing this?
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
I put this line before the test, and it fixes this up