在 Pytest 测试函数中缓存异步请求
Caching async requests in Pytest test function
我在 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_cache
,但似乎没有用。
# 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__
,它来自 conftest.py
中的 pytest.fixture
,它的范围是 'session':
# 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
放开不透明的第 3 方缓存,自己缓存吧。
由于您不需要在单次执行期间清理缓存,因此可以使用普通字典:
_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]
我在 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_cache
,但似乎没有用。
# 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__
,它来自 conftest.py
中的 pytest.fixture
,它的范围是 'session':
# 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
放开不透明的第 3 方缓存,自己缓存吧。 由于您不需要在单次执行期间清理缓存,因此可以使用普通字典:
_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]