如何使用LifespanManager在FastAPI中测试反向代理（异步测试）

Question

根据 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

这就是失败的地方。

编辑2：

感谢 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 取消

Answer 1

1. 你应该从

   manager.app

   产生

   mock_proxy

   ，而不仅仅是

   app

2. 在

   _client

   固定第一个上下文管理器

   async with mock_proxy as app:

   并将

   mock_proxy

   分配给

   app

   而不是 (

   app = mock_proxy

   )

3. 您也不需要

   test_proxy_get_request

   中的上下文管理器（删除

   async with _client as client:

   并使用 _client）