Writing Tests for Devices#

In this tutorial we will explore how to write tests for ophyd-async Devices that do not require the real hardware. This allows us to catch bugs in our logic by inspecting what it would send to the hardware, and once it is working gives us confidence that it will stay working. Python provides some standard tools like mocking, patching and fixtures, and ophyd-async provides some utility methods to help too.

There are two categories of test that will typically be written for a Device:

  • Tests that call the bluesky verbs (like set() or read()) directly

  • Tests that execute a bluesky plan (like bp.count()) under a RunEngine

The first category are generally for low level tests like checking a motor will pass the correct units up to the progress bar or that it times out if the move is too short. The second category is for higher level tests like checking a detector will produce the correct files when used in a standard plan. Both will be needed at some point, so this tutorial will cover how to write the tests and when to use them.

Tests that call the bluesky verbs directly#

If we need to add a feature to a particular Device, or fix a bug, and it only affects a single verb, then we will probably test the device outside the bluesky RunEngine, calling the verbs directly. This means we need to:

  • Create the Device

  • Set some mock values for the Signals on it

  • Call the verb

  • Inspect the results

  • Possibly do some cleanup

Create a fixture and set signal values#

We will be writing a test using the pytest framework which encourages fixtures to setup and teardown the Devices we wish to test. In this case we will create the DemoMotor from the previous tutorial:

@pytest.fixture
async def mock_motor():
    # Connect with a plain LazyMock, rather than mock=True that will use
    # a InstantMovableMock, so we can have full control of how the readback is set
    # in the tests
    mock_motor = demo.DemoMotor("BLxxI-MO-TABLE-01:X:", name="mock_motor")
    await mock_motor.connect(mock=LazyMock())
    set_mock_units(mock_motor.readback, "mm")
    set_mock_precision(mock_motor.readback, 3)
    set_mock_value(mock_motor.velocity, 1)
    yield mock_motor

This fixture opts out of the automatic mock behaviour by connecting with a plain LazyMock, giving the tests control over when the readback updates mid-move. set_mock_units and set_mock_precision inject units and precision metadata directly on the readback signal, without needing dedicated child signals on the device.

If we had any cleanup to do, we would do that after the yield statement.

Automatic mock behavior injection#

If you find yourself repeatedly using callback_on_mock_put to set up the same mock behavior for a Device type across many tests, you can define a DeviceMock subclass to automatically inject that behavior when the Device is connected in mock mode. This is especially useful for defining standard mock behavior alongside your Device definitions.

For example:

class InstantMotorMock(DeviceMock["Motor"]):
    """Mock behaviour that instantly moves readback to setpoint."""

    async def connect(self, device: Motor) -> None:
        """Mock signals to do an instant move on setpoint write."""
        # Set sensible defaults to avoid runtime errors
        set_mock_value(device.velocity, 1000)  # Prevent ZeroDivisionError
        set_mock_value(device.max_velocity, 1000)  # Prevent ZeroDivisionError

        # Motor starts in "done" state (not moving)
        set_mock_value(device.motor_done_move, 1)

        # When setpoint is written to, immediately update readback and done flag
        def _instant_move(value):
            set_mock_value(device.motor_done_move, 0)  # Moving
            set_mock_value(device.user_readback, value)  # Arrive instantly
            set_mock_value(device.motor_done_move, 1)  # Done

        callback_on_mock_put(device.user_setpoint, _instant_move)

Then decorate the original class with default_mock_class so it is automatically used when connected in mock mode:

@default_mock_class(InstantMotorMock)
class Motor(StandardMovable, StandardReadable, Flyable, Preparable):

Now whenever a Motor is connected using init_devices(mock=True), it will automatically use InstantMotorMock without any fixture setup. You can still override the automatic mock for specific tests by passing an explicit DeviceMock instance or a plain LazyMock directly to connect(), as the mock_motor fixture above does.

pytest-asyncio setup#

Note

