I have a FastAPI application, I use SQLAlchemy 2.0 in async mode to connect to my postgres database.
This is my endpoint FastAPI:
@router.get("", response_model=HelloResponse)
async def get_hello(
current_user: UserResponse = Security(
get_current_active_user,
scopes=[Scoper.admin, Scoper.staff],
),
db=Depends(get_session),
):
hello = await HelloManager(db).get(id=1)
if hello.foo == "bar":
return hello
return await HelloManager(db).get(id=2)
My test : test_hello.py
@pytest.mark.asyncio
async def test_hello_with_mocked_security(async_client: AsyncClient):
app.dependency_overrides[get_current_user] = mock_get_current_user
response = await async_client.get("/api/hello", headers=headers_hocked_token)
assert response.status_code == 200
data = response.json()
assert data["id"] == 2
assert data["name"] = "Hello2 !!"
The test passed correctly but Coverage says that these 3 lines (after hello = await HelloManager(db).get(id=1) ) are not tested:
if hello.foo == "bar":
return hello
return await HelloManager(db).get(id=2)
I have the same situation in several places: the use of managers and everything that goes with it is said to be untested, but the tests pass with the right feedback.
Here's my conftest.py, with the fixtures. I think the problem must be here, I must have done something wrong, because to tell you the truth I had a hard time overriding my db=Depends(get_session),...
from typing import AsyncIterator
from my_app.db.database import get_session
from httpx import AsyncClient
import pytest
from my_app.db.models.base_model import BaseModelORM
from my_app.settings import settings
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from my_app.main import app
from my_app.settings import get_settings as original_get_settings
@pytest.fixture(scope="session", autouse=True)
async def init_database() -> None:
"""
Initialize the database before running the tests.
"""
engine = create_async_engine(
settings.sqlalchemy_database_uri_async or "",
echo=False,
)
async with engine.begin() as conn:
await conn.run_sync(BaseModelORM.metadata.drop_all)
await conn.run_sync(BaseModelORM.metadata.create_all)
@pytest.fixture(scope="function")
async def async_engine():
engine = create_async_engine(
settings.sqlalchemy_database_uri_async, # type: ignore
echo=False,
)
return engine
@pytest.fixture(scope="function")
def TestingSessionLocalAsync(async_engine):
return async_sessionmaker(
async_engine,
class_=AsyncSession,
expire_on_commit=False,
)
@pytest.fixture(scope="function", autouse=True)
async def db(async_engine, TestingSessionLocalAsync) -> AsyncIterator[AsyncSession]:
async with async_engine.connect() as conn:
async_session = TestingSessionLocalAsync()
yield async_session
await async_session.close()
await conn.close()
@pytest.fixture(scope="session", autouse=True)
def get_settings():
settings.fapi_mode = "TESTING"
settings.sqlalchemy_database_uri_async = (
f"postgresql+asyncpg://postgres:postgres@localhost:5432/test_myapp"
)
return settings
@pytest.fixture(scope="function")
def override_get_db(db: AsyncSession):
async def _override_get_db():
yield db
return _override_get_db
@pytest.fixture(scope="function")
async def async_client(override_get_db, get_settings):
app.dependency_overrides[get_session] = override_get_db
app.dependency_overrides[original_get_settings] = get_settings
async with AsyncClient(app=app, base_url="http://test.test") as ac:
yield ac
I hope someone can help me :)