Move test_initial_values_set_in_ca to own file, use task instead of subprocess for serve

Author: James Souter

tests/conftest.py

@@ -26,7 +26,6 @@
 from tests.assertable_controller import MyTestAttributeIORef, MyTestController
 from tests.example_p4p_ioc import run as _run_p4p_ioc
 from tests.example_softioc import run as _run_softioc
-from tests.example_softioc import run_initial_value as _run_initial
 
 
 @pytest.fixture(scope="function", autouse=True)
@@ -172,12 +171,6 @@ def softioc_subprocess():
     yield from run_ioc_as_subprocess(_run_softioc, multiprocessing.get_context())
 
 
-@pytest.fixture(scope="module")
-def initial_softioc_subprocess():
-    multiprocessing.set_start_method("spawn", force=True)
-    yield from run_ioc_as_subprocess(_run_initial, multiprocessing.get_context())
-
-
 @pytest.fixture(scope="session")
 def tango_system():
     subprocess.run(

tests/example_softioc.py

@@ -1,22 +1,12 @@
-import enum
-
-import numpy as np
-
 from fastcs.attributes import AttrR, AttrRW, AttrW
 from fastcs.controller import Controller
-from fastcs.datatypes import Bool, Enum, Float, Int, String, Waveform
+from fastcs.datatypes import Int
 from fastcs.launch import FastCS
 from fastcs.transport.epics.ca.transport import EpicsCATransport
 from fastcs.transport.epics.options import EpicsIOCOptions
 from fastcs.wrappers import command
 
 
-class InitialEnum(enum.Enum):
-    A = 0
-    B = 1
-    C = 2
-
-
 class ParentController(Controller):
     a: AttrR = AttrR(Int())
     b: AttrRW = AttrRW(Int())
@@ -30,33 +20,6 @@ async def d(self):
         pass
 
 
-class InitialValuesController(Controller):
-    int = AttrRW(Int(), initial_value=4)
-    float = AttrRW(Float(), initial_value=3.1)
-    bool = AttrRW(Bool(), initial_value=True)
-    enum = AttrRW(Enum(InitialEnum), initial_value=InitialEnum.B)
-    str = AttrRW(String(), initial_value="initial")
-    waveform = AttrRW(
-        Waveform(np.int64, shape=(10,)),
-        initial_value=np.array(range(10), dtype=np.int64),
-    )
-    int_r = AttrR(Int(), initial_value=5)
-    float_r = AttrR(Float(), initial_value=4.1)
-    bool_r = AttrR(Bool(), initial_value=False)
-    enum_r = AttrR(Enum(InitialEnum), initial_value=InitialEnum.C)
-    str_r = AttrR(String(), initial_value="initial_r")
-    waveform_r = AttrR(
-        Waveform(np.int64, shape=(10,)),
-        initial_value=np.array(range(10, 20), dtype=np.int64),
-    )
-    int_w = AttrW(Int())
-    float_w = AttrW(Float())
-    bool_w = AttrW(Bool())
-    enum_w = AttrW(Enum(InitialEnum))
-    str_w = AttrW(String())
-    waveform_w = AttrW(Waveform(np.int64, shape=(10,)))
-
-
 def run(pv_prefix="SOFTIOC_TEST_DEVICE"):
     controller = ParentController()
     controller.child = ChildController()
@@ -66,13 +29,5 @@ def run(pv_prefix="SOFTIOC_TEST_DEVICE"):
     fastcs.run(interactive=False)
 
 
-def run_initial_value(pv_prefix="SOFTIOC_INITIAL_DEVICE"):
-    controller = InitialValuesController()
-    fastcs = FastCS(
-        controller, [EpicsCATransport(ca_ioc=EpicsIOCOptions(pv_prefix=pv_prefix))]
-    )
-    fastcs.run()
-
-
 if __name__ == "__main__":
     run()

tests/transport/epics/ca/test_initial_value.py

@@ -0,0 +1,114 @@
+import asyncio
+import enum
+
+import numpy as np
+import pytest
+from aioca import caget
+
+from fastcs.attributes import AttrR, AttrRW, AttrW
+from fastcs.controller import Controller
+from fastcs.datatypes import Bool, Enum, Float, Int, String, Waveform
+from fastcs.launch import FastCS
+from fastcs.transport.epics.ca.transport import EpicsCATransport
+from fastcs.transport.epics.options import EpicsIOCOptions
+
+
+class InitialEnum(enum.Enum):
+    A = 0
+    B = 1
+    C = 2
+
+
+class InitialValuesController(Controller):
+    int = AttrRW(Int(), initial_value=4)
+    float = AttrRW(Float(), initial_value=3.1)
+    bool = AttrRW(Bool(), initial_value=True)
+    enum = AttrRW(Enum(InitialEnum), initial_value=InitialEnum.B)
+    str = AttrRW(String(), initial_value="initial")
+    waveform = AttrRW(
+        Waveform(np.int64, shape=(10,)),
+        initial_value=np.array(range(10), dtype=np.int64),
+    )
+    int_r = AttrR(Int(), initial_value=5)
+    float_r = AttrR(Float(), initial_value=4.1)
+    bool_r = AttrR(Bool(), initial_value=False)
+    enum_r = AttrR(Enum(InitialEnum), initial_value=InitialEnum.C)
+    str_r = AttrR(String(), initial_value="initial_r")
+    waveform_r = AttrR(
+        Waveform(np.int64, shape=(10,)),
+        initial_value=np.array(range(10, 20), dtype=np.int64),
+    )
+    int_w = AttrW(Int())
+    float_w = AttrW(Float())
+    bool_w = AttrW(Bool())
+    enum_w = AttrW(Enum(InitialEnum))
+    str_w = AttrW(String())
+    waveform_w = AttrW(Waveform(np.int64, shape=(10,)))
+
+
+@pytest.mark.asyncio
+async def test_initial_values_set_in_ca():
+    pv_prefix = "SOFTIOC_INITIAL_DEVICE"
+
+    loop = asyncio.get_event_loop()
+    fastcs = FastCS(
+        InitialValuesController(),
+        [EpicsCATransport(ca_ioc=EpicsIOCOptions(pv_prefix=pv_prefix))],
+        loop,
+    )
+
+    task = asyncio.create_task(fastcs.serve(interactive=False))
+    # combine cagets to reduce timeouts
+
+    # AttrRWs
+    scalar_sets = await caget(
+        [
+            f"{pv_prefix}:Int",
+            f"{pv_prefix}:Float",
+            f"{pv_prefix}:Bool",
+            f"{pv_prefix}:Enum",
+        ]
+    )
+    assert scalar_sets == [4, 3.1, 1, 1]
+    scalar_rbvs = await caget(
+        [
+            f"{pv_prefix}:Int_RBV",
+            f"{pv_prefix}:Float_RBV",
+            f"{pv_prefix}:Bool_RBV",
+            f"{pv_prefix}:Enum_RBV",
+        ]
+    )
+    assert scalar_rbvs == [4, 3.1, 1, 1]
+    assert (await caget(f"{pv_prefix}:Str")).tobytes() == b"initial\0"
+    assert np.array_equal((await caget(f"{pv_prefix}:Waveform")), list(range(10)))
+    assert (await caget(f"{pv_prefix}:Str_RBV")).tobytes() == b"initial\0"
+    assert np.array_equal((await caget(f"{pv_prefix}:Waveform_RBV")), list(range(10)))
+
+    # AttrRs
+    scalars = await caget(
+        [
+            f"{pv_prefix}:IntR",
+            f"{pv_prefix}:FloatR",
+            f"{pv_prefix}:BoolR",
+            f"{pv_prefix}:EnumR",
+        ]
+    )
+    assert scalars == [5, 4.1, 0, 2]
+    assert (await caget(f"{pv_prefix}:StrR")).tobytes() == b"initial_r\0"
+    assert np.array_equal((await caget(f"{pv_prefix}:WaveformR")), list(range(10, 20)))
+
+    # Check AttrWs use the datatype initial value
+    w_scalars = await caget(
+        [
+            f"{pv_prefix}:IntW",
+            f"{pv_prefix}:FloatW",
+            f"{pv_prefix}:BoolW",
+            f"{pv_prefix}:EnumW",
+        ]
+    )
+    assert w_scalars == [0, 0, 0, 0]
+    assert (await caget(f"{pv_prefix}:StrW")).tobytes() == b"\0"
+    # initial waveforms not set with zeros
+    assert np.array_equal((await caget(f"{pv_prefix}:WaveformW")), [])
+
+    task.cancel()

tests/transport/epics/ca/test_softioc_system.py

@@ -1,8 +1,5 @@
 from multiprocessing import Queue
 
-import numpy as np
-import pytest
-from aioca import caget
 from p4p import Value
 from p4p.client.thread import Context
 
@@ -32,61 +29,3 @@ def test_ioc(softioc_subprocess: tuple[str, Queue]):
         "c": {"w": f"{pv_prefix}:Child:C"},
         "d": {"x": f"{pv_prefix}:Child:D"},
     }
