Add system test

Add a fixture to run an EPICS IOC in a subprocess.
Add a test that queries the PVI PV of an IOC and verifies the structure
is as expected.

Author: GDYendell

pyproject.toml

@@ -41,6 +41,8 @@ dev = [
     "sphinx-design",
     "tox-direct",
     "types-mock",
+    "aioca",
+    "p4p",
 ]
 
 [project.scripts]

tests/backends/epics/test_ioc_system.py

@@ -0,0 +1,24 @@
+from p4p.client.thread import Context
+
+
+def test_ioc(ioc: None):
+    ctxt = Context("pva")
+
+    parent_pvi = ctxt.get("DEVICE:PVI").todict()
+    assert all(f in parent_pvi for f in ("alarm", "display", "timeStamp", "value"))
+    assert parent_pvi["display"] == {"description": "The records in this controller"}
+    assert parent_pvi["value"] == {
+        "a": {"r": "DEVICE:A"},
+        "b": {"r": "DEVICE:B_RBV", "w": "DEVICE:B"},
+        "child": {"d": "DEVICE:Child:PVI"},
+    }
+
+    child_pvi_pv = parent_pvi["value"]["child"]["d"]
+    child_pvi = ctxt.get(child_pvi_pv).todict()
+    assert all(f in child_pvi for f in ("alarm", "display", "timeStamp", "value"))
+    assert child_pvi["display"] == {"description": "The records in this controller"}
+    assert child_pvi["value"] == {
+        "c": {"w": "DEVICE:Child:C"},
+        "d": {"x": "DEVICE:Child:D"},
+    }
+    pass

tests/conftest.py

@@ -1,6 +1,12 @@
 import os
+import random
+import string
+import subprocess
+import time
+from pathlib import Path
 
 import pytest
+from aioca import purge_channel_caches
 
 from fastcs.attributes import AttrR, AttrRW, AttrW, Handler, Sender, Updater
 from fastcs.controller import Controller
@@ -72,3 +78,35 @@ def controller():
 @pytest.fixture
 def mapping(controller):
     return Mapping(controller)
+
+
+PV_PREFIX = "".join(random.choice(string.ascii_lowercase) for _ in range(12))
+HERE = Path(os.path.dirname(os.path.abspath(__file__)))
+
+
+@pytest.fixture(scope="module")
+def ioc():
+    process = subprocess.Popen(
+        ["python", HERE / "ioc.py"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        universal_newlines=True,
+    )
+
+    start_time = time.monotonic()
+    while "iocRun: All initialization complete" not in (
+        process.stdout.readline().strip()
+    ):
+        if time.monotonic() - start_time > 10:
+            raise TimeoutError("IOC did not start in time")
+
+    yield
+
+    # close backend caches before the event loop
+    purge_channel_caches()
+    try:
+        print(process.communicate("exit")[0])
+    except ValueError:
+        # Someone else already called communicate
+        pass

tests/ioc.py

@@ -0,0 +1,30 @@
+from fastcs.attributes import AttrR, AttrRW, AttrW
+from fastcs.backends.epics.backend import EpicsBackend
+from fastcs.controller import Controller, SubController
+from fastcs.datatypes import Int
+from fastcs.wrappers import command
+
+
+class ParentController(Controller):
+    a: AttrR = AttrR(Int())
+    b: AttrRW = AttrRW(Int())
+
+
+class ChildController(SubController):
+    c: AttrW = AttrW(Int())
+
+    @command()
+    async def d(self):
+        pass
+
+
+def run():
+    controller = ParentController()
+    controller.register_sub_controller("Child", ChildController())
+
+    backend = EpicsBackend(controller, "DEVICE")
+    backend.run()
+
+
+if __name__ == "__main__":
+    run()