Fixtures and tests for async Devices must be async. To enable this, install and configure pytest-asyncio in your project’s pyproject.toml:

[project.optional-dependencies]
dev = [
    "pytest-asyncio",
    # other dependencies
]

[tool.pytest.ini_options]
asyncio_mode = "auto"
# other options

Checking the output of verbs in tests#

Let’s test some verbs. We want to check that we can read() and read_configuration() on a DemoMotor while staged, and that we can still call them when unstaged:

async def test_read_motor(mock_motor: demo.DemoMotor):
    await mock_motor.stage()
    await assert_reading(
        mock_motor,
        {"mock_motor": {"value": 0.0, "timestamp": ANY, "alarm_severity": 0}},
    )
    await assert_configuration(
        mock_motor,
        {
            "mock_motor-velocity": {
                "value": 1.0,
                "timestamp": ANY,
                "alarm_severity": 0,
            },
        },
    )
    # Check that changing the readback value changes the reading
    set_mock_value(mock_motor.readback, 0.5)
    await assert_value(mock_motor.readback, 0.5)
    await assert_reading(
        mock_motor,
        {"mock_motor": {"value": 0.5, "timestamp": ANY, "alarm_severity": 0}},
    )
    # Check we can still read when not staged
    await mock_motor.unstage()
    set_mock_value(mock_motor.readback, 0.1)
    await assert_reading(
        mock_motor,
        {"mock_motor": {"value": 0.1, "timestamp": ANY, "alarm_severity": 0}},
    )

We write an async test method so we can await our calls to verbs. We include the fixture we defined earlier in the function arguments and pytest will automatically create it for us and pass it to the function call. We make use of the assert_reading, assert_value and assert_configuration helpers to check that our motor gives the right output, then use set_mock_value to change the value of the read only Signal before checking the verbs give the right output.

Note

Some of our tests produce timestamps, instead of checking their values we use unittest.mock.ANY to say that the timestamp just has to be present to pass.

Checking that commands and signals were called#

Now let’s call some verbs and check that they do the right thing. We want to check that stop() triggers the TriggerableCommand stop_, waiting for it to complete:

async def test_motor_stopped(mock_motor: demo.DemoMotor):
    # Check it hasn't already been called
    stop_mock = get_mock_execute(mock_motor.stop_)
    stop_mock.assert_not_called()
    # Call stop and check execute() is called with no arguments
    await mock_motor.stop()
    stop_mock.assert_awaited_once_with()
    # We can also track all the mock calls that have happened on the device
    parent_mock = get_mock(mock_motor)
    await mock_motor.velocity.set(15)
    assert parent_mock.mock_calls == [
        call.stop_.execute(),
        call.velocity.put(15),
    ]

This time we use get_mock_execute to get an unittest.mock.AsyncMock that will be called every time stop_.trigger() is called. We check it hasn’t been called, then call our method, then check it has been called with no arguments. We also show that we can call get_mock on the parent to see all of the mock calls that have been made on all its children, useful to check ordering.

For Signals, the equivalent is get_mock_put, which returns an AsyncMock that records every Signal.set() / put() call.

Checking for watcher updates#

Now let’s pretend to be a progress bar and check that we get the right outputs. We want to check that set() will call any progress watchers with appropriate updates, and also terminate when the readback value reaches the correct value:

async def test_motor_moving_well(mock_motor: demo.DemoMotor) -> None:
    # Start it moving
    s = mock_motor.set(0.55)
    # Watch for updates, and make sure the first update is the current position
    watcher = StatusWatcher(s)
    await watcher.wait_for_call(
        name="mock_motor",
        current=0.0,
        initial=0.0,
        target=0.55,
        unit="mm",
        precision=3,
        time_elapsed=pytest.approx(0.0, abs=0.18),
    )
    await wait_for_pending_wakeups()
    await assert_value(mock_motor.setpoint, 0.55)
    assert not s.done
    # Wait a bit and give it an update, checking that the watcher is called with it
    await asyncio.sleep(0.2)
    set_mock_value(mock_motor.readback, 0.1)
    await watcher.wait_for_call(
        name="mock_motor",
        current=0.1,
        initial=0.0,
        target=0.55,
        unit="mm",
        precision=3,
        time_elapsed=pytest.approx(0.2, abs=0.18),
    )
    # Make it almost get there and check that it completes
    set_mock_value(mock_motor.readback, 0.5499999)
    await wait_for_pending_wakeups()
    assert s.done
    assert s.success

