diff --git a/src/ophyd_async/testing/_assert.py b/src/ophyd_async/testing/_assert.py index 25d38d12b6..849c1a74ec 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 ( + _approx_readable_value(expected_reading) == 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 ( + _approx_readable_value(configuration) == 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..c88597ec12 100644 --- a/tests/core/test_signal.py +++ b/tests/core/test_signal.py @@ -11,11 +11,13 @@ from bluesky.protocols import Reading from ophyd_async.core import ( + Array1D, DeviceCollector, SignalR, SignalRW, SoftSignalBackend, StandardReadable, + StrictEnum, set_and_wait_for_other_value, set_and_wait_for_value, soft_signal_r_and_setter, @@ -250,76 +252,148 @@ async def test_create_soft_signal(signal_method, signal_class): assert (await signal.get_value()) == INITIAL_VALUE -@pytest.fixture -async def mock_signal(): - mock_signal = epics_signal_rw(int, "pva://mock_signal", name="mock_signal") - await mock_signal.connect(mock=True) - yield mock_signal - - -async def test_assert_value(mock_signal: SignalRW): - set_mock_value(mock_signal, 168) - await assert_value(mock_signal, 168) - - -async def test_assert_reaading(mock_signal: SignalRW): - set_mock_value(mock_signal, 888) - dummy_reading = { - "mock_signal": Reading({"alarm_severity": 0, "timestamp": ANY, "value": 888}) - } - await assert_reading(mock_signal, dummy_reading) - - -async def test_failed_assert_reading(mock_signal: SignalRW): - set_mock_value(mock_signal, 888) - dummy_reading = { - "mock_signal": Reading({"alarm_severity": 0, "timestamp": ANY, "value": 88}) - } - with pytest.raises(AssertionError): - await assert_reading(mock_signal, dummy_reading) +class MockEnum(StrictEnum): + GOOD = "Good" + OK = "Ok" -class DummyReadable(StandardReadable): +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(float, prefix + "Value") - with self.add_children_as_readables(Format.CONFIG_SIGNAL): - self.mode = epics_signal_rw(str, prefix + "Mode") - self.mode2 = epics_signal_rw(str, prefix + "Mode2") + self.int_value = epics_signal_r(int, prefix + "int") + self.int_array = epics_signal_r(Array1D[np.int8], prefix + "Value") + self.float_array = 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.str_value = epics_signal_rw(str, prefix + "Value") + self.strictEnum_value = epics_signal_rw(MockEnum, prefix + "array2") super().__init__(name=name) @pytest.fixture async def mock_readable(): async with DeviceCollector(mock=True): - mock_readable = DummyReadable("SIM:READABLE:", name="mock_readable") - + mock_readable = DummyReadableArray("SIM:READABLE:", name="mock_readable") yield mock_readable -async def test_assert_configuration(mock_readable: DummyReadable): - set_mock_value(mock_readable.value, 123) - set_mock_value(mock_readable.mode, "super mode") - set_mock_value(mock_readable.mode2, "slow mode") - dummy_config_reading = { - "mock_readable-mode": ( +async def test_assert_value(mock_readable: DummyReadableArray): + set_mock_value(mock_readable.int_value, 168) + await assert_value(mock_readable.int_value, 168) + + +async def test_assert_reaading(mock_readable: DummyReadableArray): + set_mock_value(mock_readable.int_value, 188) + set_mock_value(mock_readable.int_array, np.array([1, 2, 4, 7])) + set_mock_value(mock_readable.float_array, np.array([1.1231, -2.3, 451.15, 6.6233])) + + dummy_reading = { + "mock_readable-int_value": Reading( + {"alarm_severity": 0, "timestamp": ANY, "value": 188} + ), + "mock_readable-int_array": Reading( + {"alarm_severity": 0, "timestamp": ANY, "value": [1, 2, 4, 7]} + ), + "mock_readable-float_array": Reading( { "alarm_severity": 0, "timestamp": ANY, - "value": "super mode", + "value": [1.1231, -2.3, 451.15, 6.6233], } ), - "mock_readable-mode2": { - "alarm_severity": 0, - "timestamp": ANY, - "value": "slow mode", - }, } - await assert_configuration(mock_readable, dummy_config_reading) + await assert_reading(mock_readable, dummy_reading) + + +@pytest.mark.parametrize( + "int_value, int_array, float_array", + [ + ([128, np.array([1, 2, 4, 7]), np.array([1.1231, -2.3, 451.15, 6.6233])]), + ([188, np.array([-5, 2, 4, 7]), np.array([1.1231, -2.3, 451.15, 6.6233])]), + ([188, np.array([1, 2, 4, 7]), np.array([1.231, -2.3, 451.15, 6.6233])]), + ], +) +async def test_failed_assert_reading( + mock_readable: DummyReadableArray, int_value, int_array, float_array +): + set_mock_value(mock_readable.int_value, 188) + set_mock_value(mock_readable.int_array, np.array([1, 2, 4, 7])) + set_mock_value(mock_readable.float_array, np.array([1.1231, -2.3, 451.15, 6.6233])) + + dummy_reading = { + "mock_readable-int_value": Reading( + {"alarm_severity": 0, "timestamp": ANY, "value": int_value} + ), + "mock_readable-int_array": Reading( + {"alarm_severity": 0, "timestamp": ANY, "value": int_array} + ), + "mock_readable-float_array": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": float_array, + } + ), + } + with pytest.raises(AssertionError): + await assert_reading(mock_readable, dummy_reading) + + +async def test_assert_configuraion(mock_readable: DummyReadableArray): + set_mock_value(mock_readable.str_value, "haha") + set_mock_value(mock_readable.strictEnum_value, MockEnum.GOOD) + dummy_reading = { + "mock_readable-str_value": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": "haha", + } + ), + "mock_readable-strictEnum_value": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": MockEnum.GOOD, + } + ), + } + await assert_configuration(mock_readable, dummy_reading) + + +@pytest.mark.parametrize( + "str_value, enum_value", + [ + ("ha", MockEnum.OK), + ("not funny", MockEnum.GOOD), + ], +) +async def test_assert_configuraion_fail( + mock_readable: DummyReadableArray, str_value, enum_value +): + set_mock_value(mock_readable.str_value, "haha") + set_mock_value(mock_readable.strictEnum_value, MockEnum.GOOD) + dummy_reading = { + "mock_readable-mode": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": str_value, + } + ), + "mock_readable-mode2": Reading( + { + "alarm_severity": 0, + "timestamp": ANY, + "value": enum_value, + } + ), + } + with pytest.raises(AssertionError): + await assert_configuration(mock_readable, dummy_reading) async def test_signal_get_and_set_logging(caplog):