我正在尝试FastAPI
使用测试构建完全覆盖的应用程序python 3.9
为此我选择了堆栈:FastAPI、uvicorn、SQLAlchemy、asyncpg、pytest(+ async、cov 插件)、覆盖率和 httpx AsyncClient
这是我的最低要求.txt
所有测试运行顺利,我得到了预期的结果。但是我遇到了这个问题,覆盖范围没有正确收集。await
当协程将控制权返回给事件循环时,它在第一个关键字之后中断
这是关于如何重现此行为的最小设置(也可以在 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
像魅力这样的作品