199 test helpers for assering value reading and configuration

docs/how-to/write-tests-for-devices.rst

@@ -35,6 +35,8 @@ 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: :doc:`API.core<../generated/ophyd_async.core>`
+
 .. literalinclude:: ../../tests/epics/demo/test_demo.py
    :pyobject: test_sensor_reading_shows_value
 
@@ -43,3 +45,12 @@ There is another utility function, ``set_sim_callback``, for hooking in logic wh
 
 .. literalinclude:: ../../tests/epics/demo/test_demo.py
    :pyobject: test_mover_stopped
+
+
+Testing device in a plan with RunEngine
+---------------------------------------
+.. literalinclude:: ../../tests/epics/demo/test_demo.py
+   :pyobject: test_sensor_in_plan
+
+
+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.

src/ophyd_async/core/__init__.py

@@ -30,6 +30,10 @@
     SignalRW,
     SignalW,
     SignalX,
+    assert_configuration,
+    assert_emitted,
+    assert_reading,
+    assert_value,
     observe_value,
     set_and_wait_for_value,
     set_sim_callback,
@@ -103,4 +107,8 @@
     "walk_rw_signals",
     "load_device",
     "save_device",
+    "assert_reading",
+    "assert_value",
+    "assert_configuration",
+    "assert_emitted",
 ]

src/ophyd_async/core/signal.py

@@ -2,7 +2,19 @@
 
 import asyncio
 import functools
-from typing import AsyncGenerator, Callable, Dict, Generic, Optional, Tuple, Type, Union
+from typing import (
+    Any,
+    AsyncGenerator,
+    Awaitable,
+    Callable,
+    Dict,
+    Generic,
+    Mapping,
+    Optional,
+    Tuple,
+    Type,
+    Union,
+)
 
 from bluesky.protocols import (
     Descriptor,
@@ -13,8 +25,9 @@
     Stageable,
     Subscribable,
 )
+from event_model.documents import DocumentType
 
-from ophyd_async.protocols import AsyncReadable
+from ophyd_async.protocols import AsyncConfigurable, AsyncReadable
 
 from .async_status import AsyncStatus
 from .device import Device
