[FIXED] Caching async requests in Pytest test function

Issue

I have implemented a test function in pytest which loads data from files, casts it into Python objects and provides a new object for each test.

Each one of these objects contains a request I need to make to the server and the expected responses, the function looks like this:

@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...

Many times I send many requests that contain the same http endpoint with the same body so the response is the same, but I'm testing for different things in the response.

Because of that I thought of implementing an in-memory cache so that for each test run the same requests won't be implemented twice.

What I've tried to do is create a request object, with its own __hash__ implementation and use the @asyncstdlib.lru_cache on the function, it didn't seem to work.

# 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)

The client I'm using: httpx.AsyncClient also implements __hash__, it's coming from a pytest.fixture in conftest.py and it has a scope of '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

Solution

Just let go of the opaque 3rd party cache, and cache yourself. Since you don't require cleaning-up the cache during a single execution, a plain dictionary will work:

_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]

Answered By - jsbueno