Note

Ophyd async is included on a provisional basis until the v1.0 release and may change API on minor release numbers before then

Write Tests for Devices

Testing ophyd-async devices using tools like mocking, patching, and fixtures can become complicated very quickly. The library provides several utilities to make it easier.

Async Tests

pytest-asyncio is required for async tests. It is should be included as a dev dependency of your project. Tests can either be decorated with @pytest.mark.asyncio or the project can be automatically configured to detect async tests.

# pyproject.toml

[tool.pytest.ini_options]
...
asyncio_mode = "auto"

Sim Backend

Ophyd devices initialized with a sim backend behave in a similar way to mocks, without requiring you to mock out all the dependencies and internals. The DeviceCollector can initialize any number of devices, and their signals and sub-devices (recursively), with a sim backend.

@pytest.fixture
async def sim_sensor() -> demo.Sensor:
    async with DeviceCollector(sim=True):
        sim_sensor = demo.Sensor("SIM:SENSOR:")
        # Signals connected here

    assert sim_sensor.name == "sim_sensor"
    return sim_sensor

Sim Utility Functions

Sim signals behave as simply as possible, holding a sensible default value when initialized and retaining any value (in memory) to which they are set. This model breaks down in the case of read-only signals, which cannot be set because there is an expectation of some external device setting them in the real world. There is a utility function, set_sim_value, to mock-set values for sim signals, including read-only ones.

In addition this example also utilizes helper functions like assert_reading and assert_value to ensure the validity of device readings and values. For more information see: API.core

async def test_sensor_reading_shows_value(sim_sensor: demo.Sensor):
    # Check default value
    await assert_value(sim_sensor.value, pytest.approx(0.0))
    assert (await sim_sensor.value.get_value()) == pytest.approx(0.0)
    await assert_reading(
        sim_sensor,
        {
            "sim_sensor-value": {
                "value": 0.0,
                "alarm_severity": 0,
                "timestamp": ANY,
            }
        },
    )
    # Check different value
    set_sim_value(sim_sensor.value, 5.0)
    await assert_reading(
        sim_sensor,
        {
            "sim_sensor-value": {
                "value": 5.0,
                "timestamp": ANY,
                "alarm_severity": 0,
            }
        },
    )

There is another utility function, set_sim_callback, for hooking in logic when a sim value changes (e.g. because someone puts to it).

async def test_mover_stopped(sim_mover: demo.Mover):
    callbacks = []
    set_sim_callback(sim_mover.stop_, lambda r, v: callbacks.append(v))

    assert callbacks == [None]
    await sim_mover.stop()
    assert callbacks == [None, None]

Testing a Device in a Plan with the RunEngine

async def test_sensor_in_plan(RE: RunEngine, sim_sensor: demo.Sensor):
    """Tests sim sensor behavior within a RunEngine plan.

    This test verifies that the sensor emits the expected documents
     when used in plan(count).
    """
    docs = defaultdict(list)

    def capture_emitted(name, doc):
        docs[name].append(doc)

    RE(bp.count([sim_sensor], num=2), capture_emitted)
    assert_emitted(docs, start=1, descriptor=1, event=2, stop=1)

This test verifies that the sim_sensor behaves as expected within a plan. The plan we use here is a count, which takes a specified number of readings from the sim_sensor. Since we set the repeat to two in this test, the sensor should emit two “event” documents along with “start”, “stop” and “descriptor” documents. Finally, we use the helper function assert_emitted to confirm that the emitted documents match our expectations.