pythonsqlalchemypython-asynciofastapipytest-asyncio

FastAPI, SQLAlchemy, pytest, unable to get 100% coverage, it doesn't properly collected


I'm trying to build FastAPI application fully covered with test using python 3.9 For this purpose I've chosen stack: FastAPI, uvicorn, SQLAlchemy, asyncpg, pytest (+ async, cov plugins), coverage and httpx AsyncClient

Here is my minimal requirements.txt

All tests run smoothly and I get the expected results. But I've faced the problem, coverage doesn't properly collected. It breaks after a first await keyword, when coroutine returns control back to the event loop

Here is a minimal set on how to reproduce this behavior (it's also available on a GitHub).

Appliaction code main.py:

import sqlalchemy as sa
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from starlette.requests import Request

app = FastAPI()
DATABASE_URL = 'sqlite+aiosqlite://?cache=shared'


@app.on_event('startup')
async def startup_event():
    engine = create_async_engine(DATABASE_URL, future=True)
    app.state.session = AsyncSession(engine, expire_on_commit=False)
    app.state.engine = engine


@app.on_event('shutdown')
async def shutdown_event():
    await app.state.session.close()


@app.get('/', name="home")
async def get_home(request: Request):
    res = await request.app.state.session.execute(sa.text('SELECT 1'))
    # after this line coverage breaks
    row = res.first()
    assert str(row[0]) == '1'
    return {"message": "OK"}

test setup conftest.py looks like this:

import asyncio

import pytest
from asgi_lifespan import LifespanManager
from httpx import AsyncClient


@pytest.fixture(scope='session')
async def get_app():
    from main import app
    async with LifespanManager(app):
        yield app


@pytest.fixture(scope='session')
async def get_client(get_app):
    async with AsyncClient(app=get_app, base_url="http://testserver") as client:
        yield client


@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

test is simple as it is (just check status code is 200) test_main.py:

import pytest
from starlette import status


@pytest.mark.asyncio
async def test_view_health_check_200_ok(get_client):
    res = await get_client.get('/')
    assert res.status_code == status.HTTP_200_OK
pytest -vv --cov=. --cov-report term-missing --cov-report html

As a result coverage I get:

Name           Stmts   Miss  Cover   Missing
--------------------------------------------
conftest.py       18      0   100%
main.py           20      3    85%   26-28
test_main.py       6      0   100%
--------------------------------------------
TOTAL             44      3    93%

enter image description here

  1. Example code above uses aiosqlite instead of asyncpg but coverage failure also reproduces persistently
  2. I've concluded this problem is with SQLAlchemy, because this example with asyncpg without using the SQLAlchemy works like charm

Solution

  • it's an issue with SQLAlchemy 1.4 in coveragepy: https://github.com/nedbat/coveragepy/issues/1082, https://github.com/nedbat/coveragepy/issues/1012

    you can try with --concurrency==greenlet option