Testing

Opsdroid contains tooling for testing which is useful for testing opsdroid itself but also for testing your skills and any Connectors, Parsers and Databases created outside the core project.

Opsdroid tests are run using pytest. Fixtures can be imported into your own projects, and are available by default in opsdroid core tests.

There are also some utilities for mocking our and running tests specific to opsdroid.

Fixtures

opsdroid.testing.opsdroid()

Fixture with a plain instance of opsdroid.

Will yield an instance of opsdroid.core.OpsDroid which hasn’t been loaded.

Return type

OpsDroid

opsdroid.testing.mock_api(mock_api_obj)

Fixture for mocking API calls to a web service.

Will yield a running instance of opsdroid.testing.ExternalAPIMockServer, which has been configured with any routes specified through @pytest.mark.add_response() decorators, or modification of the mock_api_obj() fixture before the test is called.

All arguments and keyword arguments passed to pytest.mark.add_response are passed through to .ExternalAPIMockServer.add_response.

An example test would look like:

@pytest.mark.add_response("/test", "GET")
@pytest.mark.add_response("/test2", "GET", status=500)
@pytest.mark.asyncio
async def test_hello(mock_api):
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{mock_api.base_url}/test") as resp:
            assert resp.status == 200
            assert mock_api.called("/test")

        async with session.get(f"{mock_api.base_url}/test2") as resp:
            assert resp.status == 500
            assert mock_api.called("/test2")

Return type

ExternalAPIMockServer

Utilities

class opsdroid.testing.ExternalAPIMockServer

A webserver which can pretend to be an external API.

The general idea with this class is to allow you to push expected responses onto a stack for each API call you expect your test to make. Then as your tests make those calls each response is popped from the stack.

You can then assert that routes were called and that data and headers were sent correctly.

Your test will need to switch the URL of the API calls, so the thing you are testing should be configurable at runtime. You will also need to capture the responses from the real API your are mocking and store them as JSON files. Then you can push those responses onto the stack at the start of your test.

Examples

A simple example of pushing a response onto a stack and making a request:

import pytest
import aiohttp

from opsdroid.testing import ExternalAPIMockServer

@pytest.mark.asyncio
async def test_example():
    # Construct the mock API server and push on a test method
    mock_api = ExternalAPIMockServer()
    mock_api.add_response("/test", "GET", None, 200)

    # Create a closure function. We will have our mock_api run this concurrently with the
    # web server later.
    async with mock_api.running():
        # Make an HTTP request to our mock_api
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{mock_api.base_url}/test") as resp:

                # Assert that it gives the expected responses
                assert resp.status == 200
                assert mock_api.called("/test")

add_response(route, method, response=None, status=200)

Push a mocked response onto a route.

Return type

None

property base_url: str

Return the base url of the web server.

call_count(route, method=None)

Route has been called n times.

Parameters

route (str) – The API route that we want to know if was called.

Return type

int

Returns

The number of times it was called.

called(route, method=None)

Route has been called.

Parameters

route (str) – The API route that we want to know if was called.

Return type

bool

Returns

Wether or not it was called.

get_payload(route, idx=0)

Return data payload that the route was called with.

Parameters

• route (str) – The API route that we want to get the payload for.

• idx (int) – The index of the call. Useful if it was called multiple times and we want something other than the first one.

Return type

Dict

Returns

The data payload which was sent in the POST request.

get_request(route, method, idx=0)

Route has been called n times.

Parameters

• route (str) – The API route that we want to get the request for.

• idx (int) – The index of the call. Useful if it was called multiple times and we want something other than the first one.

Return type

Request

Returns

The request that was made.

reset()

Reset the mock back to a clean state.

Return type

None

running()

Start the External API server within a context manager.

Return type

ExternalAPIMockServer

async start()

Start the server.

Return type

None

async stop()

Stop the web server.

Return type

None

async opsdroid.testing.run_unit_test(opsdroid, test, *args, start_timeout=1, **kwargs)

Run a unit test function against opsdroid.

This method should be used when testing on a loaded but stopped instance of opsdroid. The instance will be started concurrently with the test runner. The test runner will block until opsdroid is ready and then the test will be called. Once the test has returned opsdroid will be stopped and unloaded.

Parameters

• opsdroid (OpsDroid) – A loaded but stopped instance of opsdroid.

• test (Awaitable) – A test to execute concurrently with opsdroid once it has been started.

• start_timeout – Wait up to this timeout for opsdroid to say that it is running.

Return type

Any

Returns

Passes on the return of the test coroutine.

Examples

An example of running a coroutine test against opsdroid:

import pytest
from opsdroid.testing import (
    opsdroid,
    run_unit_test,
    MINIMAL_CONFIG
    )

@pytest.mark.asyncio
async def test_example(opsdroid):
    # Using the opsdrid fixture we load it with the
    # minimal example config
    await opsdroid.load(config=MINIMAL_CONFIG)

    # Check that opsdroid is not currently running
    assert not opsdroid.is_running()

    # Define an awaitable closure which asserts that
    # opsdroid is now running
    async def test():
        assert opsdroid.is_running()
        return True  # So that we can assert our run test

    # Run our closure against opsdroid. This will start opsdroid,
    # await our closure and then stop opsdroid again.
    assert await run_unit_test(opsdroid, test)

async opsdroid.testing.call_endpoint(opsdroid, endpoint, method='GET', data_path=None, data=None, headers=None, **kwargs)

Call an opsdroid API endpoint with the provided data.

This method should be used when testing on a running instance of opsdroid. The endpoint will be appended to the base url of the running opsdroid, so you do not need to know the address of the running opsdroid. An HTTP request will be made with the provided method and data or data_path for methods that support it.

For methods like "POST" either data or data_path should be set.

Parameters

• opsdroid (OpsDroid) – A running instance of opsdroid.

• endpoint (str) – The API route to call.

• method (str) – The HTTP method to use when calling.

• data_path (Optional[str]) – A local file path to load a JSON payload from to be sent in supported methods.

• data (Union[Dict, str, None]) – A dictionary payload to be sent in supported methods.

• headers (Optional[Dict]) – A dictionary of headers to be sent in supported methods.

Return type

Response

Returns

The response from the HTTP request.

Examples

Call the /stats endpoint of opsdroid without having to know what address opsdroid is serving at:

import pytest
from opsdroid.testing import (
    opsdroid,
    call_endpoint,
    run_unit_test,
    MINIMAL_CONFIG
    )

@pytest.mark.asyncio
async def test_example(opsdroid):
    # Using the opsdrid fixture we load it with the
    # minimal example config
    await opsdroid.load(config=MINIMAL_CONFIG)

    async def test():
        # Call our endpoint by just passing
        # opsdroid, the endpoint and the method
        resp = await call_endpoint(opsdroid, "/stats", "GET")

        # Make assertions that opsdroid responded successfully
        assert resp.status == 200
        return True

    assert await run_unit_test(opsdroid, test)