Allow starting IOC without PVA

CHANGELOG.rst

@@ -16,6 +16,7 @@ Added:
 - `Add int64In/Out record support <../../pull/161>`_
 - `Enable setting alarm status of Out records <../../pull/157>`_
 - `Adding the non_interactive_ioc function <../../pull/156>`_
+- `Allow starting IOC without PVA <../../pull/186>`_
 
 Removed:
 

softioc/__init__.py

@@ -1,12 +1,8 @@
 '''Python soft IOC module.'''
 import os
-import ctypes
 
-from setuptools_dso.runtime import find_dso
 import epicscorelibs.path
-import pvxslibs.path
-from epicscorelibs.ioc import \
-    iocshRegisterCommon, registerRecordDeviceDriver, pdbbase
+from epicscorelibs.ioc import iocshRegisterCommon, pdbbase
 
 # Do this as early as possible, in case we happen to use cothread
 # This will set the CATOOLS_LIBCA_PATH environment variable in case we use
@@ -15,20 +11,16 @@
 
 # This import will also pull in the extension, which is needed
 # before we call iocshRegisterCommon
-from .imports import dbLoadDatabase
+from .imports import dbLoadDatabase, registerRecordDeviceDriver
 from ._version_git import __version__
 
 # Need to do this before calling anything in device.py
 iocshRegisterCommon()
 base_dbd_path = os.path.join(epicscorelibs.path.base_path, 'dbd')
 dbLoadDatabase('base.dbd', base_dbd_path, None)
-dbLoadDatabase('pvxsIoc.dbd', pvxslibs.path.dbd_path, None)
 iocStats = os.path.join(os.path.dirname(__file__), "iocStats", "devIocStats")
 dbLoadDatabase('devIocStats.dbd', iocStats, None)
 
-ctypes.CDLL(find_dso('pvxslibs.lib.pvxsIoc'), ctypes.RTLD_GLOBAL)
-os.environ.setdefault('PVXS_QSRV_ENABLE', 'YES')
-if registerRecordDeviceDriver(pdbbase):
-    raise RuntimeError('Error registering')
+registerRecordDeviceDriver(pdbbase)
 
 __all__ = ['__version__']

softioc/imports.py

@@ -105,6 +105,9 @@ def from_param(cls, value):
 epicsExitCallAtExits.argtypes = ()
 epicsExitCallAtExits.restype = None
 
+registerRecordDeviceDriver = dbCore.registerAllRecordDeviceDrivers
+registerRecordDeviceDriver.argtypes = (c_void_p,)
+registerRecordDeviceDriver.errcheck = expect_success
 
 __all__ = [
     'get_field_offsets',

softioc/softioc.py

@@ -1,9 +1,14 @@
+import ctypes
 import os
 import sys
 import atexit
 from ctypes import *
 from tempfile import NamedTemporaryFile
 
+import pvxslibs.path
+from epicscorelibs.ioc import pdbbase
+from setuptools_dso.runtime import find_dso
+
 from . import autosave, imports, device
 from . import cothread_dispatcher
 
@@ -16,14 +21,15 @@ def epicsAtPyExit():
     imports.epicsExitCallAtExits()
 
 
-def iocInit(dispatcher=None):
+def iocInit(dispatcher=None, enable_pva=True):
     '''This must be called exactly once after loading all EPICS database files.
     After this point the EPICS IOC is running and serving PVs.
 
     Args:
         dispatcher: A callable with signature ``dispatcher(func, *args)``. Will
-            be called in response to caput on a record. If not supplied use
-            `cothread` as a dispatcher.
+            be called in response to caput on a record. If not supplied uses
+            ``cothread`` as the dispatcher.
+        enable_pva: Specify whether to enable the PV Access Server in this IOC.
 
     See Also:
         `softioc.asyncio_dispatcher` is a dispatcher for `asyncio` applications
@@ -33,6 +39,13 @@ def iocInit(dispatcher=None):
         dispatcher = cothread_dispatcher.CothreadDispatcher()
     # Set the dispatcher for record processing callbacks
     device.dispatcher = dispatcher
+
+    if enable_pva:
+        dbLoadDatabase('pvxsIoc.dbd', pvxslibs.path.dbd_path, None)
+        ctypes.CDLL(find_dso('pvxslibs.lib.pvxsIoc'), ctypes.RTLD_GLOBAL)
+
+        imports.registerRecordDeviceDriver(pdbbase)
+
     imports.iocInit()
     autosave.start_autosave_thread()
 

tests/test_pvaccess.py

@@ -0,0 +1,95 @@
+import asyncio
+from contextlib import nullcontext
+
+import pytest
+
+from conftest import (
+    aioca_cleanup,
+    log,
+    create_random_prefix,
+    TIMEOUT,
+    select_and_recv,
+    get_multiprocessing_context,
+)
+
+from softioc import asyncio_dispatcher, builder, softioc
+
+
+class TestPVAccess:
+    """Tests related to PVAccess"""
+
+    record_name = "PVA_AOut"
+    record_value = 10
+
+    def pva_test_func(self, device_name, conn, use_pva):
+        builder.SetDeviceName(device_name)
+
+        builder.aOut(self.record_name, initial_value=self.record_value)
+
+        dispatcher = asyncio_dispatcher.AsyncioDispatcher()
+        builder.LoadDatabase()
+        softioc.iocInit(dispatcher=dispatcher, enable_pva=use_pva)
+
+        conn.send("R")  # "Ready"
+        log("CHILD: Sent R over Connection to Parent")
+
+        # Keep process alive while main thread works.
+        while (True):
+            if conn.poll(TIMEOUT):
+                val = conn.recv()
+                if val == "D":  # "Done"
+                    break
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "use_pva,expectation",
+        [
+            (True, nullcontext()),
+            (False, pytest.raises(asyncio.TimeoutError))
+        ]
+    )
+    async def test_pva_enable_disable(self, use_pva, expectation):
+        """Test that we can enable and disable PVA, perform PVAccess requests
+        when enabled, and that we can always do Channel Access requests"""
+        ctx = get_multiprocessing_context()
+        parent_conn, child_conn = ctx.Pipe()
+
+        device_name = create_random_prefix()
+
+        process = ctx.Process(
+            target=self.pva_test_func,
+            args=(device_name, child_conn, use_pva),
+        )
+
+        process.start()
+
+        from aioca import caget
+        from p4p.client.asyncio import Context
+        try:
+            # Wait for message that IOC has started
+            select_and_recv(parent_conn, "R")
+
+            record_full_name = device_name + ":" + self.record_name
+
+            ret_val = await caget(record_full_name, timeout=TIMEOUT)
+
+            assert ret_val == self.record_value
+
+            with expectation as _:
+                with Context("pva") as ctx:
+                    # Short timeout as, if the above CA connection has happened
+                    # there's no need to wait a very long time for the PVA
+                    # connection
+                    pva_val = await asyncio.wait_for(
+                        ctx.get(record_full_name),
+                        timeout=2
+                    )
+                    assert pva_val == self.record_value
+
+
+        finally:
+            # Clear the cache before stopping the IOC stops
+            # "channel disconnected" error messages
+            aioca_cleanup()
+            parent_conn.send("D")  # "Done"
+            process.join(timeout=TIMEOUT)