From 58de902a065e7ea47ae1a16676f887362aa1b09a Mon Sep 17 00:00:00 2001 From: Relm-Arrowny Date: Mon, 9 Dec 2024 16:27:53 +0000 Subject: [PATCH] add pytest.approx to assert reading --- src/ophyd_async/testing/_assert.py | 16 +++- tests/core/test_signal.py | 123 +++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+), 2 deletions(-) diff --git a/src/ophyd_async/testing/_assert.py b/src/ophyd_async/testing/_assert.py index 25d38d12b6..2e6f2c98e4 100644 --- a/src/ophyd_async/testing/_assert.py +++ b/src/ophyd_async/testing/_assert.py @@ -1,6 +1,7 @@ from collections.abc import Mapping from typing import Any +import pytest from bluesky.protocols import Reading from ophyd_async.core import AsyncConfigurable, AsyncReadable, SignalDatatypeT, SignalR @@ -41,6 +42,13 @@ async def assert_value(signal: SignalR[SignalDatatypeT], value: Any) -> None: ) +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 + + async def assert_reading( readable: AsyncReadable, expected_reading: Mapping[str, Reading] ) -> None: @@ -61,7 +69,9 @@ async def assert_reading( """ actual_reading = await readable.read() - assert expected_reading == actual_reading, _generate_assert_error_msg( + assert expected_reading == _approx_readable_value( + actual_reading + ), _generate_assert_error_msg( name=readable.name, expected_result=expected_reading, actual_result=actual_reading, @@ -89,7 +99,9 @@ async def assert_configuration( """ actual_configurable = await configurable.read_configuration() - assert configuration == actual_configurable, _generate_assert_error_msg( + assert configuration == _approx_readable_value( + actual_configurable + ), _generate_assert_error_msg( name=configurable.name, expected_result=configuration, actual_result=actual_configurable, diff --git a/tests/core/test_signal.py b/tests/core/test_signal.py index 8f88c653a8..3f291f7703 100644 --- a/tests/core/test_signal.py +++ b/tests/core/test_signal.py @@ -11,6 +11,7 @@ from bluesky.protocols import Reading from ophyd_async.core import ( + Array1D, DeviceCollector, SignalR, SignalRW, @@ -257,6 +258,15 @@ async def mock_signal(): yield mock_signal +@pytest.fixture +async def mock_signal_array(): + mock_signal_array = epics_signal_rw( + Array1D[np.int8], "pva://mock_signal", name="mock_signal" + ) + await mock_signal_array.connect(mock=True) + yield mock_signal_array + + async def test_assert_value(mock_signal: SignalRW): set_mock_value(mock_signal, 168) await assert_value(mock_signal, 168) @@ -279,6 +289,119 @@ async def test_failed_assert_reading(mock_signal: SignalRW): await assert_reading(mock_signal, dummy_reading) +class DummyReadableArray(StandardReadable): + """A demo Readable to produce read and config signal""" + + def __init__(self, prefix: str, name="") -> None: + # Define some signals + with self.add_children_as_readables(Format.HINTED_SIGNAL): + self.value = epics_signal_r(Array1D[np.int8], prefix + "Value") + self.value2 = epics_signal_r(Array1D[np.float32], prefix + "Value") + # Set name and signals for read() and read_configuration() + with self.add_children_as_readables(Format.CONFIG_SIGNAL): + self.mode = epics_signal_rw(Array1D[np.int8], prefix + "array1") + self.mode2 = epics_signal_rw(Array1D[np.float64], prefix + "array2") + super().__init__(name=name) + + +@pytest.fixture +async def mock_readable_array(): + async with DeviceCollector(mock=True): + mock_readable_array = DummyReadableArray("SIM:READABLE:", name="mock_readable") + + yield mock_readable_array + + +async def test_assert_reading_array(mock_readable_array: DummyReadableArray): + set_mock_value(mock_readable_array.value, np.array([1, 2, 4, 6])) + set_mock_value(mock_readable_array.value2, np.array([1, 2, 4, 7])) + dummy_reading = { + "mock_readable-value": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": [1, 2, 4, 6], + } + ), + "mock_readable-value2": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": [1, 2, 4, 7], + } + ), + } + await assert_reading(mock_readable_array, dummy_reading) + + +async def test_assert_reading_array_fail(mock_readable_array: DummyReadableArray): + set_mock_value(mock_readable_array.value, np.array([1, 2, 4, 6])) + set_mock_value(mock_readable_array.value2, np.array([1, 2, 4, 7])) + dummy_reading = { + "mock_readable-value": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": [1, 2, 4, 6], + } + ), + "mock_readable-value2": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": [1, 2, 4, 7.1], + } + ), + } + with pytest.raises(AssertionError): + await assert_reading(mock_readable_array, dummy_reading) + + +async def test_assert_configuraion_array(mock_readable_array: DummyReadableArray): + set_mock_value(mock_readable_array.mode, np.array([1, 2, 4, 6])) + set_mock_value(mock_readable_array.mode2, np.array([1, 2, 4, 7])) + dummy_reading = { + "mock_readable-mode": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": [1, 2, 4, 6], + } + ), + "mock_readable-mode2": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": [1, 2, 4, 7], + } + ), + } + await assert_configuration(mock_readable_array, dummy_reading) + + +async def test_assert_configuraion_array_fail(mock_readable_array: DummyReadableArray): + set_mock_value(mock_readable_array.mode, np.array([1, 2, 4, 6])) + set_mock_value(mock_readable_array.mode2, np.array([1, 2, 4, 7])) + dummy_reading = { + "mock_readable-mode": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": [1, 2, 4, 6], + } + ), + "mock_readable-mode2": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": [1, 2, 0.4, 7], + } + ), + } + with pytest.raises(AssertionError): + await assert_configuration(mock_readable_array, dummy_reading) + + class DummyReadable(StandardReadable): """A demo Readable to produce read and config signal"""