Testing¶

Basics ¶

To test an agent when unit testing or functional testing, use the special Agent.test() mode to send items to the stream while processing it locally:

app = faust.App('test-example')

class Order(faust.Record, serializer='json'):
    account_id: str
    product_id: str
    amount: int
    price: float

orders_topic = app.topic('orders', value_type=Order)
orders_for_account = app.Table('order-count-by-account', default=int)

@app.agent(orders_topic)
async def order(orders):
    async for order in orders.group_by(Order.account_id):
        orders_for_account[order.account_id] += 1
        yield order

Our agent reads a stream of orders and keeps a count of them by account id in a distributed table also partitioned by the account id.

To test this agent we use order.test_context():

async def test_order():
    # start and stop the agent in this block
    async with order.test_context() as agent:
        order = Order(account_id='1', product_id='2', amount=1, price=300)
        # sent order to the test agents local channel, and wait
        # the agent to process it.
        await agent.put(order)
        # at this point the agent already updated the table
        assert orders_for_account[order.account_id] == 1
        await agent.put(order)
        assert orders_for_account[order.account_id] == 2

async def run_tests():
    app.conf.store = 'memory://'   # tables must be in-memory
    await test_order()

if __name__ == '__main__':
    import asyncio
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run_tests())

For the rest of this guide we’ll be using https://pypi.org/project/pytest/ and https://pypi.org/project/pytest-asyncio/ for our examples. If you’re using a different testing framework you may have to adapt them a bit to work.

Testing with https://pypi.org/project/pytest/¶

Testing that an agent sends to topic/calls another agent.¶

When unit testing you should mock any dependencies of the agent being tested,

If your agent calls another function: mock that function to verify it was called.
If your agent sends a message to a topic: mock that topic to verify a message was sent.
If your agent calls another agent: mock the other agent to verify it was called.

Here’s an example agent that calls another agent:

import faust

app = faust.App('example-test-agent-call')

@app.agent()
async def foo(stream):
    async for value in stream:
        await bar.send(value)
        yield value

@app.agent()
async def bar(stream):
    async for value in stream:
        yield value + 'YOLO'

To test these two agents you have to test them in isolation of each other: first test foo with bar mocked, then in a different test do bar:

import pytest
from unittest.mock import Mock, patch

from example import app, foo, bar

@pytest.fixture(scope="function")
def test_app(event_loop):
    """passing in event_loop helps avoid 'attached to a different loop' error"""
    app.loop = event_loop
    app.finalize()
    app.conf.store = 'memory://'
    app.flow_control.resume()
    return app

@pytest.mark.asyncio()
async def test_foo(test_app):
    with patch(__name__ + '.bar') as mocked_bar:
            mocked_bar.send = mock_coro()
        async with foo.test_context() as agent:
            await agent.put('hey')
            mocked_bar.send.assert_called_with('hey')

def mock_coro(return_value=None, **kwargs):
    """Create mock coroutine function."""
    async def wrapped(*args, **kwargs):
        return return_value
    return Mock(wraps=wrapped, **kwargs)

@pytest.mark.asyncio()
async def test_bar(test_app):
    async with bar.test_context() as agent:
        event = await agent.put('hey')
        assert agent.results[event.message.offset] == 'heyYOLO'

