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():
    async with init_devices(mock=True):
        mock_motor = demo.DemoMotor("BLxxI-MO-TABLE-01:X:")
    set_mock_value(mock_motor.units, "mm")
    set_mock_value(mock_motor.precision, 3)
    set_mock_value(mock_motor.velocity, 1)
    yield mock_motor

This will use init_devices to call Device.connect with mock=True. This will recursively replace the real connection to hardware with a mock that allows us to change the Signal’s value that our code will see and capture any attempts to set the value from our code. In this case we know our tests will expect units="mm" and precision=3, so we use set_mock_value to those here.

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

Note

This is an async fixture, and we will be using async tests, so we need to install and configure pytest-asyncio in our projects’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-units": {
                "value": "mm",
                "timestamp": ANY,
                "alarm_severity": 0,
            },
            "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 signals were changed#

Now let’s call some verbs and check that they do the right thing. We want to check that stop() triggers the SignalX 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_put(mock_motor.stop_)
    stop_mock.assert_not_called()
    # Call stop and check it's called with the default value
    await mock_motor.stop()
    stop_mock.assert_called_once_with(None, wait=True)
    # We can also track all the mock puts 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_.put(None, wait=True),
        call.velocity.put(15, wait=True),
    ]

This time we use get_mock_put to get a unittest.mock.Mock 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 None (what a SignalX sends to tell the backend to put the value needed to trigger). 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 it’s children, useful to check ordering.

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.08),
    )
    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.1)
    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.1, abs=0.08),
    )
    # 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.

Other test utilities#

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

  • set_mock_values if you want to set a series of mock values, with repeated checks at each value

  • callback_on_mock_put to allow setting a Signal to have side effects, like setting another Signal

  • set_mock_put_proceeds to block or unblock Signal.set(..., wait=True) from completing

  • mock_puts_blocked a context manager that blocks put proceeds at the start, and unblocks at the end

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.