如何在测试FastAPI应用程序时触发寿命启动和关闭？
如何在测试FastAPI应用程序时触发寿命启动和关闭？
提问于 2020-11-28 16:01:06
作为FastAPI的新手，我不愿意测试比我在教程中看到的更困难的代码。我使用fastapi_cache模块和Redis，如下所示：
from fastapi import Depends, FastAPI, Query, Request
from fastapi_cache.backends.redis import CACHE_KEY, RedisCacheBackend
from fastapi_cache import caches, close_caches
app = FastAPI()
def redis_cache():
return caches.get(CACHE_KEY)
@app.get('/cache')
async def test(
cache: RedisCacheBackend = Depends(redis_cache),
n: int = Query(
...,
gt=-1
)
):
# code that uses redis cache
@app.on_event('startup')
async def on_startup() -> None:
rc = RedisCacheBackend('redis://redis')
caches.set(CACHE_KEY, rc)
@app.on_event('shutdown')
async def on_shutdown() -> None:
await close_caches()test_main.py看起来是这样的：
import pytest
from httpx import AsyncClient
from .main import app
@pytest.mark.asyncio
async def test_cache():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/cache?n=150")当我运行pytest时，它将cache变量设置为None，然后测试失败。我想我理解为什么代码不起作用了。但是，如何修复它以正确测试缓存呢？
回答 1
Stack Overflow用户
回答已采纳
发布于 2020-11-29 14:25:30
关键是httpx没有实现生存期协议并触发startup事件处理程序。为此，您需要使用LifespanManager。
安装：pip install asgi_lifespan
代码应该是这样的：
import pytest
from asgi_lifespan import LifespanManager
from httpx import AsyncClient
from .main import app
@pytest.mark.asyncio
async def test_cache():
async with LifespanManager(app):
async with AsyncClient(app=app, base_url="http://localhost") as ac:
response = await ac.get("/cache")页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/65051581
复制相关文章
相似问题
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号