Here we call the verb, but don’t wait for it to complete (as that would wait forever). Instead we attach a StatusWatcher to the WatchableAsyncStatus that set() returns, and periodically call set_mock_value on the readback, checking that our watcher was called with the right values. When we give it a value that should make set() terminate, we call wait_for_pending_wakeups to make sure the background tasks get some time to finish correctly before checking the status completed successfully.

Setting side effects on mocks#

By default, a Signal connected in mock mode records all put() calls and stores the put value as the readback. Use callback_on_mock_put to inject side effects — for example, to propagate a setpoint write through to a readback:

with callback_on_mock_put(motor.setpoint, lambda v: set_mock_value(motor.readback, v)):
    await motor.setpoint.set(10.0)
# motor.readback is now 10.0

The callback is cleared automatically when the context exits. For a persistent side effect across a whole test, call it as a plain function (without with).

For a Command backed by soft_command and connected in mock mode, the original Python function is called by default — mock mode behaves identically to real mode unless you intervene. Use get_mock_execute to assert the call was made, or use callback_on_mock_execute to suppress the real function and return something else.

For hardware-backed Commands (e.g. EPICS), there is no underlying Python function to call: mock mode returns a manufactured “empty” default for the declared return type (e.g. 0 for ints, [] for arrays). The same callback_on_mock_execute override applies.

Other test utilities#

There are a few other things we may wish to do in tests:

Tests that execute a bluesky plan#

If we need to check that our Device performs correctly within a plan that calls multiple verbs, it is best to test it under an actual RunEngine. This allows you to check that when the verbs are called in the order that they are in the plan, the correct behavior occurs.

Create a RunEngine in a fixture#

First you need to define a RunEngine that could be used in any test. If you don’t already have one in your project you could define one like this:

@pytest.fixture(scope="function")
def RE():
    RE = RunEngine(call_returns_result=True)
    yield RE
    if RE.state not in ("idle", "panicked"):
        RE.halt()

Run a plan and inspect the documents it produces#

Now you can run a plan, and check that it produces the correct bluesky documents. Let’s go back to the demo and test the DemoPointDetector in a bp.count plan:

@pytest.fixture
async def mock_point_detector():
    async with init_devices(mock=True):
        mock_point_detector = demo.DemoPointDetector("MOCK:DET:")
    yield mock_point_detector

async def test_point_detector_in_plan(
    RE: RunEngine, mock_point_detector: demo.DemoPointDetector
):
    # Subscribe to new documents produce, putting them in a dict by type
    docs = defaultdict(list)
    RE.subscribe(lambda name, doc: docs[name].append(doc))
    # Set the channel values to a known value
    for i, channel in mock_point_detector.channel.items():
        set_mock_value(channel.value, 100 + i)
    # Run the plan and assert the right docs are produced
    RE(bp.count([mock_point_detector], num=2))
    assert_emitted(docs, start=1, descriptor=1, event=2, stop=1)
    assert docs["event"][1]["data"] == {
        "mock_point_detector-channel-1-value": 101,
        "mock_point_detector-channel-2-value": 102,
        "mock_point_detector-channel-3-value": 103,
    }

Here we create a collections.defaultdict and put the RunEngine produced documents in it. Then we use set_mock_value to set the channels of the detector to some known values. Finally we run the plan and use assert_emitted to check the correct numbers of documents have been produced. We can also inspect individual documents for more details.

Conclusion#

In this tutorial we have explored how to write tests for Devices without having the hardware available, by using connection in mock mode.