TST: Implement test IOC

conda-recipe/meta.yaml

@@ -33,7 +33,9 @@ test:
   imports:
     - {{ import_name }}
   requires:
+    - caproto
     - coverage
+    - numpy
     - pytest
     - pytest-asyncio
     - pytest-qt

dev-requirements.txt

@@ -1,6 +1,8 @@
 # These are required for developing the package (running the tests) but not
 # necessarily required for _using_ it.
+caproto
 coverage
+numpy
 pytest
 pytest-asyncio
 pytest-cov

docs/source/upcoming_release_notes/34-ioc_infrastructure.rst

@@ -0,0 +1,23 @@
+34 ioc infrastructure
+#################
+
+API Breaks
+----------
+- N/A
+
+Features
+--------
+- implement fixture for running IOCs that can be queried for integration tests
+- implement module that can run IOCs for demos
+
+Bugfixes
+--------
+- don't change control_layer/core.py:SHIMS when creating dummy CLs
+
+Maintenance
+-----------
+- define linac testing structure outside of fixture
+
+Contributors
+------------
+- shilorigins

superscore/control_layers/core.py

@@ -32,7 +32,7 @@ class ControlLayer:
     def __init__(self, *args, shims: Optional[List[str]] = None, **kwargs):
         if shims is None:
             # load all available shims
-            self.shims = SHIMS
+            self.shims = SHIMS.copy()
             logger.debug('No shims specified, loading all available communication '
                          f'shims: {list(self.shims.keys())}')
         else:

superscore/tests/conftest.py

@@ -13,10 +13,10 @@
 from superscore.control_layers.core import ControlLayer
 from superscore.model import (Collection, Parameter, Readback, Root, Setpoint,
                               Snapshot)
+from superscore.tests.ioc import IOCFactory
 
 
-@pytest.fixture(scope='function')
-def linac_backend():
+def linac_data():
     lasr_gunb_pv1 = Parameter(
         uuid="5544c58f-88b6-40aa-9076-f180a44908f5",
         pv_name="LASR:GUNB:TEST1",
@@ -615,6 +615,12 @@ def linac_backend():
         ]
     )
 
+    return all_col, all_snapshot
+
+
+@pytest.fixture(scope='function')
+def linac_backend():
+    all_col, all_snapshot = linac_data()
     return TestBackend([all_col, all_snapshot])
 
 
@@ -736,8 +742,7 @@ def monitor(self, *args, **kwargs):
 @pytest.fixture(scope='function')
 def dummy_cl() -> ControlLayer:
     cl = ControlLayer()
-    cl.shims['ca'] = DummyShim()
-    cl.shims['pva'] = DummyShim()
+    cl.shims = {protocol: DummyShim() for protocol in ['ca', 'pva']}
     return cl
 
 
@@ -772,3 +777,10 @@ def sample_client(
     client.cl = dummy_cl
 
     return client
+
+
+@pytest.fixture(scope='module')
+def linac_ioc():
+    _, snapshot = linac_data()
+    with IOCFactory.from_entries(snapshot.children)(prefix="SCORETEST:") as ioc:
+        yield ioc

superscore/tests/ioc/__init__.py

@@ -0,0 +1 @@
+from .ioc_factory import IOCFactory  # noqa: F401

superscore/tests/ioc/ioc_factory.py

@@ -0,0 +1,71 @@
+from multiprocessing import Process
+from typing import Iterable, Mapping, Union
+
+from caproto.server import PVGroup, pvproperty
+from caproto.server import run as run_ioc
+from epicscorelibs.ca import dbr
+
+from superscore.model import Entry, Nestable, Parameter, Readback, Setpoint
+
+
+class TempIOC(PVGroup):
+    """
+    Makes PVs accessible via EPICS when running. Instances automatically start
+    and stop running when used as a context manager, and are thus suitable for
+    use in tests.
+    """
+    def __enter__(self):
+        self.running_process = Process(
+            target=run_ioc,
+            args=(self.pvdb,),
+            daemon=True,
+        )
+        self.running_process.start()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        pass
+
+
+class IOCFactory:
+    """
+    Generates TempIOC subclasses bound to a set of PVs.
+    """
+    @staticmethod
+    def from_entries(entries: Iterable[Entry], **ioc_options) -> PVGroup:
+        """
+        Defines and instantiates a TempIOC subclass containing all PVs reachable
+        from entries.
+        """
+        attrs = IOCFactory.prepare_attrs(entries)
+        IOC = type("IOC", (TempIOC,), attrs)
+        return IOC
+
+    @staticmethod
+    def collect_pvs(entries: Iterable[Entry]) -> Iterable[Union[Parameter, Setpoint, Readback]]:
+        """Returns a collection of all PVs reachable from entries"""
+        pvs = []
+        q = entries.copy()
+        while len(q) > 0:
+            entry = q.pop()
+            if isinstance(entry, Nestable):
+                q.extend(entry.children)
+            else:
+                pvs.append(entry)
+        return pvs
+
+    @staticmethod
+    def prepare_attrs(entries: Iterable[Entry]) -> Mapping[str, pvproperty]:
+        """
+        Turns a collecton of PVs into a Mapping from attribute names to
+        caproto.pvproperties. The mapping is suitable for passing into a type()
+        call as the dict arg.
+        """
+        pvs = IOCFactory.collect_pvs(entries)
+        attrs = {}
+        for entry in pvs:
+            value = entry.data if isinstance(entry, (Setpoint, Readback)) else None
+            pv = pvproperty(name=entry.pv_name, doc=entry.description, value=value, dtype=dbr.DBR_STRING if isinstance(entry.data, str) else None)
+            attr = "".join([c.lower() for c in entry.pv_name if c != ':'])
+            attrs[attr] = pv
+        return attrs

superscore/tests/ioc/linac.py

@@ -0,0 +1,15 @@
+from caproto.server import ioc_arg_parser, run
+
+from superscore.tests.conftest import linac_data
+from superscore.tests.ioc import IOCFactory
+
+if __name__ == '__main__':
+    _, snapshot = linac_data()
+    LinacIOC = IOCFactory.from_entries(snapshot.children)
+
+    ioc_options, run_options = ioc_arg_parser(
+        default_prefix='SCORETEST:',
+        desc="IOC evoking the structure of a linac",
+    )
+    ioc = LinacIOC(**ioc_options)
+    run(ioc.pvdb, **run_options)

superscore/tests/test_ioc.py

@@ -0,0 +1,20 @@
+from superscore.control_layers.core import ControlLayer
+
+
+def test_ioc(linac_ioc):
+    cl = ControlLayer()
+    assert cl.get("SCORETEST:MGNT:GUNB:TEST0").data == 1
+    cl.put("SCORETEST:MGNT:GUNB:TEST0", 0)
+    assert cl.get("SCORETEST:MGNT:GUNB:TEST0").data == 0
+
+    assert cl.get("SCORETEST:VAC:GUNB:TEST1").data == "Ion Pump"
+    cl.put("SCORETEST:VAC:GUNB:TEST1", "new value")
+    assert cl.get("SCORETEST:VAC:GUNB:TEST1").data == "new value"
+
+    assert cl.get("SCORETEST:LASR:GUNB:TEST2").data == 5
+    cl.put("SCORETEST:LASR:GUNB:TEST2", 10)
+    assert cl.get("SCORETEST:LASR:GUNB:TEST2").data == 10
+
+    assert cl.get("SCORETEST:LASR:IN10:TEST0").data == 645.26
+    cl.put("SCORETEST:LASR:IN10:TEST0", 600.0)
+    assert cl.get("SCORETEST:LASR:IN10:TEST0").data == 600.0