pytest-asyncio-concurrent 0.5.2

Key Features

• Giving the capability to run pytest async functions.

• Providing granular control over Concurrency – See Async Group for details

  • Specifying Async Group to control tests that can run together.

  • Limitation: Only test functions defined under same direct parent can be put into same group.

• asyncio_concurrent mark accept a timeout parameter, which would throw error when test reach the given time.

• Compatible with pytest-asyncio.

Key Concept: Async Group

The plugin control tests concurrency by putting them into different groups.

• Limitation: Each group can only contain tests under same direct parent. For example, class methods within same class, or functions within same file.

• Explicitly specify group by providing group parameter to asyncio_concurrent mark.

  • All tests marked with same group name will be executed together.

  • All tests will be skipped if the Same Parent rule got violated.

• Use default grouping strategy. The plugin accept --default-group-strategy cli parameter, or default_group_strategy in ini or toml file. Possible value: self, parent.

  • self(default): Each test will be executed by itself if no group provided.

  • parent: Tests will grouped by their parent node. For example, all method within same class will be grouped together.

Installation

You can install “pytest-asyncio-concurrent” via pip from PyPI:

$ pip install pytest-asyncio-concurrent

How this work?

This plugin extend pytest_runtestloop hook to handle async tests seperately.

In the ‘async loop’, tests are grouped by the group provided in the mark. and executed one group at a time.

In each group, instead of sequentially calling setup, call, teardown for individual test, these hooks are called for all tests in the group in batch.

The fixture lifecycle within group is handled by this plugin instead of pytest core, to work around pytest sequential test execution assumption.

Usage

Run test Sequentially

@pytest.mark.asyncio_concurrent
async def async_test_A():
    res = await wait_for_something_async()
    assert result.is_valid()

@pytest.mark.asyncio_concurrent
async def async_test_B():
    res = await wait_for_something_async()
    assert result.is_valid()

Run tests Concurrently

# the test below will run by itself
@pytest.mark.asyncio_concurrent
async def test_by_itself():
    res = await wait_for_something_async()
    assert result.is_valid()

# the two tests below will run concurrently
@pytest.mark.asyncio_concurrent(group="my_group")
async def test_groupA():
    res = await wait_for_something_async()
    assert result.is_valid()

# this one will have a 10s timeout
@pytest.mark.asyncio_concurrent(group="my_group", timeout=10)
async def test_groupB():
    res = await wait_for_something_async()
    assert result.is_valid()

Parametrized Tests

# the parametrized tests below will run sequential
@pytest.mark.asyncio_concurrent
@pytest.parametrize("p", [0, 1, 2])
async def test_parametrize_sequential(p):
    res = await wait_for_something_async()
    assert result.is_valid()

# the parametrized tests below will run concurrently
@pytest.mark.asyncio_concurrent(group="my_group")
@pytest.parametrize("p", [0, 1, 2])
async def test_parametrize_concurrent():
    res = await wait_for_something_async()
    assert result.is_valid()

Contributing

Contributions are very welcome. Tests can be run with tox, please ensure the coverage at least stays the same before you submit a pull request.

License

Distributed under the terms of the MIT license, “pytest-asyncio-concurrent” is free and open source software