How to test a SharingObserver

Components that inherit from SharingObserver rely on a signal bus to communicate. However, because the concrete signal bus implementation is private to ska-tango-base, testing these components in isolation requires a specific harness to construct the signal bus.

The ska-tango-base package provides a set of testing utilities in ska_tango_base.software_bus.testing to provide a temporary bus and a convenient callback mock class to verify emissions on the bus. This module is only imported on demand and requires ska-tango-testing to be available for some features.

This guide describes how to use these utilities to write unit tests for sharing observers.

Using the testing bus context

The bus_test_context() context manager provides a temporary signal bus and injects it into the observers you are testing.

When started, it performs the following steps:

1. Creates a new internal signal bus instance.

2. Registers all provided observers with the bus. If an observer inherits from SharingObserver, its shared_bus attribute is set as if it was part of a Tango device.

3. Starts processing emissions

As a convenience, ska-tango-base also provides MockSignalObserver to easily verify that a sharing observer emits signal values as expected. The MockSignalObserver is a ska_tango_testing.mock.MockCallableGroup that needs to be initialised with all the absolute signal names that you wish to emit on the bus. You can then use the standard assertion methods to check that these values have been emitted.

When the bus_test_context() context manager exits it will ensure that the signal bus is cleaned up properly.

Example:

import ska_tango_base as stb
import ska_tango_base.software_bus.testing as stbst

class MySharingObserver(stb.software_bus.SharingObserver):
    value = stb.software_bus.Signal[int]()

    def set_value(self, val: int) -> None:
        self.value = val

def test_my_sharing_observer():
    comp = MySharingObserver()
    # We want to monitor the signal ".value" emitted by the component
    mock = stbst.MockSignalObserver(".value")

    with stbst.bus_test_context(comp, mock):
        comp.set_value(100)
        mock.assert_call(".value", 100)

        comp.set_value(200)
        mock.assert_call(".value", 200)