-
-
-@pytest.mark.asyncio
-async def test_initial_values_set_in_ca(initial_softioc_subprocess):
-    pv_prefix, _ = initial_softioc_subprocess
-    # combine cagets to reduce timeouts
-
-    # AttrRWs
-
-    scalar_sets = await caget(
-        [
-            f"{pv_prefix}:Int",
-            f"{pv_prefix}:Float",
-            f"{pv_prefix}:Bool",
-            f"{pv_prefix}:Enum",
-        ]
-    )
-    assert scalar_sets == [4, 3.1, 1, 1]
-    scalar_rbvs = await caget(
-        [
-            f"{pv_prefix}:Int_RBV",
-            f"{pv_prefix}:Float_RBV",
-            f"{pv_prefix}:Bool_RBV",
-            f"{pv_prefix}:Enum_RBV",
-        ]
-    )
-    assert scalar_rbvs == [4, 3.1, 1, 1]
-    assert (await caget(f"{pv_prefix}:Str")).tobytes() == b"initial\0"
-    assert np.array_equal((await caget(f"{pv_prefix}:Waveform")), list(range(10)))
-    assert (await caget(f"{pv_prefix}:Str_RBV")).tobytes() == b"initial\0"
-    assert np.array_equal((await caget(f"{pv_prefix}:Waveform_RBV")), list(range(10)))
-
-    # AttrRs
-    scalars = await caget(
-        [
-            f"{pv_prefix}:IntR",
-            f"{pv_prefix}:FloatR",
-            f"{pv_prefix}:BoolR",
-            f"{pv_prefix}:EnumR",
-        ]
-    )
-    assert scalars == [5, 4.1, 0, 2]
-    assert (await caget(f"{pv_prefix}:StrR")).tobytes() == b"initial_r\0"
-    assert np.array_equal((await caget(f"{pv_prefix}:WaveformR")), list(range(10, 20)))
-
-    # Check AttrWs use the datatype initial value
-    w_scalars = await caget(
-        [
-            f"{pv_prefix}:IntW",
-            f"{pv_prefix}:FloatW",
-            f"{pv_prefix}:BoolW",
-            f"{pv_prefix}:EnumW",
-        ]
-    )
-    return
-    assert w_scalars == [0, 0, 0, 0]
-    assert (await caget(f"{pv_prefix}:StrW")).tobytes() == b"\0"
-    assert np.array_equal((await caget(f"{pv_prefix}:WaveformW")), 10 * [0])