Update core and tango tests to match structure of epics tests

src/ophyd_async/testing/__init__.py

@@ -2,6 +2,7 @@
 from ._assert import (
     ApproxTable,
     MonitorQueue,
+    approx_value,
     assert_configuration,
     assert_describe_signal,
     assert_emitted,
@@ -27,6 +28,7 @@
 from ._wait_for_pending import wait_for_pending_wakeups
 
 __all__ = [
+    "approx_value",
     "assert_configuration",
     "assert_describe_signal",
     "assert_emitted",

src/ophyd_async/testing/_assert.py

@@ -1,6 +1,5 @@
 import asyncio
 import time
-from collections.abc import Mapping
 from contextlib import AbstractContextManager
 from typing import Any
 
@@ -17,15 +16,8 @@
 )
 
 
-def _generate_assert_error_msg(name: str, expected_result, actual_result) -> str:
-    WARNING = "\033[93m"
-    FAIL = "\033[91m"
-    ENDC = "\033[0m"
-    return (
-        f"Expected {WARNING}{name}{ENDC} to produce"
-        + f"\n{FAIL}{expected_result}{ENDC}"
-        + f"\nbut actually got \n{FAIL}{actual_result}{ENDC}"
-    )
+def approx_value(value: Any):
+    return ApproxTable(value) if isinstance(value, Table) else pytest.approx(value)
 
 
 async def assert_value(signal: SignalR[SignalDatatypeT], value: Any) -> None:
@@ -45,22 +37,11 @@ async def assert_value(signal: SignalR[SignalDatatypeT], value: Any) -> None:
 
     """
     actual_value = await signal.get_value()
-    assert actual_value == value, _generate_assert_error_msg(
-        name=signal.name,
-        expected_result=value,
-        actual_result=actual_value,
-    )
-
-
-def _approx_readable_value(reading: Mapping[str, Reading]) -> Mapping[str, Reading]:
-    """Change Reading value to pytest.approx(value)"""
-    for i in reading:
-        reading[i]["value"] = pytest.approx(reading[i]["value"])
-    return reading
+    assert approx_value(value) == actual_value
 
 
 async def assert_reading(
-    readable: AsyncReadable, expected_reading: Mapping[str, Reading]
+    readable: AsyncReadable, expected_reading: dict[str, Reading]
 ) -> None:
     """Assert readings from readable.
 
@@ -79,18 +60,16 @@ async def assert_reading(
 
     """
     actual_reading = await readable.read()
-    assert (
-        _approx_readable_value(expected_reading) == actual_reading
-    ), _generate_assert_error_msg(
-        name=readable.name,
-        expected_result=expected_reading,
-        actual_result=actual_reading,
-    )
+    approx_expected_reading = {
+        k: dict(v, value=approx_value(expected_reading[k]["value"]))
+        for k, v in expected_reading.items()
+    }
+    assert actual_reading == approx_expected_reading
 
 
 async def assert_configuration(
     configurable: AsyncConfigurable,
-    configuration: Mapping[str, Reading],
+    configuration: dict[str, Reading],
 ) -> None:
     """Assert readings from Configurable.
 
@@ -108,14 +87,13 @@ async def assert_configuration(
         await assert_configuration(configurable configuration)
 
     """
-    actual_configurable = await configurable.read_configuration()
-    assert (
-        _approx_readable_value(configuration) == actual_configurable
-    ), _generate_assert_error_msg(
-        name=configurable.name,
-        expected_result=configuration,
-        actual_result=actual_configurable,
-    )
+
+    actual_configuration = await configurable.read_configuration()
+    approx_expected_configuration = {
+        k: dict(v, value=approx_value(configuration[k]["value"]))
+        for k, v in configuration.items()
+    }
+    assert actual_configuration == approx_expected_configuration
 
 
 async def assert_describe_signal(signal: SignalR, /, **metadata):
@@ -126,7 +104,7 @@ async def assert_describe_signal(signal: SignalR, /, **metadata):
     assert actual_datakey == expected_datakey
 
 
-def assert_emitted(docs: Mapping[str, list[dict]], **numbers: int):
+def assert_emitted(docs: dict[str, list[dict]], **numbers: int):
     """Assert emitted document generated by running a Bluesky plan
 
     Parameters
@@ -145,17 +123,9 @@ def assert_emitted(docs: Mapping[str, list[dict]], **numbers: int):
         RE(my_plan())
         assert_emitted(docs, start=1, descriptor=1, event=1, stop=1)
     """
-    assert list(docs) == list(numbers), _generate_assert_error_msg(
-        name="documents",
-        expected_result=list(numbers),
-        actual_result=list(docs),
-    )
+    assert list(docs) == list(numbers)
     actual_numbers = {name: len(d) for name, d in docs.items()}
-    assert actual_numbers == numbers, _generate_assert_error_msg(
-        name="emitted",
-        expected_result=numbers,
-        actual_result=actual_numbers,
-    )
+    assert actual_numbers == numbers
 
 
 class ApproxTable:
@@ -175,28 +145,26 @@ def __eq__(self, value):
 
 
 class MonitorQueue(AbstractContextManager):
-    def __init__(self, signal: SignalR):
+    def __init__(self, signal: SignalR, monotonic=False):
         self.signal = signal
         self.updates: asyncio.Queue[dict[str, Reading]] = asyncio.Queue()
-        self.signal.subscribe(self.updates.put_nowait)
+        self._monotonic_timestamps = monotonic
 
     async def assert_updates(self, expected_value):
         # Get an update, value and reading
         expected_type = type(expected_value)
-        if isinstance(expected_value, Table):
-            expected_value = ApproxTable(expected_value)
-        else:
-            expected_value = pytest.approx(expected_value)
+        expected_value = approx_value(expected_value)
         update = await self.updates.get()
         value = await self.signal.get_value()
         reading = await self.signal.read()
         # Check they match what we expected
         assert value == expected_value
         assert type(value) is expected_type
+        timestamp = time.monotonic() if self._monotonic_timestamps else time.time()
         expected_reading = {
             self.signal.name: {
                 "value": expected_value,
-                "timestamp": pytest.approx(time.time(), rel=0.1),
+                "timestamp": pytest.approx(timestamp, rel=0.1),
                 "alarm_severity": 0,
             }
         }

src/ophyd_async/testing/_one_of_everything.py

@@ -7,7 +7,9 @@
     Array1D,
     Device,
     DTypeScalar_co,
+    SignalR,
     SignalRW,
+    StandardReadable,
     StrictEnum,
     Table,
     soft_signal_r_and_setter,
@@ -30,14 +32,16 @@ class ExampleTable(Table):
     enum: Sequence[ExampleEnum]
 
 
-def int_array_signal(dtype: type[DTypeScalar_co]) -> SignalRW[Array1D[DTypeScalar_co]]:
+def int_array_signal(
+    dtype: type[DTypeScalar_co], name: str = ""
+) -> SignalRW[Array1D[DTypeScalar_co]]:
     iinfo = np.iinfo(dtype)  # type: ignore
     value = np.array([iinfo.min, iinfo.max, 0, 1, 2, 3, 4], dtype=dtype)
-    return soft_signal_rw(Array1D[dtype], value)
+    return soft_signal_rw(Array1D[dtype], value, name)
 
 
 def float_array_signal(
-    dtype: type[DTypeScalar_co],
+    dtype: type[DTypeScalar_co], name: str = ""
 ) -> SignalRW[Array1D[DTypeScalar_co]]:
     finfo = np.finfo(dtype)  # type: ignore
     value = np.array(
@@ -53,10 +57,11 @@ def float_array_signal(
         ],
         dtype=dtype,
     )
-    return soft_signal_rw(Array1D[dtype], value)
+    return soft_signal_rw(Array1D[dtype], value, name)
 
 
-class OneOfEverythingDevice(Device):
+class OneOfEverythingDevice(StandardReadable):
+    # make a detector to test assert_configuration
     def __init__(self, name=""):
         self.int = soft_signal_rw(int, 1)
         self.float = soft_signal_rw(float, 1.234)
@@ -73,9 +78,13 @@ def __init__(self, name=""):
         self.uint64a = int_array_signal(np.uint64)
         self.float32a = float_array_signal(np.float32)
         self.float64a = float_array_signal(np.float64)
-        self.stra = soft_signal_rw(Sequence[str], ["one", "two", "three"])
+        self.stra = soft_signal_rw(
+            Sequence[str],
+            ["one", "two", "three"],
+        )
         self.enuma = soft_signal_rw(
-            Sequence[ExampleEnum], [ExampleEnum.A, ExampleEnum.C]
+            Sequence[ExampleEnum],
+            [ExampleEnum.A, ExampleEnum.C],
         )
         self.table = soft_signal_rw(
             ExampleTable,
@@ -88,7 +97,11 @@ def __init__(self, name=""):
             ),
         )
         self.ndarray = soft_signal_rw(np.ndarray, np.array(([1, 2, 3], [4, 5, 6])))
-        super().__init__(name=name)
+        # add all signals to configuration
+        self._read_config_funcs = tuple(  # type: ignore
+            sig.read for sig in self.__dict__.values() if isinstance(sig, SignalR)
+        )
+        super().__init__(name)
 
 
 async def _get_signal_values(child: Device) -> dict[SignalRW, Any]: