From a819a881cf8628938a2759ef4002aa1effe81f6b Mon Sep 17 00:00:00 2001 From: James Souter Date: Thu, 14 Nov 2024 15:13:54 +0000 Subject: [PATCH] Move example IOC and device for epics signal test to epics.testing module --- src/ophyd_async/epics/testing/__init__.py | 27 ++ src/ophyd_async/epics/testing/_example_ioc.py | 78 ++++++ .../epics/testing/{utils.py => _utils.py} | 19 +- .../epics/testing}/test_records.db | 38 +-- .../epics/testing}/test_records_pva.db | 46 ++-- tests/epics/signal/test_signals.py | 241 +++++++----------- 6 files changed, 236 insertions(+), 213 deletions(-) create mode 100644 src/ophyd_async/epics/testing/__init__.py create mode 100644 src/ophyd_async/epics/testing/_example_ioc.py rename src/ophyd_async/epics/testing/{utils.py => _utils.py} (75%) rename {tests/epics/signal => src/ophyd_async/epics/testing}/test_records.db (77%) rename {tests/epics/signal => src/ophyd_async/epics/testing}/test_records_pva.db (79%) diff --git a/src/ophyd_async/epics/testing/__init__.py b/src/ophyd_async/epics/testing/__init__.py new file mode 100644 index 000000000..9509f15d4 --- /dev/null +++ b/src/ophyd_async/epics/testing/__init__.py @@ -0,0 +1,27 @@ +from ._example_ioc import ( + CaAndPvaDevice, + ExampleEnum, + ExampleTable, + PvaDevice, + connect_example_device, + create_example_ioc_fixture, +) +from ._utils import ( + Template, + create_device_fixture, + create_ioc_fixture, + make_ioc_from_templates, +) + +__all__ = [ + "CaAndPvaDevice", + "ExampleEnum", + "ExampleTable", + "PvaDevice", + "connect_example_device", + "create_example_ioc_fixture", + "Template", + "create_device_fixture", + "create_ioc_fixture", + "make_ioc_from_templates", +] diff --git a/src/ophyd_async/epics/testing/_example_ioc.py b/src/ophyd_async/epics/testing/_example_ioc.py new file mode 100644 index 000000000..dc22e65af --- /dev/null +++ b/src/ophyd_async/epics/testing/_example_ioc.py @@ -0,0 +1,78 @@ +from collections.abc import Sequence +from pathlib import Path +from typing import Annotated as A +from typing import Literal + +import numpy as np + +from ophyd_async.core import ( + Array1D, + SignalRW, + StrictEnum, + Table, +) +from ophyd_async.epics.core import ( + EpicsDevice, + PvSuffix, +) + +from ._utils import Template, create_ioc_fixture + +CA_PVA_RECORDS = str(Path(__file__).parent / "test_records.db") +PVA_RECORDS = str(Path(__file__).parent / "test_records_pva.db") + + +class ExampleEnum(StrictEnum): + a = "Aaa" + b = "Bbb" + c = "Ccc" + + +class ExampleTable(Table): + bool: Array1D[np.bool_] + int: Array1D[np.int32] + float: Array1D[np.float64] + str: Sequence[str] + enum: Sequence[ExampleEnum] + + +class CaAndPvaDevice(EpicsDevice): + my_int: A[SignalRW[int], PvSuffix("int")] + my_float: A[SignalRW[float], PvSuffix("float")] + my_str: A[SignalRW[str], PvSuffix("str")] + my_bool: A[SignalRW[bool], PvSuffix("bool")] + enum: A[SignalRW[ExampleEnum], PvSuffix("enum")] + enum2: A[SignalRW[ExampleEnum], PvSuffix("enum2")] + bool_unnamed: A[SignalRW[bool], PvSuffix("bool_unnamed")] + partialint: A[SignalRW[int], PvSuffix("partialint")] + lessint: A[SignalRW[int], PvSuffix("lessint")] + uint8a: A[SignalRW[Array1D[np.uint8]], PvSuffix("uint8a")] + int16a: A[SignalRW[Array1D[np.int16]], PvSuffix("int16a")] + int32a: A[SignalRW[Array1D[np.int32]], PvSuffix("int32a")] + float32a: A[SignalRW[Array1D[np.float32]], PvSuffix("float32a")] + float64a: A[SignalRW[Array1D[np.float64]], PvSuffix("float64a")] + stra: A[SignalRW[Sequence[str]], PvSuffix("stra")] + + +class PvaDevice(CaAndPvaDevice): + int8a: A[SignalRW[Array1D[np.int8]], PvSuffix("int8a")] + uint16a: A[SignalRW[Array1D[np.uint16]], PvSuffix("uint16a")] + uint32a: A[SignalRW[Array1D[np.uint32]], PvSuffix("uint32a")] + int64a: A[SignalRW[Array1D[np.int64]], PvSuffix("int64a")] + uint64a: A[SignalRW[Array1D[np.uint64]], PvSuffix("uint64a")] + table: A[SignalRW[ExampleTable], PvSuffix("table")] + ntndarray_data: A[SignalRW[Array1D[np.int64]], PvSuffix("ntndarray:data")] + + +async def connect_example_device(prefix: str, protocol: Literal["ca", "pva"]): + device_cls = PvaDevice if protocol == "pva" else CaAndPvaDevice + device = device_cls(f"{protocol}://{prefix}") + await device.connect() + return device + + +def create_example_ioc_fixture(prefix: str): + return create_ioc_fixture( + Template(CA_PVA_RECORDS, f"P={prefix}"), + Template(PVA_RECORDS, f"P={prefix}"), + ) diff --git a/src/ophyd_async/epics/testing/utils.py b/src/ophyd_async/epics/testing/_utils.py similarity index 75% rename from src/ophyd_async/epics/testing/utils.py rename to src/ophyd_async/epics/testing/_utils.py index ab6110a28..cd895f8a0 100644 --- a/src/ophyd_async/epics/testing/utils.py +++ b/src/ophyd_async/epics/testing/_utils.py @@ -48,8 +48,8 @@ def make_ioc_from_templates(*templates: Template) -> subprocess.Popen[Any]: return process -def create_ioc_fixture(*templates: Template, fixture_name: str | None): - def make_ioc() -> subprocess.Popen[Any]: # TODO fix the typing here +def create_ioc_fixture(*templates: Template, fixture_name: str | None = None): + def make_ioc(): process = make_ioc_from_templates(*templates) yield process @@ -73,18 +73,3 @@ async def create_device(): return device return pytest.fixture(create_device, name=fixture_name) - - -def create_device_and_ioc_fixtures( - device_cls: type[Device], - prefix: str, - *templates: Template, - device_name: str | None = None, - ioc_name: str | None = None, -): - # Use a module level fixture per protocol so it's fast to run tests. This means - # we need to add a record for every PV that we will modify in tests to stop - # tests interfering with each other - _ioc_fixture = create_ioc_fixture(*templates, fixture_name=ioc_name) - _device_fixture = create_device_fixture(device_cls, prefix, device_name) - return (_ioc_fixture, _device_fixture) diff --git a/tests/epics/signal/test_records.db b/src/ophyd_async/epics/testing/test_records.db similarity index 77% rename from tests/epics/signal/test_records.db rename to src/ophyd_async/epics/testing/test_records.db index 59eea2bee..89e44fedc 100644 --- a/tests/epics/signal/test_records.db +++ b/src/ophyd_async/epics/testing/test_records.db @@ -1,16 +1,16 @@ -record(bo, "$(P)$(R)bool") { +record(bo, "$(P)bool") { field(ZNAM, "No") field(ONAM, "Yes") field(VAL, "1") field(PINI, "YES") } -record(bo, "$(P)$(R)bool_unnamed") { +record(bo, "$(P)bool_unnamed") { field(VAL, "1") field(PINI, "YES") } -record(longout, "$(P)$(R)int") { +record(longout, "$(P)int") { field(LLSV, "MAJOR") # LOLO is alarm field(LSV, "MINOR") # LOW is warning field(HSV, "MINOR") # HIGH is warning @@ -27,7 +27,7 @@ record(longout, "$(P)$(R)int") { field(PINI, "YES") } -record(longout, "$(P)$(R)partialint") { +record(longout, "$(P)partialint") { field(LLSV, "MAJOR") # LOLO is alarm field(HHSV, "MAJOR") # HIHI is alarm field(HOPR, "100") @@ -40,7 +40,7 @@ record(longout, "$(P)$(R)partialint") { field(PINI, "YES") } -record(longout, "$(P)$(R)lessint") { +record(longout, "$(P)lessint") { field(HSV, "MINOR") # LOW is warning field(LSV, "MINOR") # HIGH is warning field(HOPR, "100") @@ -51,33 +51,33 @@ record(longout, "$(P)$(R)lessint") { field(PINI, "YES") } -record(ao, "$(P)$(R)float") { +record(ao, "$(P)float") { field(PREC, "1") field(EGU, "mm") field(VAL, "3.141") field(PINI, "YES") } -record(ao, "$(P)$(R)float_prec_0") { +record(ao, "$(P)float_prec_0") { field(PREC, "0") field(EGU, "mm") field(VAL, "3") field(PINI, "YES") } -record(ao, "$(P)$(R)float_prec_1") { +record(ao, "$(P)float_prec_1") { field(PREC, "1") field(EGU, "mm") field(VAL, "3") field(PINI, "YES") } -record(stringout, "$(P)$(R)str") { +record(stringout, "$(P)str") { field(VAL, "hello") field(PINI, "YES") } -record(mbbo, "$(P)$(R)enum") { +record(mbbo, "$(P)enum") { field(ZRST, "Aaa") field(ZRVL, "5") field(ONST, "Bbb") @@ -88,7 +88,7 @@ record(mbbo, "$(P)$(R)enum") { field(PINI, "YES") } -record(mbbo, "$(P)$(R)enum2") { +record(mbbo, "$(P)enum2") { field(ZRST, "Aaa") field(ONST, "Bbb") field(TWST, "Ccc") @@ -96,56 +96,56 @@ record(mbbo, "$(P)$(R)enum2") { field(PINI, "YES") } -record(waveform, "$(P)$(R)uint8a") { +record(waveform, "$(P)uint8a") { field(NELM, "3") field(FTVL, "UCHAR") field(INP, {const:[0, 255]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)int16a") { +record(waveform, "$(P)int16a") { field(NELM, "3") field(FTVL, "SHORT") field(INP, {const:[-32768, 32767]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)int32a") { +record(waveform, "$(P)int32a") { field(NELM, "3") field(FTVL, "LONG") field(INP, {const:[-2147483648, 2147483647]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)float32a") { +record(waveform, "$(P)float32a") { field(NELM, "3") field(FTVL, "FLOAT") field(INP, {const:[0.000002, -123.123]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)float64a") { +record(waveform, "$(P)float64a") { field(NELM, "3") field(FTVL, "DOUBLE") field(INP, {const:[0.1, -12345678.123]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)stra") { +record(waveform, "$(P)stra") { field(NELM, "3") field(FTVL, "STRING") field(INP, {const:["five", "six", "seven"]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)longstr") { +record(waveform, "$(P)longstr") { field(NELM, "80") field(FTVL, "CHAR") field(INP, {const:"a string that is just longer than forty characters"}) field(PINI, "YES") } -record(lsi, "$(P)$(R)longstr2") { +record(lsi, "$(P)longstr2") { field(SIZV, "80") field(INP, {const:"a string that is just longer than forty characters"}) field(PINI, "YES") diff --git a/tests/epics/signal/test_records_pva.db b/src/ophyd_async/epics/testing/test_records_pva.db similarity index 79% rename from tests/epics/signal/test_records_pva.db rename to src/ophyd_async/epics/testing/test_records_pva.db index 2ccc8e237..5dd2e55ac 100644 --- a/tests/epics/signal/test_records_pva.db +++ b/src/ophyd_async/epics/testing/test_records_pva.db @@ -1,25 +1,25 @@ -record(waveform, "$(P)$(R)int8a") { +record(waveform, "$(P)int8a") { field(NELM, "3") field(FTVL, "CHAR") field(INP, {const:[-128, 127]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)uint16a") { +record(waveform, "$(P)uint16a") { field(NELM, "3") field(FTVL, "USHORT") field(INP, {const:[0, 65535]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)uint32a") { +record(waveform, "$(P)uint32a") { field(NELM, "3") field(FTVL, "ULONG") field(INP, {const:[0, 4294967295]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)int64a") { +record(waveform, "$(P)int64a") { field(NELM, "3") field(FTVL, "INT64") # Can't do 64-bit int with JSON numbers in a const link... @@ -27,20 +27,20 @@ record(waveform, "$(P)$(R)int64a") { field(PINI, "YES") } -record(waveform, "$(P)$(R)uint64a") { +record(waveform, "$(P)uint64a") { field(NELM, "3") field(FTVL, "UINT64") field(INP, {const:[0, 4294967297]}) field(PINI, "YES") } -record(waveform, "$(P)$(R)table:labels") { +record(waveform, "$(P)table:labels") { field(FTVL, "STRING") field(NELM, "5") field(INP, {const:["Bool", "Int", "Float", "Str", "Enum"]}) field(PINI, "YES") info(Q:group, { - "$(P)$(R)table": { + "$(P)table": { "+id": "epics:nt/NTTable:1.0", "labels": { "+type": "plain", @@ -50,14 +50,14 @@ record(waveform, "$(P)$(R)table:labels") { }) } -record(waveform, "$(P)$(R)table:bool") +record(waveform, "$(P)table:bool") { field(FTVL, "UCHAR") field(NELM, "4096") field(INP, {const:[false, false, true, true]}) field(PINI, "YES") info(Q:group, { - "$(P)$(R)table": { + "$(P)table": { "value.bool": { "+type": "plain", "+channel": "VAL", @@ -67,14 +67,14 @@ record(waveform, "$(P)$(R)table:bool") }) } -record(waveform, "$(P)$(R)table:int") +record(waveform, "$(P)table:int") { field(FTVL, "LONG") field(NELM, "4096") field(INP, {const:[1, 8, -9, 32]}) field(PINI, "YES") info(Q:group, { - "$(P)$(R)table": { + "$(P)table": { "value.int": { "+type": "plain", "+channel": "VAL", @@ -84,14 +84,14 @@ record(waveform, "$(P)$(R)table:int") }) } -record(waveform, "$(P)$(R)table:float") +record(waveform, "$(P)table:float") { field(FTVL, "DOUBLE") field(NELM, "4096") field(INP, {const:[1.8, 8.2, -6, 32.9887]}) field(PINI, "YES") info(Q:group, { - "$(P)$(R)table": { + "$(P)table": { "value.float": { "+type": "plain", "+channel": "VAL", @@ -101,14 +101,14 @@ record(waveform, "$(P)$(R)table:float") }) } -record(waveform, "$(P)$(R)table:str") +record(waveform, "$(P)table:str") { field(FTVL, "STRING") field(NELM, "4096") field(INP, {const:["Hello", "World", "Foo", "Bar"]}) field(PINI, "YES") info(Q:group, { - "$(P)$(R)table": { + "$(P)table": { "value.str": { "+type": "plain", "+channel": "VAL", @@ -118,14 +118,14 @@ record(waveform, "$(P)$(R)table:str") }) } -record(waveform, "$(P)$(R)table:enum") +record(waveform, "$(P)table:enum") { field(FTVL, "STRING") field(NELM, "4096") field(INP, {const:["Aaa", "Bbb", "Aaa", "Ccc"]}) field(PINI, "YES") info(Q:group, { - "$(P)$(R)table": { + "$(P)table": { "value.enum": { "+type": "plain", "+channel": "VAL", @@ -137,34 +137,34 @@ record(waveform, "$(P)$(R)table:enum") }) } -record(longout, "$(P)$(R)ntndarray:ArraySize0_RBV") { +record(longout, "$(P)ntndarray:ArraySize0_RBV") { field(VAL, "3") field(PINI, "YES") info(Q:group, { - "$(P)$(R)ntndarray":{ + "$(P)ntndarray":{ "dimension[0].size":{+channel:"VAL", +type:"plain", +putorder:0} } }) } -record(longout, "$(P)$(R)ntndarray:ArraySize1_RBV") { +record(longout, "$(P)ntndarray:ArraySize1_RBV") { field(VAL, "2") field(PINI, "YES") info(Q:group, { - "$(P)$(R)ntndarray":{ + "$(P)ntndarray":{ "dimension[1].size":{+channel:"VAL", +type:"plain", +putorder:0} } }) } -record(waveform, "$(P)$(R)ntndarray:data") +record(waveform, "$(P)ntndarray:data") { field(FTVL, "INT64") field(NELM, "6") field(INP, {const:[0, 0, 0, 0, 0, 0]}) field(PINI, "YES") info(Q:group, { - "$(P)$(R)ntndarray":{ + "$(P)ntndarray":{ +id:"epics:nt/NTNDArray:1.0", "value":{ +type:"any", diff --git a/tests/epics/signal/test_signals.py b/tests/epics/signal/test_signals.py index bc7d020e9..b641d9579 100644 --- a/tests/epics/signal/test_signals.py +++ b/tests/epics/signal/test_signals.py @@ -21,10 +21,8 @@ from ophyd_async.core import ( Array1D, - Device, NotConnected, SignalBackend, - SignalDatatypeT, StrictEnum, SubsetEnum, T, @@ -33,6 +31,7 @@ save_to_yaml, ) from ophyd_async.epics.core import ( + EpicsDevice, epics_signal_r, epics_signal_rw, epics_signal_rw_rbv, @@ -40,68 +39,31 @@ epics_signal_x, ) from ophyd_async.epics.core._signal import _epics_signal_backend # noqa: PLC2701 -from ophyd_async.epics.testing.utils import Template, create_device_and_ioc_fixtures - -CA_PVA_RECORDS = str(Path(__file__).parent / "test_records.db") -PVA_RECORDS = str(Path(__file__).parent / "test_records_pva.db") -PV_PREFIX = "".join(random.choice(string.ascii_lowercase) for _ in range(12)) - -Protocol = Literal["ca", "pva"] +from ophyd_async.epics.testing import ( + ExampleEnum, + ExampleTable, + connect_example_device, + create_example_ioc_fixture, +) +PV_PREFIX = "".join(random.choice(string.ascii_lowercase) for _ in range(12)) + ":" -PARAMETERISE_PROTOCOLS = pytest.mark.parametrize("protocol", get_args(Protocol)) +class MySubsetEnum(SubsetEnum): + a = "Aaa" + b = "Bbb" + c = "Ccc" -class DeviceTestRecordsGroup(Device): - def __init__(self, name: str): - super().__init__(name) - self._all_signals = {"ca": {}, "pva": {}} - for protocol in get_args(Protocol): - self.__create_signal(protocol, int, "int") - self.__create_signal(protocol, float, "float") - self.__create_signal(protocol, str, "str") - self.__create_signal(protocol, MyEnum, "enum") - self.__create_signal(protocol, MyEnum, "enum2") - self.__create_signal(protocol, bool, "bool") - self.__create_signal(protocol, bool, "bool_unnamed") - self.__create_signal(protocol, int, "partialint") - self.__create_signal(protocol, int, "lessint") - self.__create_signal(protocol, Array1D[np.uint8], "uint8a") - self.__create_signal(protocol, Array1D[np.int16], "int16a") - self.__create_signal(protocol, Array1D[np.int32], "int32a") - self.__create_signal(protocol, Array1D[np.float32], "float32a") - self.__create_signal(protocol, Array1D[np.float64], "float64a") - self.__create_signal(protocol, Sequence[str], "stra") - self.__create_signal("pva", Array1D[np.int8], "int8a") - self.__create_signal("pva", Array1D[np.uint16], "uint16a") - self.__create_signal("pva", Array1D[np.uint32], "uint32a") - self.__create_signal("pva", Array1D[np.int64], "int64a") - self.__create_signal("pva", Array1D[np.uint64], "uint64a") - self.__create_signal("pva", MyTable, "table") - self.__create_signal("pva", Array1D[np.int64], "ntndarray:data") - - def __create_signal( - self, protocol: Protocol, dtype: type[SignalDatatypeT], name: str - ): - signal = epics_signal_rw(dtype, f"{protocol}://{self._name}:{protocol}:{name}") - setattr(self, f"{protocol}_{name}".replace(":", "_"), signal) - self._all_signals[protocol][name] = signal - def get_signal(self, protocol: Protocol, name: str): - return self._all_signals[protocol][name] +Protocol = Literal["ca", "pva"] +PARAMETERISE_PROTOCOLS = pytest.mark.parametrize("protocol", get_args(Protocol)) -ioc, device_test_records = create_device_and_ioc_fixtures( - DeviceTestRecordsGroup, - PV_PREFIX, - Template(CA_PVA_RECORDS, f"P={PV_PREFIX}:,R=ca:"), - Template(CA_PVA_RECORDS, f"P={PV_PREFIX}:,R=pva:"), - Template(PVA_RECORDS, f"P={PV_PREFIX}:,R=pva:"), -) +ioc = create_example_ioc_fixture(PV_PREFIX) async def _make_backend(typ: type | None, protocol: str, suff: str, timeout=10.0): - pv = f"{protocol}://{PV_PREFIX}:{protocol}:{suff}" + pv = f"{protocol}://{PV_PREFIX}{suff}" # Make and connect the backend backend = _epics_signal_backend(typ, pv, pv) await backend.connect(timeout=timeout) @@ -165,7 +127,7 @@ def _is_numpy_subclass(t): async def assert_monitor_then_put( - device: DeviceTestRecordsGroup, + device: EpicsDevice, suffix: str, protocol: Protocol, datakey: dict, @@ -174,13 +136,13 @@ async def assert_monitor_then_put( datatype: type[T] | None = None, check_type: bool | None = True, ): - signal = device.get_signal(protocol, suffix) + signal = getattr(device, suffix) backend = signal._connector.backend # Make a monitor queue that will monitor for updates q = MonitorQueue(backend) try: # Check datakey - source = f"{protocol}://{PV_PREFIX}:{protocol}:{suffix}" + source = f"{protocol}://{PV_PREFIX}{suffix}" assert dict(source=source, **datakey) == await backend.get_datakey(source) # Check initial value await q.assert_updates( @@ -196,18 +158,6 @@ async def assert_monitor_then_put( q.close() -class MyEnum(StrictEnum): - a = "Aaa" - b = "Bbb" - c = "Ccc" - - -class MySubsetEnum(SubsetEnum): - a = "Aaa" - b = "Bbb" - c = "Ccc" - - _metadata: dict[str, dict[str, dict[str, Any]]] = { "ca": { "boolean": {"units": ANY, "limits": ANY}, @@ -298,19 +248,18 @@ def get_dtype_numpy(suffix: str) -> str: # type: ignore async def assert_backend_get_put_monitor( - ioc, datatype: type[T], suffix: str, initial_value: T, put_value: T, tmp_path: Path, protocol: Protocol, - device_test_records, + device, ): # With the given datatype, check we have the correct initial value and putting # works await assert_monitor_then_put( - device_test_records, + device, suffix, protocol, datakey(protocol, suffix, initial_value), # type: ignore @@ -320,7 +269,7 @@ async def assert_backend_get_put_monitor( ) # With datatype guessed from CA/PVA, check we can set it back to the initial value await assert_monitor_then_put( - device_test_records, + device, suffix, protocol, datakey(protocol, suffix, put_value), # type: ignore @@ -340,10 +289,10 @@ async def assert_backend_get_put_monitor( "datatype, suffix, initial_value, put_value", [ # python builtin scalars - (int, "int", 42, 43), - (float, "float", 3.141, 43.5), - (str, "str", "hello", "goodbye"), - (MyEnum, "enum", MyEnum.b, MyEnum.c), + (int, "my_int", 42, 43), + (float, "my_float", 3.141, 43.5), + (str, "my_str", "hello", "goodbye"), + (ExampleEnum, "enum", ExampleEnum.b, ExampleEnum.c), # numpy arrays of numpy types ( Array1D[np.uint8], @@ -394,17 +343,16 @@ async def test_backend_get_put_monitor( put_value: T, tmp_path: Path, protocol: Protocol, - device_test_records, ): + device = await connect_example_device(PV_PREFIX, protocol) await assert_backend_get_put_monitor( - ioc, datatype, suffix, initial_value, put_value, tmp_path, protocol, - device_test_records, + device, ) @@ -453,25 +401,24 @@ async def test_backend_get_put_monitor_pva( initial_value: T, put_value: T, tmp_path: Path, - device_test_records, ): protocol = "pva" + device = await connect_example_device(PV_PREFIX, protocol) await assert_backend_get_put_monitor( - ioc, datatype, suffix, initial_value, put_value, tmp_path, protocol, - device_test_records, + device, ) @PARAMETERISE_PROTOCOLS -@pytest.mark.parametrize("suffix", ["bool", "bool_unnamed"]) +@pytest.mark.parametrize("suffix", ["my_bool", "bool_unnamed"]) async def test_bool_conversion_of_enum( - suffix: str, tmp_path: Path, ioc, device_test_records, protocol + suffix: str, tmp_path: Path, ioc, protocol ) -> None: """Booleans are converted to Short Enumerations with values 0,1 as database does not support boolean natively. @@ -482,8 +429,9 @@ async def test_bool_conversion_of_enum( """ # With the given datatype, check we have the correct initial value and putting # works + device = await connect_example_device(PV_PREFIX, protocol) await assert_monitor_then_put( - device_test_records, + device, suffix, protocol, datakey(protocol, suffix), @@ -493,7 +441,7 @@ async def test_bool_conversion_of_enum( ) # With datatype guessed from CA/PVA, check we can set it back to the initial value await assert_monitor_then_put( - device_test_records, + device, suffix, protocol, datakey(protocol, suffix, True), @@ -509,16 +457,15 @@ async def test_bool_conversion_of_enum( @PARAMETERISE_PROTOCOLS -async def test_error_raised_on_disconnected_PV( - ioc, device_test_records, protocol -) -> None: +async def test_error_raised_on_disconnected_PV(ioc, protocol) -> None: if protocol == "pva": expected = "pva://Disconnect" elif protocol == "ca": expected = "ca://Disconnect" else: raise TypeError() - signal = device_test_records.get_signal(protocol, "bool") + device = await connect_example_device(PV_PREFIX, protocol) + signal = device.my_bool backend = signal._connector.backend # The below will work without error await signal.set(False) @@ -641,12 +588,12 @@ async def test_backend_wrong_type_errors(ioc, typ, suff, errors, protocol): @PARAMETERISE_PROTOCOLS -async def test_backend_put_enum_string(ioc, device_test_records, protocol) -> None: - sig = device_test_records.get_signal(protocol, "enum2") - backend = sig._connector.backend +async def test_backend_put_enum_string(ioc, protocol) -> None: + device = await connect_example_device(PV_PREFIX, protocol) + backend = device.enum2._connector.backend # Don't do this in production code, but allow on CLI await backend.put("Ccc", wait=True) # type: ignore - assert MyEnum.c == await backend.get_value() + assert ExampleEnum.c == await backend.get_value() @PARAMETERISE_PROTOCOLS @@ -656,11 +603,11 @@ async def test_backend_enum_which_doesnt_inherit_string(ioc, protocol) -> None: @PARAMETERISE_PROTOCOLS -async def test_backend_get_setpoint(ioc, device_test_records, protocol) -> None: - sig = device_test_records.get_signal(protocol, "enum2") - backend = sig._connector.backend +async def test_backend_get_setpoint(ioc, protocol) -> None: + device = await connect_example_device(PV_PREFIX, protocol) + backend = device.enum2._connector.backend await backend.put("Ccc", wait=True) - assert await backend.get_setpoint() == MyEnum.c + assert await backend.get_setpoint() == ExampleEnum.c def approx_table(datatype: type[Table], table: Table): @@ -673,33 +620,25 @@ def approx_table(datatype: type[Table], table: Table): return new_table -class MyTable(Table): - bool: Array1D[np.bool_] - int: Array1D[np.int32] - float: Array1D[np.float64] - str: Sequence[str] - enum: Sequence[MyEnum] - - async def test_pva_table(ioc) -> None: protocol: Protocol = "pva" # CA can't do tables - initial = MyTable( + initial = ExampleTable( bool=np.array([False, False, True, True], np.bool_), int=np.array([1, 8, -9, 32], np.int32), float=np.array([1.8, 8.2, -6, 32.9887], np.float64), str=["Hello", "World", "Foo", "Bar"], - enum=[MyEnum.a, MyEnum.b, MyEnum.a, MyEnum.c], + enum=[ExampleEnum.a, ExampleEnum.b, ExampleEnum.a, ExampleEnum.c], ) - put = MyTable( + put = ExampleTable( bool=np.array([True, False], np.bool_), int=np.array([-5, 32], np.int32), float=np.array([8.5, -6.97], np.float64), str=["Hello", "Bat"], - enum=[MyEnum.c, MyEnum.b], + enum=[ExampleEnum.c, ExampleEnum.b], ) # Make and connect the backend - for t, i, p in [(MyTable, initial, put), (None, put, initial)]: + for t, i, p in [(ExampleTable, initial, put), (None, put, initial)]: backend = await _make_backend(t, protocol, "table") # Make a monitor queue that will monitor for updates q = MonitorQueue(backend) @@ -730,7 +669,7 @@ async def test_pva_table(ioc) -> None: q.close() -async def test_pva_ntdarray(ioc, device_test_records): +async def test_pva_ntndarray(ioc): protocol = "pva" # CA can't do ndarray @@ -738,11 +677,10 @@ async def test_pva_ntdarray(ioc, device_test_records): initial = np.zeros_like(put) backend = await _make_backend(np.ndarray, protocol, "ntndarray") - # Backdoor into the "raw" data underlying the NDArray in QSrv # not supporting direct writes to NDArray at the moment. - signal = device_test_records.get_signal(protocol, "ntndarray:data") - raw_data_backend = signal._connector.backend + device = await connect_example_device(PV_PREFIX, protocol) + raw_data_backend = device.ntndarray_data._connector.backend # Make a monitor queue that will monitor for updates for i, p in [(initial, put), (put, initial)]: @@ -814,17 +752,18 @@ def test_signal_helpers(): @PARAMETERISE_PROTOCOLS -async def test_str_enum_returns_enum(ioc, device_test_records, protocol): - sig = device_test_records.get_signal(protocol, "enum") - val = await sig.get_value() - assert repr(val) == "" - assert val is MyEnum.b +async def test_str_enum_returns_enum(ioc, protocol): + device = await connect_example_device(PV_PREFIX, protocol) + val = await device.enum.get_value() + assert repr(val) == "" + assert val is ExampleEnum.b assert val == "Bbb" @PARAMETERISE_PROTOCOLS -async def test_str_datatype_in_mbbo(ioc, device_test_records, protocol): - sig = device_test_records.get_signal(protocol, "enum") +async def test_str_datatype_in_mbbo(ioc, protocol): + device = await connect_example_device(PV_PREFIX, protocol) + sig = device.enum backend = sig._connector.backend datakey = await backend.get_datakey(sig.source) assert datakey["choices"] == ["Aaa", "Bbb", "Ccc"] @@ -837,7 +776,7 @@ async def test_str_datatype_in_mbbo(ioc, device_test_records, protocol): @PARAMETERISE_PROTOCOLS async def test_runtime_enum_returns_str(ioc, protocol): - pv_name = f"{protocol}://{PV_PREFIX}:{protocol}:enum" + pv_name = f"{protocol}://{PV_PREFIX}enum" sig = epics_signal_rw(MySubsetEnum, pv_name) await sig.connect() @@ -846,26 +785,24 @@ async def test_runtime_enum_returns_str(ioc, protocol): @PARAMETERISE_PROTOCOLS -async def test_signal_returns_units_and_precision(ioc, device_test_records, protocol): - sig = device_test_records.get_signal(protocol, "float") - await sig.connect() +async def test_signal_returns_units_and_precision(ioc, protocol): + device = await connect_example_device(PV_PREFIX, protocol) + sig = device.my_float datakey = (await sig.describe())[""] assert datakey["units"] == "mm" assert datakey["precision"] == 1 @PARAMETERISE_PROTOCOLS -async def test_signal_not_return_none_units_and_precision( - ioc, device_test_records, protocol -): - sig = device_test_records.get_signal(protocol, "str") - datakey = (await sig.describe())[""] +async def test_signal_not_return_none_units_and_precision(ioc, protocol): + device = await connect_example_device(PV_PREFIX, protocol) + datakey = (await device.my_str.describe())[""] assert not hasattr(datakey, "units") assert not hasattr(datakey, "precision") @PARAMETERISE_PROTOCOLS -async def test_signal_returns_limits(ioc, device_test_records, protocol): +async def test_signal_returns_limits(ioc, protocol): expected_limits = Limits( # LOW, HIGH warning=LimitsRange(low=5.0, high=96.0), @@ -877,13 +814,13 @@ async def test_signal_returns_limits(ioc, device_test_records, protocol): alarm=LimitsRange(low=2.0, high=98.0), ) - sig = device_test_records.get_signal(protocol, "int") - limits = (await sig.describe())[""]["limits"] + device = await connect_example_device(PV_PREFIX, protocol) + limits = (await device.my_int.describe())[""]["limits"] assert limits == expected_limits @PARAMETERISE_PROTOCOLS -async def test_signal_returns_partial_limits(ioc, device_test_records, protocol): +async def test_signal_returns_partial_limits(ioc, protocol): expected_limits = Limits( # LOLO, HIHI alarm=LimitsRange(low=2.0, high=98.0), @@ -895,16 +832,13 @@ async def test_signal_returns_partial_limits(ioc, device_test_records, protocol) if protocol == "ca": # HSV, LSV not set, but still present for CA expected_limits["warning"] = LimitsRange(low=0, high=0) - - sig = device_test_records.get_signal(protocol, "partialint") - limits = (await sig.describe())[""]["limits"] + device = await connect_example_device(PV_PREFIX, protocol) + limits = (await device.partialint.describe())[""]["limits"] assert limits == expected_limits @PARAMETERISE_PROTOCOLS -async def test_signal_returns_warning_and_partial_limits( - ioc, device_test_records, protocol -): +async def test_signal_returns_warning_and_partial_limits(ioc, protocol): expected_limits = Limits( # control = display if DRVL, DRVH not set control=LimitsRange(low=0.0, high=100.0), @@ -916,42 +850,41 @@ async def test_signal_returns_warning_and_partial_limits( if protocol == "ca": # HSV, LSV not set, but still present for CA expected_limits["alarm"] = LimitsRange(low=0, high=0) - - sig = device_test_records.get_signal(protocol, "lessint") + device = await connect_example_device(PV_PREFIX, protocol) + sig = device.lessint await sig.connect() limits = (await sig.describe())[""]["limits"] assert limits == expected_limits @PARAMETERISE_PROTOCOLS -async def test_signal_not_return_no_limits(ioc, device_test_records, protocol): - signal = device_test_records.get_signal(protocol, "enum") - datakey = (await signal.describe())[""] +async def test_signal_not_return_no_limits(ioc, protocol): + device = await connect_example_device(PV_PREFIX, protocol) + datakey = (await device.enum.describe())[""] assert not hasattr(datakey, "limits") @PARAMETERISE_PROTOCOLS async def test_signals_created_for_prec_0_float_can_use_int(ioc, protocol): - pv_name = f"{protocol}://{PV_PREFIX}:{protocol}:float_prec_0" + pv_name = f"{protocol}://{PV_PREFIX}float_prec_0" sig = epics_signal_rw(int, pv_name) await sig.connect() @PARAMETERISE_PROTOCOLS async def test_signals_created_for_not_prec_0_float_cannot_use_int(ioc, protocol): - pv_name = f"{protocol}://{PV_PREFIX}:{protocol}:float_prec_1" + pv_name = f"{protocol}://{PV_PREFIX}float_prec_1" sig = epics_signal_rw(int, pv_name) with pytest.raises( TypeError, - match=f"{protocol}:float_prec_1 with inferred datatype float" - ".* cannot be coerced to int", + match="float_prec_1 with inferred datatype float" ".* cannot be coerced to int", ): await sig.connect() @PARAMETERISE_PROTOCOLS async def test_bool_works_for_mismatching_enums(ioc, protocol): - pv_name = f"{protocol}://{PV_PREFIX}:{protocol}:bool" + pv_name = f"{protocol}://{PV_PREFIX}bool" sig = epics_signal_rw(bool, pv_name, pv_name + "_unnamed") await sig.connect() @@ -959,8 +892,8 @@ async def test_bool_works_for_mismatching_enums(ioc, protocol): @pytest.mark.skipif(os.name == "nt", reason="Hangs on windows for unknown reasons") @PARAMETERISE_PROTOCOLS async def test_can_read_using_ophyd_async_then_ophyd(ioc, protocol): - oa_read = f"{protocol}://{PV_PREFIX}:{protocol}:float_prec_1" - ophyd_read = f"{PV_PREFIX}:{protocol}:float_prec_0" + oa_read = f"{protocol}://{PV_PREFIX}float_prec_1" + ophyd_read = f"{PV_PREFIX}float_prec_0" ophyd_async_sig = epics_signal_rw(float, oa_read) await ophyd_async_sig.connect()