AsyncIO Support

Testix offers support for testing asynchronous code, that is code which takes advantage of Python’s async and await keywords.

Testix’s async support has been tested to work with pytest-asyncio.

AsyncIO Expectations

You can specify that a method call should be awaited using the __await_on__ modifier of the Scenario, as in the example below. As you can see, you may also mix and match sync and async expectations.

from testix import *
import pytest


@pytest.mark.asyncio
async def test_async_expectations():
    with scenario.Scenario('awaitable test') as s:
        s.__await_on__.my_fake('some data') >> fake.Fake('another_fake')
        s.__await_on__.another_fake() >> fake.Fake('yet_another')
        s.__await_on__.yet_another() >> Fake('last_one')
        s.last_one.sync_func(1, 2, 3) >> 'sync value'

        assert await my_code(Fake('my_fake')) == 'sync value'


async def my_code(thing):
    another = await thing('some data')
    yet_another = await another()
    last_one = await yet_another()
    return last_one.sync_func(1, 2, 3)

Note that the test function itself is async and that you have to use the pytest.mark.asyncio decorator on the test - this decorator makes sure the test runs inside an asyncio event loop.

Note that the __await_on__ changes the expectation .my_fake('some data') into two expectations - the function call, and the use of await-ing. You can see this, if, e.g., you cut my_code(thing) short by replacing its first line with return 'sync value'. This is the correct value, so the assert statement passes, however the Scenario context will inform you that

E       Scenario ended, but not all expectations were met. Pending expectations (ordered):
[my_fake('some data'), await on my_fake('some data')@cb28287a42fc(),
another_fake(), await on another_fake()@94bb8afd7e45(),
yet_another(), await on yet_another()@b09a73bf3ee0(),
last_one.sync_func(1, 2, 3)]

You can see that every __await_on__ results in a special expectation representing it.

AsyncIO Context Managers

You can specify your expectation for an object to be used as an async context manager (i.e. in an async with statement) by using the __async_with__ modifier. Here’s an example testing a module called async_read which has an async function go() which reads the contents of a file asynchronously.

from testix import *
import pytest

import async_read


@pytest.fixture(autouse=True)
def override_import(patch_module):
    patch_module(async_read, 'aiofiles')


@pytest.mark.asyncio
async def test_read_write_from_async_file():
    with scenario.Scenario() as s:
        s.__async_with__.aiofiles.open('file_name.txt') >> Fake('the_file')
        s.__await_on__.the_file.read() >> 'the text'

        assert 'the text' == await async_read.go('file_name.txt')

Note our use of patch_module to mock the aiofiles library, which we assume is imported and used by our async_read module.

The code which passes this test is

import aiofiles


async def go(filename):
    async with aiofiles.open(filename) as f:
        return await f.read()

Note you do not have to specify a return value with >> for the __async_with__ expectation if you want to use the “anonymous” form of the async with statement:

async with lock(): # no "as" part
    await handle_critical_data()

AsyncIO Async For Loops

You can specify your expectation for an object to be used in an async for loop by using the __async_for__ modifier. This allows you to test code that uses async iterators. Here’s an example of a test with the code that passes it:

from testix import *
import pytest


@pytest.mark.asyncio
async def test_async_iterator():
    with scenario.Scenario() as s:
        s.__async_for__.my_iterator >> ['a', 'b', 'c']

        async_capitalizer = AsyncCapitalizer(Fake('my_iterator'))
        result = []
        async for item in async_capitalizer:
            result.append(item)
        assert result == ['A', 'B', 'C']


class AsyncCapitalizer:
    def __init__(self, iterable):
        self._iterable = iterable

    async def __aiter__(self):
        async for item in self._iterable:
            yield item.upper()

The __async_for__ modifier expects you to specify an iterable of values to be yielded when async for is used on your fake object.

NOTE

The __async_for__ modifier works directly on fake objects, not method calls. For example, s.__async_for__.alpha.beta() is not supported. If you need something like that, do it in stages:

s.alpha.beta() >> Fake('gamma')
s.__async_for__.gamma >> ['sequence', 'of', 'values']