@@ -272,6 +285,119 @@ def soft_signal_r_and_backend(
     return (signal, backend)
 
 
+async def _verify_readings(
+    func: Callable[[], Mapping[str, Reading] | Awaitable[Mapping[str, Reading]]],
+    expectation: Mapping[str, Reading],
+) -> None:
+    """Take a read/read_configuration function that return a dictionary and
+        compare it with the expected result (expectation)
+
+    Parameters
+    ----------
+    func:
+        read/read_configuration function.
+
+    expectation:
+        The expected value from the readable/Configurable.
+
+    Notes
+    -----
+    Example usage::
+        await _verify_readings(readable.read, reading)
+    Or::
+        await _verify_readings(configurable.read_configuration, configuration)
+
+    """
+    result = await func()
+
+    assert result == expectation
+
+
+async def assert_value(signal: SignalR[T], value: Any) -> None:
+    """Assert a signal's value and compare it an expected signal.
+
+    Parameters
+    ----------
+    signal:
+        signal with get_value.
+    value:
+        The expected value from the signal.
+
+    Notes
+    -----
+    Example usage::
+        await assert_value(signal, value)
+
+    """
+    assert await signal.get_value() == value
+
+
+async def assert_reading(
+    readable: AsyncReadable, reading: Mapping[str, Reading]
+) -> None:
+    """Assert readings from readable.
+
+    Parameters
+    ----------
+    readable:
+        Callable with readable.read function that generate readings.
+
+    reading:
+        The expected readings from the readable.
+
+    Notes
+    -----
+    Example usage::
+        await assert_reading(readable, reading)
+
+    """
+    await _verify_readings(readable.read, reading)
+
+
+async def assert_configuration(
+    configurable: AsyncConfigurable,
+    configuration: Mapping[str, Reading],
+) -> None:
+    """Assert readings from Configurable.
+
+    Parameters
+    ----------
+    configurable:
+        Configurable with Configurable.read function that generate readings.
+
+    configuration:
+        The expected readings from configurable.
+
+    Notes
+    -----
+    Example usage::
+        await assert_configuration(configurable configuration)
+
+    """
+    await _verify_readings(configurable.read_configuration, configuration)
+
+
+def assert_emitted(docs: Mapping[str, list[DocumentType]], **numbers: int):
+    """Assert emitted document generated by running a Bluesky plan
+
+    Parameters
+    ----------
+    Doc:
+        A dictionary
+
+    numbers:
+        expected emission in kwarg from
+
+    Notes
+    -----
+    Example usage::
+        assert_emitted(docs, start=1, descriptor=1,
+        resource=1, datum=1, event=1, stop=1)
+    """
+    assert list(docs) == list(numbers)
+    assert {name: len(d) for name, d in docs.items()} == numbers
+
+
 async def observe_value(signal: SignalR[T], timeout=None) -> AsyncGenerator[T, None]:
     """Subscribe to the value of a signal so it can be iterated from.
 

tests/core/test_signal.py

@@ -1,15 +1,22 @@
 import asyncio
 import re
 import time
+from unittest.mock import ANY
 
 import numpy
 import pytest
+from bluesky.protocols import Reading
 
 from ophyd_async.core import (
+    DeviceCollector,
     Signal,
     SignalR,
     SignalRW,
     SimSignalBackend,
+    StandardReadable,
+    assert_configuration,
+    assert_reading,
+    assert_value,
     set_and_wait_for_value,
     set_sim_put_proceeds,
     set_sim_value,
@@ -18,6 +25,7 @@
     wait_for_value,
 )
 from ophyd_async.core.utils import DEFAULT_TIMEOUT
+from ophyd_async.epics.signal import epics_signal_r, epics_signal_rw
 
 
 class MySignal(Signal):
@@ -153,3 +161,70 @@ async def test_soft_signal_numpy():
     await int_signal.connect()
     assert (await float_signal.describe())["float_signal"]["dtype"] == "number"
     assert (await int_signal.describe())["int_signal"]["dtype"] == "integer"
+
+
+@pytest.fixture
+async def sim_signal():
+    sim_signal = SignalRW(SimSignalBackend(int, "test"))
+    sim_signal.set_name("sim_signal")
+    await sim_signal.connect(sim=True)
+    yield sim_signal
+
+
+async def test_assert_value(sim_signal: SignalRW):
+    set_sim_value(sim_signal, 168)
+    await assert_value(sim_signal, 168)
+
+
+async def test_assert_reaading(sim_signal: SignalRW):
+    set_sim_value(sim_signal, 888)
+    dummy_reading = {
+        "sim_signal": Reading({"alarm_severity": 0, "timestamp": ANY, "value": 888})
+    }
+    await assert_reading(sim_signal, dummy_reading)
+
+
+class DummyReadable(StandardReadable):
+    """A demo Readable to produce read and config signal"""
+
+    def __init__(self, prefix: str, name="") -> None:
+        # Define some signals
+        self.value = epics_signal_r(float, prefix + "Value")
+        self.mode = epics_signal_rw(str, prefix + "Mode")
+        self.mode2 = epics_signal_rw(str, prefix + "Mode2")
+        # Set name and signals for read() and read_configuration()
+        self.set_readable_signals(
+            read=[self.value],
+            config=[self.mode, self.mode2],
+        )
+        super().__init__(name=name)
+
+
+@pytest.fixture
+async def sim_readable():
+    async with DeviceCollector(sim=True):
+        sim_readable = DummyReadable("SIM:READABLE:")
+        # Signals connected here
+    assert sim_readable.name == "sim_readable"
+    yield sim_readable
+
+
+async def test_assert_configuration(sim_readable: DummyReadable):
+    set_sim_value(sim_readable.value, 123)
+    set_sim_value(sim_readable.mode, "super mode")
+    set_sim_value(sim_readable.mode2, "slow mode")
+    dummy_config_reading = {
+        "sim_readable-mode": (
+            {
+                "alarm_severity": 0,
+                "timestamp": ANY,
+                "value": "super mode",
+            }
+        ),
+        "sim_readable-mode2": {
+            "alarm_severity": 0,
+            "timestamp": ANY,
+            "value": "slow mode",
+        },
+    }
+    await assert_configuration(sim_readable, dummy_config_reading)