Skip to content

Commit

Permalink
add pytest.approx to assert reading
Browse files Browse the repository at this point in the history
  • Loading branch information
Relm-Arrowny committed Dec 9, 2024
1 parent f7126df commit 58de902
Show file tree
Hide file tree
Showing 2 changed files with 137 additions and 2 deletions.
16 changes: 14 additions & 2 deletions src/ophyd_async/testing/_assert.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
123 changes: 123 additions & 0 deletions tests/core/test_signal.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from bluesky.protocols import Reading

from ophyd_async.core import (
Array1D,
DeviceCollector,
SignalR,
SignalRW,
Expand Down Expand Up @@ -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)
Expand All @@ -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"""

Expand Down

0 comments on commit 58de902

Please sign in to comment.