0

我已经实现了一个测试函数，pytest它从文件中加载数据，将其转换为 Python 对象，并为每个测试提供一个新对象。

这些对象中的每一个都包含我需要向服务器发出的请求和预期的响应，函数如下所示：

@pytest.mark.asyncio
@pytest.mark.parametrize('test', TestLoader.load(JSONTest, 'json_tests'))
async def test_json(test: JSONTest, groups: Set[TestGroup], client: httpx.AsyncClient):
    skip_if_not_in_groups(test, groups)

    request = Request(url=test.url, body=test.body.dict())
    response = await client.post(request.url, json=request.body)

    # Assertions down here...

很多时候，我发送了许多包含相同http端点的请求，body因此响应相同，但我正在测试响应中的不同内容。

正因为如此，我想实现一个内存缓存，这样每次测试运行时，相同的请求就不会被实现两次。

我试图做的是创建一个请求对象，它有自己的__hash__实现并使用@asyncstdlib.lru_cacheon 函数，它似乎不起作用。

# Does not work...

@asyncstdlib.lru_cache
async def send_request(request: Request, client: httpx.AsyncClient):
    return await client.post(request.url, json=request.body)


@pytest.mark.asyncio
@pytest.mark.parametrize('test', TestLoader.load(JSONTest, 'json_tests'))
async def test_json(test: JSONTest, groups: Set[TestGroup], client: httpx.AsyncClient):
    skip_if_not_in_groups(test, groups)

    request = Request(url=test.url, body=test.body.dict())
    response = await send_request(request)

我正在使用的客户端：httpx.AsyncClient也实现__hash__了，它来自一个pytest.fixtureinconftest.py并且它的范围为“会话”：

# conftest.py

@pytest.fixture(scope='session')
def event_loop(request):
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope='session')
async def client() -> httpx.AsyncClient:
    async with httpx.AsyncClient() as client:
        yield client
4

1 回答 1

0

只需放开不透明的 3rd 方缓存，自己缓存即可。由于您不需要在单次执行期间清理缓存，因此可以使用普通字典：

_cache = {}


async def send_request(request: Request, client: httpx.AsyncClient):
    if request.url not in _cache:
        _cache[request.url] = await client.post(request.url, json=request.body)
    return _cache[request.url]
于 2022-01-12T12:58:54.767 回答