pytest async function as the argument for parameterize

Question

I have an async function which computes a list of urls asynchronously, and I want to use parameterize to generate an async test for each of those urls , which will assert the status code. What I'm trying to do is something like this:

import pytest

@pytest.fixture
async def compute_urls():
    urls = await compute_urls_helper()
    return urls

@pytest.mark.asyncio 
@pytest.mark.parameterize('url',await compute_urls()) 
async def test_url(compute_urls,url):
    resp = await get_url(url)
    assert resp.status_code == 200

I know that using 'await' inside the parameterize is not possible, so I would love to hear suggestions for this kind of operation.

Answer 1

You can use asyncio.run to create an event loop just to compute the params:

import asyncio
from unittest.mock import AsyncMock
import pytest


async def compute_urls_helper():
    return ["stackoverflow.com", "jooj.com"]


async def get_url(url: str) -> AsyncMock:
    return AsyncMock(status_code=200)


@pytest.mark.asyncio
@pytest.mark.parametrize("url", asyncio.run(compute_urls_helper()))
async def test_url(url):
    resp = await get_url(url)
    assert resp.status_code == 200

However I wouldn't recommend to use this method frequently, because as stated in the docs:

This function (asyncio.run) always creates a new event loop and closes it at the end. It should be used as a main entry point for asyncio programs, and should ideally only be called once.

Therefore, you can create a session scoped fixture so that you can reuse a fixture containing each url object like so:

import asyncio
from unittest.mock import AsyncMock
import pytest


async def compute_urls_helper():
    return ["google.com", "jooj.com"]

#Can be moved to conftest.py in the root of the project
@pytest.fixture(scope="session", params=asyncio.run(compute_urls_helper()))
def url(request):
    return request.param

async def get_url(url: str) -> AsyncMock:
    return AsyncMock(status_code=200)


@pytest.mark.asyncio
async def test_url(url):
    resp = await get_url(url)
    assert resp.status_code == 200