You can put the test_app fixture into a [conftest.py file](https://docs.pytest.org/en/6.2.x/fixture.html#scope-sharing-fixtures-across-classes-modules-packages-or-session). If the fixture is not in the same file as the app’s definition (which should be the case) you must import the app the fixture definition:

Note

The https://pypi.org/project/pytest-asyncio/ extension must be installed to run these tests. If you don’t have it use pip to install it:

$ pip install -U pytest-asyncio

Setting up a test suite ¶

A faust app is normally a module-level singleton whose agents are attached to it at import time, so a suite needs a little setup before those agents can be started in isolation.

The four files below are a complete, working suite: copy them into an empty directory, pip install faust-streaming pytest pytest-asyncio, run pytest, and four tests pass. The rest of this section explains what each piece is doing.

The application under test – `myapp.py`¶

The agent here is a pure consumer: it updates a table, calls a service and forwards to another topic, and never yields. See Testing agents that don’t yield for why that shape needs no sink.

import faust


class Order(faust.Record, serializer='json'):
    account_id: str
    amount: int


app = faust.App('orders-app', broker='kafka://localhost:9092')

orders_topic = app.topic('orders', value_type=Order)
shipped_topic = app.topic('shipped', value_type=Order)

orders_for_account = app.Table('order-count-by-account', default=int)


async def notify_warehouse(order: Order) -> None:
    """Pretend this calls an external HTTP service."""
    ...


@app.agent(orders_topic)
async def track_order(orders):
    # NOTE: no ``yield`` -- this agent is a pure consumer.
    async for order in orders:
        orders_for_account[order.account_id] += 1
        await notify_warehouse(order)
        await shipped_topic.send(value=order)

Enabling async tests – `pytest.ini`¶

These examples use https://pypi.org/project/pytest-asyncio/ in auto mode, so async tests and fixtures need no decorator:

[pytest]
asyncio_mode = auto

Fixtures – `conftest.py`¶

import asyncio
from unittest.mock import AsyncMock

import pytest

import myapp
from myapp import app as _app


@pytest.fixture()
async def app():
    # Bind the app to the event loop pytest-asyncio created for this test.
    _app.loop = asyncio.get_running_loop()
    _app.finalize()
    _app.conf.store = 'memory://'
    _app.flow_control.resume()
    return _app


@pytest.fixture()
def shipped(monkeypatch):
    """Keep the agent from reaching a real broker."""
    send = AsyncMock()
    monkeypatch.setattr(myapp.shipped_topic, 'send', send)
    return send


@pytest.fixture()
def warehouse(monkeypatch):
    """Keep the agent from calling the real service."""
    notify = AsyncMock()
    monkeypatch.setattr(myapp, 'notify_warehouse', notify)
    return notify

The tests – `test_myapp.py`¶

from myapp import Order, orders_for_account, track_order


async def test_counts_orders_per_account(app, shipped, warehouse):
    async with track_order.test_context() as agent:
        await agent.put(Order(account_id='A', amount=1))
        assert orders_for_account['A'] == 1

        await agent.put(Order(account_id='A', amount=2))
        assert orders_for_account['A'] == 2


async def test_notifies_warehouse(app, shipped, warehouse):
    async with track_order.test_context() as agent:
        order = Order(account_id='B', amount=3)
        await agent.put(order)
        warehouse.assert_awaited_once_with(order)


async def test_forwards_to_shipped_topic(app, shipped, warehouse):
    async with track_order.test_context() as agent:
        order = Order(account_id='C', amount=4)
        await agent.put(order)
        shipped.assert_awaited_once_with(value=order)


async def test_results_records_input_values(app, shipped, warehouse):
    async with track_order.test_context() as agent:
        order = Order(account_id='D', amount=5)
        await agent.put(order)
        # For a sink-less agent ``results`` holds what went IN.
        assert agent.results[0] == order

What the `app` fixture is doing ¶

Each line earns its place:

_app.loop: Re-binds the app to the event loop https://pypi.org/project/pytest-asyncio/ created for this test. A module-level app otherwise stays bound to the loop of whichever test ran first, and every later test fails with RuntimeError: Event loop is closed.
_app.finalize(): Completes app configuration. Normally the worker does this for you.
_app.conf.store = 'memory://': Tables must be in-memory in tests; the default (RocksDB) wants a real data directory. Faust warns that the setting arrives after your topics and agents were declared – expected here and harmless, since no table has started yet. Set store on the App itself if you would rather not see the warning.
_app.flow_control.resume(): Stream queues start out suspended. Omit this and the agent never receives anything, so put() hangs forever.

Note

The test_app fixture shown earlier does the same job, but reaches the loop by requesting https://pypi.org/project/pytest-asyncio/’s event_loop fixture. Recent releases deprecate that (“Asynchronous fixtures and test functions should use asyncio.get_running_loop() instead”), so prefer the async fixture above in new test suites.

Mock anything that leaves the process ¶

test_context() feeds the agent through a local channel, but it does not stub out the rest of your app. If the agent sends to a topic or calls an external service it will genuinely try to, and the test fails with:

aiokafka.errors.KafkaConnectionError: Unable to bootstrap from [('localhost', 9092, ...)]

That is what the shipped and warehouse fixtures are for: give every outbound dependency its own fixture, so each test both stays offline and gets a mock it can assert on.

Testing agents that don’t yield ¶

Every earlier example ends in yield and reads its output back through agent.results. Plenty of real agents never yield: they update a table, call a service, or forward to another topic and stop there. Such an agent cannot use sinks at all – attaching one raises ImproperlyConfigured('Agent must yield to use sinks') – yet test_context() tests them perfectly well. track_order above is exactly that shape.

Because nothing is yielded, you assert on what the agent did rather than on what it returned: the table it wrote (test_counts_orders_per_account), the service it called (test_notifies_warehouse), and the message it forwarded (test_forwards_to_shipped_topic).

What `agent.results` holds ¶

results is populated for both kinds of agent, but it does not mean the same thing in each case:

Agent	`agent.results[offset]` contains
yields	the value the agent yielded (its output)
does not yield	the value that was sent in (its input)

Which row applies is decided from the running agent, not from how its function is written. An agent implemented as a callable object with an async-generator __call__, or as a plain function returning an async generator, yields just as much as an async def that yields directly, and its results hold its output.

There is no output to capture for a sink-less agent, so faust records the incoming value instead. That still makes results useful for confirming which values reached the agent – as test_results_records_input_values does above – but do not read it expecting a return value.

When put() returns

await agent.put(value) waits for the value to be picked up by the agent, and for an agent body that does not await anything mid-loop the side effects are already visible when put() returns – which is why the assertions above can follow it directly.

If the body does await something (an HTTP call, asyncio.sleep, a real send), put() can return before the body has finished with that value, and asserting immediately will be flaky. Wait for the effect rather than assuming it – for example by leaving the async with block, since stopping the agent drains what is in flight:

async def test_slow_agent(app):
    processed = []
    async with slow_agent.test_context() as agent:
        await agent.put('a')
    # the context manager stopped the agent: work is finished
    assert processed == ['a']

Testing and windowed tables ¶

If your table is windowed and you want to verify that the value for a key is correctly set, use table[k].current(event) to get the value placed within the window of the current event:

import faust
import pytest

@pytest.mark.asyncio()
async def test_process_order():
    app.conf.store = 'memory://'
    async with process_order.test_context() as agent:
        order = Order(account_id='1', product_id='2', amount=1, price=300)
        event = await agent.put(order)

        # windowed table: we select window relative to the current event
        assert orders_for_account['1'].current(event) == 1

        # in the window 3 hours ago there were no orders:
        assert orders_for_account['1'].delta(3600 * 3, event)


class Order(faust.Record, serializer='json'):
    account_id: str
    product_id: str
    amount: int
    price: float

app = faust.App('test-example')
orders_topic = app.topic('orders', value_type=Order)

# order count within the last hour (window is a 1-hour TumblingWindow).
orders_for_account = app.Table(
    'order-count-by-account', default=int,
).tumbling(3600).relative_to_stream()

@app.agent(orders_topic)
async def process_order(orders):
    async for order in orders.group_by(Order.account_id):
        orders_for_account[order.account_id] += 1
        yield order