Testing wampy appsΒΆ

To test any WAMP application you are going to need a Peer acting as a Router.

wampy provides a pytest fixture for this: router which must be installed via $ pip install --editable .[dev]. Usage is then simple.

For example

def test_my_wampy_applications(router):
    # do stuff here

The router is Crossbar.io and will be started and shutdown between each test.

It has a default configuration which you can override in your tests by creating a config_path fixture in your own conftest or test module.

For example

import pytest


@pytest.fixture
def config_path():
    return './path/to/my/preferred/crossbar.json'

Now any test using router will be a Crossbar.io server configured yourself.

For example

def test_my_app(router):
    # this router's configuration has been overridden!

If you require even more control you can import the router itself from wampy.peers.routers and setup your tests however you need to.

wampy also provides a pytest fixture for a WAMP client.

Here is an example testing a wampy HelloService application.

import pytest

from wampy.roles.callee import callee
from wampy.peers.clients import Client
from wampy.testing import wait_for_registrations

class HelloService(Client):

    @callee
    def say_hello(self, name):
        message = "Hello {}".format(name)
        return message


@pytest.yield_fixture
def hello_service(router):
    with HelloService(router=router) as service:
        wait_for_registrations(service, 1)
        yield


def test_get_hello_message(hello_service, router, client):
    response = client.rpc.say_hello(name="wampy")

    assert response == "Hello wampy"

Notice the use of wait_for_registrations. All wampy actions are asynchronous, and so it’s easy to get confused when setting up tests wondering why an application hasn’t registered Callees or subscribed to Topics, or a Session even opened yet.

So to help you setup your tests and avoid race conditions there are some helpers that you can execute to wait for async certain actions to perform before you start actually running test cases. These are:

# execute with the client you're waiting for as the only argument
from wampy.testing import wait_for_session
# e.g. ```wait_for_session(client)```

# wait for a specific number of registrations on a client
from wampy.testing import wait_for_registrations
# e.g. ``wait_for_registrations(client, number_of_registrations=5)

# wait for a specific number of subscriptions on a client
from wampy.testing import wait_for_subscriptions
# e.g. ``wait_for_subscriptions(client, number_of_subscriptions=7)

# provied a function that raises until the test passes
from test.helpers import assert_stops_raising
# e.g. assert_stops_raising(my_func_that_raises_until_condition_met)

For far more examples, see the wampy test suite.