Move example IOC and device for epics signal test to epics.testing mo…

…dule

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",
+]

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}"),
+    )

src/ophyd_async/epics/testing/utils.py → 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)

tests/epics/signal/test_records.db → 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,64 +88,64 @@ 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")
   field(VAL, "1")
   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")