199 test helpers for assering value reading and configuration

docs/conf.py

@@ -118,7 +118,7 @@
 # docs in the python documentation.
 intersphinx_mapping = {
     "python": ("https://docs.python.org/3", None),
-    "bluesky": ("https://blueskyproject.io/bluesky/", None),
+    "bluesky": ("https://blueskyproject.io/bluesky/main", None),
     "numpy": ("https://numpy.org/devdocs/", None),
     "databroker": ("https://blueskyproject.io/databroker/", None),
     "event-model": ("https://blueskyproject.io/event-model/main", None),

docs/user/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
 

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,
@@ -99,4 +103,8 @@
     "walk_rw_signals",
     "load_device",
     "save_device",
+    "assert_reading",
+    "assert_value",
+    "assert_configuration",
+    "assert_emitted",
 ]

src/ophyd_async/core/signal.py

@@ -2,9 +2,20 @@
 
 import asyncio
 import functools
-from typing import AsyncGenerator, Callable, Dict, Generic, Optional, Union
+from typing import (
+    Any,
+    AsyncGenerator,
+    Awaitable,
+    Callable,
+    Dict,
+    Generic,
+    Mapping,
+    Optional,
+    Union,
+)
 
 from bluesky.protocols import (
+    Configurable,
     Descriptor,
     Locatable,
     Location,
@@ -14,6 +25,7 @@
     Stageable,
     Subscribable,
 )
+from event_model.documents import DocumentType
 
 from .async_status import AsyncStatus
 from .device import Device
@@ -253,6 +265,123 @@ def set_sim_callback(signal: Signal[T], callback: ReadingValueCallback[T]) -> No
     return _sim_backends[signal].set_callback(callback)
 
 
+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)
+    Future::
+        func will only be Waitable in the future and the if loop can be removed:
+
+        result = func()
+
+    """
+    if asyncio.iscoroutinefunction(func):
+        result = await func()
+    else:
+        result = 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: Readable, 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: Configurable,
+    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,19 +1,28 @@
 import asyncio
 import re
 import time
+from typing import Mapping
+from unittest.mock import ANY
 
 import pytest
+from bluesky.protocols import Reading
 
 from ophyd_async.core import (
+    DeviceCollector,
     Signal,
     SignalRW,
     SimSignalBackend,
+    StandardReadable,
+    assert_configuration,
+    assert_reading,
+    assert_value,
     set_and_wait_for_value,
     set_sim_put_proceeds,
     set_sim_value,
     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):
@@ -119,3 +128,80 @@ async def test_set_and_wait_for_value():
     assert not st.done
     set_sim_put_proceeds(sim_signal, True)
     assert await time_taken_by(st) < 0.1
+
+
+@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: Mapping[str, 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)
+    # test for none awaitable part of verify
+    from ophyd.sim import DetWithConf
+
+    something = DetWithConf(name="det")
+    dummy_config_reading1: Mapping[str, Reading] = {
+        "det_c": {"value": 3, "timestamp": ANY},
+        "det_d": {"value": 4, "timestamp": ANY},
+    }
+
+    await assert_configuration(something, dummy_config_reading1)