Testing with CLU

CLU provides several tools to test actors. A typical example is as follows

import click
from clu.parser import command_parser
from clu.testing import setup_test_actor

@command_parser.command()
@click.argument('NAME', type=str)
async def greeter(command, name):
    command.finish(text=f'Hi {name}!')

@pytest.mark.asyncio
async def test_actor():

    test_actor = await setup_test_actor(LegacyActor('my_actor',
                                                    host='localhost',
                                                    port=9999))

    # The following is not needed, start() is replaced with a MagicMock()
    await test_actor.start()

    # Invoke command and wait until it finishes
    command = test_actor.invoke_command('greeter John')
    await command

    # Make sure the command finished successfully
    assert command.status.is_done

    # Get the last reply and check its "text" keyword
    last_reply = test_actor.mock_replies[-1]
    assert last_reply.flag == ':'
    assert last_reply['text'] == '"Hi John!"'

What setup_test_actor does is to replace the start method with a mock so that it’s not necessary to establish a real connection over TCP/IP. It also adds an invoke_command method that can be used to send test commands to the actor. Instead of replying via the normal actor channel, replies are stored in mock_replies as MockReply objects.

If using pytest, a normal design pattern is to define the test actor as a fixture

@pytest.fixture(scope='session')
async def test_actor():

    _actor = setup_test_actor(LegacyActor('my_actor', host='localhost', port=9999))

    yield _actor

    # Clear replies
    _actor.mock_replies.clear()

This usually requires instally pytest-asyncio to be able to define coroutines as fixtures.

setup_test_actor can be used with AMQPActor, JSONActor, and LegacyActor.

API

class clu.testing.MockReply(user_id, command_id, flag, data={})

Bases: dict

Stores a reply written to a transport.

The data of the message is stored as part of the dictionary.

Parameters:

• user_id (int | str | None) – The user ID of the client to which the reply was sent.

• command_id (int | str | None) – The command ID of the command that produced this reply.

• flag (str) – The message type flag.

• data (Dict[str, Any]) – The payload of the message.

class clu.testing.MockReplyList(actor)

Bases: list

Stores replies as MockReply objects.

clear()

Remove all items from list.

parse_reply(reply, routing_key=None)

Parses a reply and construct a MockReply, which is appended.

async clu.testing.setup_test_actor(actor, user_id=666) → T

Setups an actor for testing, mocking the client transport.

Takes an actor and modifies it in two ways: :rtype: TypeVar(T, bound= MockedActor)

• Adds a invoke_mock_command method to it that allows to submit a command string as if it had been received from a transport.

• Mocks a client transport with user_id that is connected to the actor. Messages written to the transport are stored as MockReply in a MockReplyList that is accessible via a new actor.mock_replies attribute.

The actor is modified in place and returned.