【发布时间】:2021-10-08 07:44:17
【问题描述】:
我正在尝试使用 python 3.9 构建完全覆盖测试的 FastAPI 应用程序
为此,我选择了堆栈:
FastAPI、uvicorn、SQLAlchemy、asyncpg、pytest(+ async、cov 插件)、coverage 和 httpx AsyncClient
这是我最小的requirements.txt
所有测试都运行顺利,我得到了预期的结果。
但是我遇到了这个问题,覆盖范围没有正确收集。它在第一个 await 关键字之后中断,此时协程将控制权返回给事件循环
这是关于如何重现此行为的最小设置 (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就像魅力一样工作
【问题讨论】:
标签: python sqlalchemy python-asyncio fastapi pytest-asyncio