Testing Asphalt components

Testing Asphalt components and component hierarchies is a relatively simple procedure:

Create a Context and enter it with async with ...
Run start_component() with your component class as the first argument (and the configuration dictionary, if you have one, as the second argument)
Run the test code itself

With Asphalt projects, it is recommended to use the pytest testing framework because it is already being used with Asphalt core and it provides easy testing of asynchronous code (via AnyIO’s pytest plugin).

Example

Let’s build a test suite for the Echo Tutorial.

The client and server components could be tested separately, but to make things easier, we’ll test them against each other.

Create a tests directory at the root of the project directory and create a module named test_client_server there (the test_ prefix is important):

# isort: off
from __future__ import annotations

from collections.abc import AsyncGenerator

import pytest
from anyio import wait_all_tasks_blocked
from pytest import CaptureFixture
from asphalt.core import Context, start_component

from echo.client import ClientComponent
from echo.server import ServerComponent

pytestmark = pytest.mark.anyio


@pytest.fixture
async def server(capsys: CaptureFixture[str]) -> AsyncGenerator[None, None]:
    async with Context():
        await start_component(ServerComponent)
        yield


async def test_client_and_server(server: None, capsys: CaptureFixture[str]) -> None:
    async with Context():
        component = await start_component(ClientComponent, {"message": "Hello!"})
        await component.run()

    # Grab the captured output of sys.stdout and sys.stderr from the capsys fixture
    await wait_all_tasks_blocked()
    out, err = capsys.readouterr()
    assert "Message from client: Hello!" in out
    assert "Server responded: Hello!" in out

In the above test module, the first thing you should note is pytestmark = pytest.mark.anyio. This is the pytest marker that marks all coroutine functions in the module to be run via AnyIO’s pytest plugin.

The next item in the module is the server asynchronous generator fixture. Fixtures like these are run by AnyIO’s pytest plugin in their respective tasks, making the practice of straddling a Context on a yield safe. This would normally be bad, as the context contains a TaskGroup which usually should not be used together with yield, unless it’s carefully managed like it is here.

The actual test function, test_client_and_server() first declares a dependency on the server fixture, and then on another fixture (capsys). This other fixture is provided by pytest, and it captures standard output and error, letting us find out what message the components printed. Note that the server fixture also depends on this fixture so that outputs from both the server and client are properly captured.

In this test function, the client component is instantiated and run. Because the client component is a CLIApplicationComponent, we can just run it directly by calling its run() method. While the client component does not contain any child components or other startup logic, we’re nevertheless calling its start() method first, as this is a “best practice”.

Finally, we exit the context and check that the server and the client printed the messages they were supposed to. When the server receives a line from the client, it prints a message to standard output using print(). Likewise, when the client gets a response from the server, it too prints out its own message. By using the capsys fixture, we can capture the output and verify it against the expected lines.

To run the test suite, make sure you’re in the project directory and then do:

PYTHONPATH=. pytest tests

For more elaborate examples, please see the test suites of various Asphalt subprojects.