根据 FastAPI 文档 我可能需要使用 LifespanManager。有人可以向我展示如何在异步测试中使用 LifespanManager 的示例吗?就像,这样的寿命:
@asynccontextmanager
async def lifespan(_app: FastAPI):
async with httpx.AsyncClient(base_url=env.proxy_url, transport=httpx.MockTransport(dummy_response)) as client:
yield {'client': client}
await client.aclose()
我正在尝试测试一个名为
proxy
的端点,它工作正常,但我需要进行回归测试:
import pytest
import pytest_asyncio
from fastapi import FastAPI
from contextlib import asynccontextmanager
from asgi_lifespan import LifespanManager
import httpx
from httpx import Response
import importlib
import uvicorn
import proxy
import env
def dummy_response(_request):
res = Response(200, content="Mock response")
res.headers['Content-Type'] = 'text/plain; charset=utf-8'
return res
@pytest_asyncio.fixture
async def mock_proxy():
importlib.reload(env)
importlib.reload(proxy)
@asynccontextmanager
async def lifespan(_app: FastAPI):
async with httpx.AsyncClient(base_url=env.proxy_url, transport=httpx.MockTransport(dummy_response)) as client:
yield {'client': client}
await client.aclose()
app = FastAPI(lifespan=lifespan)
app.add_route("/proxy/path", proxy.proxy)
async with LifespanManager(app) as manager:
yield app
@pytest_asyncio.fixture
async def _client(mock_proxy):
async with mock_proxy as app:
async with httpx.AsyncClient(app=app, base_url=env.proxy_url) as client:
yield client
@pytest.mark.anyio
async def test_proxy_get_request(_client):
async with _client as client:
response = await client.get(f"{env.proxy_url}/proxy/path", params={"query": "param"})
assert response.status_code == 200
这次尝试告诉我
TypeError:“FastAPI”对象不支持异步上下文管理器协议
这段代码看起来非常接近,但是状态的生命周期变化没有发生:
import pytest
import pytest_asyncio
from fastapi import FastAPI
from contextlib import asynccontextmanager
from asgi_lifespan import LifespanManager
import httpx
from httpx import Response
import importlib
import uvicorn
import proxy
import env
def dummy_response(_request):
res = Response(200, content="Mock response")
res.headers['Content-Type'] = 'text/plain; charset=utf-8'
return res
@pytest_asyncio.fixture
async def _client():
with pytest.MonkeyPatch.context() as monkeypatch:
monkeypatch.setenv("PROXY_URL", "http://proxy")
importlib.reload(env)
importlib.reload(proxy)
@asynccontextmanager
async def lifespan(_app: FastAPI):
async with httpx.AsyncClient(
base_url=env.proxy_url,
transport=httpx.MockTransport(dummy_response)) as client:
yield {'client': client} # startup
await client.aclose() # shutdown
app = FastAPI(lifespan=lifespan)
app.add_route("/proxy/path", proxy.proxy)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url=env.proxy_url) \
as client, LifespanManager(app):
yield client
@pytest.mark.asyncio
async def test_proxy_get_request(_client):
response = await _client.get(f"/proxy/path", params={"query": "param"})
assert response.status_code == 200
====================================================== ===简短的测试摘要信息============================================= ========= 失败的测试/回归/test_proxy.py::test_proxy_get_request - AttributeError:“状态”对象没有属性“客户端”
...事实上,LifespanManager 似乎并没有做很多工作。如果我将夹具的最后一部分更改为:
async with LifespanManager(app) as manager:
print(manager._state, app.state._state)
app.state = State(state=manager._state)
transport = httpx.ASGITransport(app=app)
print(manager._state, app.state.client)
async with httpx.AsyncClient(transport=transport, base_url=env.proxy_url) \
as client:
yield client
我得到:
------------------------------------------------- ---- 捕获的标准输出设置 ------------------------------------------------------ ---------- {'客户':
} {} {'客户': } =================================================== ==简短的测试摘要信息============================================== ======== 失败的测试/回归/test_proxy.py::test_proxy_get_request - AttributeError:“状态”对象没有属性“客户端”
因此,应用程序在启动时没有获取状态(但管理器是,它只是由于某种原因没有应用于应用程序)。同样,即使我自己手动设置状态(那么为什么此时还要使用 LifespanManager),代理函数的请求中的状态也不会像预期的那样可用。
我这样做的原因是代理中的第一行是:
async def proxy(request: Request):
client = request.state.client
这就是失败的地方。
感谢 Yurii 的评论,我解决了这个最初的问题,但不是最初导致我走上这条路的原因。我可以通过以下方式解决这个问题:
async with LifespanManager(app) as manager:
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=manager.app), base_url=env.proxy_url) as client:
yield client
然而,这一切都始于我最初使用 FastAPI 的 TestClient 的方法由于奇怪的取消而失败,触发任务组中未处理的问题,触发流耗尽然后尝试读取(如果存在流问题,则代理不起作用,但它确实起作用)。事实证明,FastAPI 不允许您使用 TestClient 进行异步测试,并推荐这种方法(有关更多信息,请参阅本文开头的链接)。我现在在这里遇到了同样的问题:
失败的测试/回归/test_proxy.py::test_proxy_get_request - 异常组:任务组中未处理的错误(1个子异常)
这是由相同的取消问题引起的:
self = <asyncio.locks.Event object at 0x105cdade0 [unset]>
async def wait(self):
"""Block until the internal flag is true.
If the internal flag is true on entry, return True
immediately. Otherwise, block until another coroutine calls
set() to set the flag to true, then return True.
"""
if self._value:
return True
fut = self._get_loop().create_future()
self._waiters.append(fut)
try:
await fut
E asyncio.exceptions.CancelledError:已被取消范围 105cdb8f0 取消
你应该从
manager.app
产生 mock_proxy
,而不仅仅是 app
在
_client
固定第一个上下文管理器 async with mock_proxy as app:
并将 mock_proxy
分配给 app
而不是 (app = mock_proxy
)
您也不需要
test_proxy_get_request
中的上下文管理器(删除 async with _client as client:
并使用 _client)