Add ability to add arbitrary lines to Db file and test

Author: coretl

softioc/builder.py

@@ -1,16 +1,20 @@
 import os
 import numpy
-from .softioc import dbLoadDatabase
 
 from epicsdbbuilder import *
+from epicsdbbuilder.recordset import recordset
 
 InitialiseDbd()
 LoadDbdFile(os.path.join(os.path.dirname(__file__), 'device.dbd'))
 
-from . import pythonSoftIoc  # noqa
+from . import pythonSoftIoc, imports  # noqa
 PythonDevice = pythonSoftIoc.PythonDevice()
 
 
+def dbLoadDatabase(database, path = None, substitutions = None):
+    '''Loads a database file and applies any given substitutions.'''
+    imports.dbLoadDatabase(database, path, substitutions)
+
 
 # ----------------------------------------------------------------------------
 #  Wrappers for PythonDevice record constructors.
@@ -181,6 +185,14 @@ def WaveformOut(name, *value, **fields):
 
 _DatabaseWritten = False
 
+def AddDatabaseLine(line, macros={}):
+    '''Add a single line to the produced database, substituting $(k) for v for
+    each entry in macros'''
+    for k, v in macros.items():
+        line = line.replace("$(%s)" % k, v)
+    recordset.AddBodyLine(line)
+
+
 def LoadDatabase():
     '''This should be called after all the builder records have been created,
     but before calling iocInit().  The database is loaded into EPICS memory,

softioc/softioc.py

@@ -4,7 +4,7 @@
 
 from epicsdbbuilder.recordset import recordset
 
-from . import imports, device
+from . import imports, device, builder
 
 __all__ = ['dbLoadDatabase', 'iocInit', 'interactive_ioc']
 
@@ -240,10 +240,9 @@ def __call__(self):
 exit = Exiter()
 command_names.append('exit')
 
+# For backwards compatibility
+dbLoadDatabase = builder.dbLoadDatabase
 
-def dbLoadDatabase(database, path = None, substitutions = None):
-    '''Loads a database file and applies any given substitutions.'''
-    imports.dbLoadDatabase(database, path, substitutions)
 
 def _add_records_from_file(dir, file, macros):
     # This is very naive, for instance macros are added to but never removed,
@@ -261,9 +260,7 @@ def _add_records_from_file(dir, file, macros):
                 _add_records_from_file(dir, line.split('"')[1], macros)
             else:
                 # A record line
-                for k, v in macros.items():
-                    line = line.replace('$(%s)' % k, v)
-                recordset.AddBodyLine(line)
+                builder.AddDatabaseLine(line, macros)
 
 
 def devIocStats(ioc_name):

tests/conftest.py

@@ -1,5 +1,30 @@
+import os
+import random
+import string
+import subprocess
 import sys
 
 if sys.version_info < (3,):
     # Python2 has no asyncio, so ignore these tests
-    collect_ignore = ["test_asyncio.py", "sim_asyncio_ioc.py"]
+    collect_ignore = [
+        "test_asyncio.py", "sim_asyncio_ioc.py", "sim_asyncio_ioc_overide.py"
+    ]
+
+PV_PREFIX = "".join(random.choice(string.ascii_uppercase) for _ in range(12))
+
+
+class SubprocessIOC:
+    def __init__(self, ioc_py):
+        sim_ioc = os.path.join(os.path.dirname(__file__), ioc_py)
+        cmd = [sys.executable, sim_ioc, PV_PREFIX]
+        self.proc = subprocess.Popen(
+            cmd, stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+
+    def kill(self):
+        if self.proc.returncode is None:
+            # still running, kill it and print the output
+            self.proc.kill()
+            out, err = self.proc.communicate()
+            print(out.decode())
+            print(err.decode())

tests/hw_records.db

@@ -0,0 +1,8 @@
+record(bo, "$(device):GAIN") {
+  field(DTYP, "Hy8001")
+  field(OMSL, "supervisory")
+  field(OUT, "#C1 S0 @")
+  field(DESC, "Gain bit 1")
+  field(ZNAM, "Off")
+  field(ONAM, "On")
+}

tests/sim_asyncio_ioc_override.py

@@ -0,0 +1,32 @@
+from argparse import ArgumentParser
+
+import asyncio
+import os
+import re
+from pathlib import Path
+
+from softioc import softioc, builder, asyncio_dispatcher
+
+
+if __name__ == "__main__":
+    # Being run as an IOC, so parse args and set prefix
+    parser = ArgumentParser()
+    parser.add_argument('prefix', help="The PV prefix for the records")
+    parsed_args = parser.parse_args()
+    builder.SetDeviceName(parsed_args.prefix)
+
+    # Load the base records without DTYP fields
+    macros = dict(device=parsed_args.prefix)
+    with open(Path(__file__).parent / "hw_records.db") as f:
+        for line in f.readlines():
+            if not re.match(r"\s*field\s*\(\s*DTYP", line):
+                builder.AddDatabaseLine(line, macros)
+
+    # Override DTYPE and OUT, and provide a callback
+    gain = builder.boolOut("GAIN", on_update=print)
+
+    # Run the IOC
+    builder.LoadDatabase()
+    event_loop = asyncio.get_event_loop()
+    softioc.iocInit(asyncio_dispatcher.AsyncioDispatcher(event_loop))
+    event_loop.run_forever()

tests/test_asyncio.py

@@ -1,45 +1,34 @@
 # Will be ignored on Python2 by conftest.py settings
 
-import random
-import string
-import subprocess
-import sys
-import os
 import atexit
+import signal
 import pytest
-import time
+from tests.conftest import SubprocessIOC, PV_PREFIX
 
-PV_PREFIX = "".join(random.choice(string.ascii_uppercase) for _ in range(12))
+
+def aioca_cleanup():
+    from aioca import purge_channel_caches, _catools
+    # Unregister the aioca atexit handler as it conflicts with the one installed
+    # by cothread. If we don't do this we get a seg fault. This is not a problem
+    # in production as we won't mix aioca and cothread, but we do mix them in
+    # the tests so need to do this.
+    atexit.unregister(_catools._catools_atexit)
+    # purge the channels before the event loop goes
+    purge_channel_caches()
 
 
 @pytest.fixture
 def asyncio_ioc():
-    sim_ioc = os.path.join(os.path.dirname(__file__), "sim_asyncio_ioc.py")
-    cmd = [sys.executable, sim_ioc, PV_PREFIX]
-    proc = subprocess.Popen(
-        cmd, stdin=subprocess.PIPE,
-        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    yield proc
-    # purge the channels before the event loop goes
-    from aioca import purge_channel_caches
-    purge_channel_caches()
-    if proc.returncode is None:
-        # still running, kill it and print the output
-        proc.kill()
-        out, err = proc.communicate()
-        print(out.decode())
-        print(err.decode(), file=sys.stderr)
+    ioc = SubprocessIOC("sim_asyncio_ioc.py")
+    yield ioc.proc
+    ioc.kill()
+    aioca_cleanup()
 
 
 @pytest.mark.asyncio
 async def test_asyncio_ioc(asyncio_ioc):
     import asyncio
     from aioca import caget, caput, camonitor, CANothing, _catools, FORMAT_TIME
-    # Unregister the aioca atexit handler as it conflicts with the one installed
-    # by cothread. If we don't do this we get a seg fault. This is not a problem
-    # in production as we won't mix aioca and cothread, but we do mix them in
-    # the tests so need to do this.
-    atexit.unregister(_catools._catools_atexit)
 
     # Start
     assert (await caget(PV_PREFIX + ":UPTIME")).startswith("00:00:0")
@@ -86,3 +75,32 @@ async def test_asyncio_ioc(asyncio_ioc):
     assert 'Starting iocInit' in err
     assert 'iocRun: All initialization complete' in err
     assert '(InteractiveConsole)' in err
+
+
+@pytest.fixture
+def asyncio_ioc_override():
+    ioc = SubprocessIOC("sim_asyncio_ioc_override.py")
+    yield ioc.proc
+    ioc.kill()
+    aioca_cleanup()
+
+
+@pytest.mark.asyncio
+async def test_asyncio_ioc_override(asyncio_ioc_override):
+    from aioca import caget, caput
+
+    # Gain bo
+    assert (await caget(PV_PREFIX + ":GAIN")) == 0
+    await caput(PV_PREFIX + ":GAIN", "On", wait=True)
+    assert (await caget(PV_PREFIX + ":GAIN")) == 1
+
+    # Stop
+    asyncio_ioc_override.send_signal(signal.SIGINT)
+    # check closed and output
+    out, err = asyncio_ioc_override.communicate()
+    out = out.decode()
+    err = err.decode()
+    # check closed and output
+    assert '1' in out
+    assert 'Starting iocInit' in err
+    assert 'iocRun: All initialization complete' in err

tests/test_cothread.py

@@ -1,37 +1,21 @@
-import random
-import string
-import subprocess
 import sys
-import os
 import signal
+from tests.conftest import SubprocessIOC, PV_PREFIX
 import pytest
 
-PV_PREFIX = "".join(random.choice(string.ascii_uppercase) for _ in range(12))
-
 
 if sys.platform.startswith("win"):
     pytest.skip("Cothread doesn't work on windows", allow_module_level=True)
 
 
 @pytest.fixture
 def cothread_ioc():
-    sim_ioc = os.path.join(os.path.dirname(__file__), "sim_cothread_ioc.py")
-    cmd = [sys.executable, sim_ioc, PV_PREFIX]
-    proc = subprocess.Popen(
-        cmd, stdin=subprocess.PIPE,
-        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    yield proc
-    if proc.returncode is None:
-        # still running, kill it and print the output
-        proc.kill()
-        out, err = proc.communicate()
-        print(out.decode())
-        print(err.decode())
-
+    ioc = SubprocessIOC("sim_cothread_ioc.py")
+    yield ioc.proc
+    ioc.kill()
 
 
 def test_cothread_ioc(cothread_ioc):
-    import epicscorelibs.path.cothread
     import cothread
     from cothread.catools import ca_nothing, caget, caput, camonitor