FastAPI、SQLAlchemy、pytest,无法获得 100% 的覆盖率,未正确收集
FastAPI, SQLAlchemy, pytest, unable to get 100% coverage, it doesn't properly collected
我正在尝试使用 python 3.9
构建完全覆盖测试的 FastAPI
应用程序
为此,我选择了堆栈:
FastAPI、uvicorn、SQLAlchemy、asyncpg、pytest(+ async、cov 插件)、coverage 和 httpx AsyncClient
这是我的最小值requirements.txt
所有测试 运行 顺利,我得到了预期的结果。
但我遇到了问题,覆盖范围没有正确收集。它在第一个 await
关键字后中断,当协程 returns 控制回到事件循环
这里是关于如何重现此行为的最小集合 (it's also available on a GitHub)。
申请代码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"}
测试设置 conftest.py
如下所示:
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()
测试很简单(只需检查状态码为 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
结果我得到:
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%
- 上面的示例代码使用
aiosqlite
而不是 asyncpg
但覆盖失败也会持续重现
- 我断定这个问题出在
SQLAlchemy
,因为这个使用 asyncpg
而不使用 SQLAlchemy
的例子就像 charm
这是 SQLAlchemy 1.4 在 coveragepy 中的一个问题:https://github.com/nedbat/coveragepy/issues/1082, https://github.com/nedbat/coveragepy/issues/1012
你可以试试 --concurrency==greenlet
选项
我正在尝试使用 python 3.9
构建完全覆盖测试的 FastAPI
应用程序
为此,我选择了堆栈:
FastAPI、uvicorn、SQLAlchemy、asyncpg、pytest(+ async、cov 插件)、coverage 和 httpx AsyncClient
这是我的最小值requirements.txt
所有测试 运行 顺利,我得到了预期的结果。
但我遇到了问题,覆盖范围没有正确收集。它在第一个 await
关键字后中断,当协程 returns 控制回到事件循环
这里是关于如何重现此行为的最小集合 (it's also available on a GitHub)。
申请代码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"}
测试设置 conftest.py
如下所示:
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()
测试很简单(只需检查状态码为 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
结果我得到:
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%
- 上面的示例代码使用
aiosqlite
而不是asyncpg
但覆盖失败也会持续重现 - 我断定这个问题出在
SQLAlchemy
,因为这个使用asyncpg
而不使用SQLAlchemy
的例子就像 charm
这是 SQLAlchemy 1.4 在 coveragepy 中的一个问题:https://github.com/nedbat/coveragepy/issues/1082, https://github.com/nedbat/coveragepy/issues/1012
你可以试试 --concurrency==greenlet
选项