More tests

.github/workflows/code.yml

@@ -28,4 +28,4 @@ jobs:
         run: |
           python3 -m venv venv && source venv/bin/activate && pip install --upgrade pip
           pip install .[dev]
-          PMAC_FILTER_CONTROL=${GITHUB_WORKSPACE}/prefix/bin/pmacFilterControl pytest -v tests/test_pmac_filter_control.py
+          PMAC_FILTER_CONTROL=${GITHUB_WORKSPACE}/prefix/bin/pmacFilterControl pytest -v tests

setup.cfg

@@ -21,7 +21,7 @@ install_requires =
     importlib_metadata
     aioca
     h5py
-    softioc
+    softioc>=4.2.0
     aiozmq
     typer>=0.7.0  # Fix incompatibility with click>=8.1.0 | https://github.com/tiangolo/typer/issues/377
 
@@ -31,7 +31,12 @@ dev =
     black
     flake8
     mypy
+    coverage
+    mock
+    pytest>=7.1.1
+    pytest-mock
     pytest-asyncio
+    pytest-cov
     sphinx-autobuild
     sphinx-external-toc
     myst-parser
@@ -72,6 +77,20 @@ extend-ignore =
 # allow Annotated[typ, some_func("some string")]
     F722,
 
+[tool:pytest]
+# Run pytest with all our checkers, and don't spam us with massive tracebacks on error
+addopts =
+    --tb=native -vv --doctest-modules --doctest-glob="*.rst"
+     --cov=pmacfiltercontrol --cov-report term --cov-report xml:cov.xml
+asyncio_mode = strict
+
+[coverage:run]
+# This is covered in the versiongit test suite so exclude it here
+omit =
+    */_version.py
+    src/pmacfiltercontrol/detector_sim.py
+    src/pmacfiltercontrol/event_subscriber.py
+
 [coverage:paths]
 # Tests are run from installed location, map back to the src directory
 source =

tests/conftest.py

@@ -0,0 +1,140 @@
+import atexit
+import os
+import random
+import string
+import subprocess
+import sys
+from datetime import datetime
+
+import pytest
+from epicsdbbuilder import ResetRecords
+
+# Must import softioc before epicsdbbuilder
+from softioc.device_core import RecordLookup
+
+requires_cothread = pytest.mark.skipif(
+    sys.platform.startswith("win"), reason="Cothread doesn't work on windows"
+)
+
+# Default length used to initialise Waveform and longString records.
+# Length picked to match string record length, so we can re-use test strings.
+WAVEFORM_LENGTH = 40
+
+# Default timeout for many operations across testing
+TIMEOUT = 10  # Seconds
+
+# Address for multiprocessing Listener/Client pair
+ADDRESS = ("localhost", 2345)
+
+
+def create_random_prefix():
+    """Create 12-character random string, for generating unique Device Names"""
+    return "".join(random.choice(string.ascii_uppercase) for _ in range(12))
+
+
+# Can't use logging as it's not multiprocess safe, and
+# alteratives are overkill
+def log(*args):
+    print(datetime.now().strftime("%H:%M:%S"), *args)
+
+
+class SubprocessIOC:
+    def __init__(self, ioc_py):
+        self.pv_prefix = create_random_prefix()
+        sim_ioc = os.path.join(os.path.dirname(__file__), ioc_py)
+        cmd = [sys.executable, sim_ioc, self.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(timeout=TIMEOUT)
+            print(out.decode())
+            print(err.decode())
+
+
+def aioca_cleanup():
+    from aioca import _catools, purge_channel_caches
+
+    # 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 pfc_ioc():
+    ioc = SubprocessIOC("test_pfc_wrapper.py")
+    yield ioc
+    ioc.kill()
+    aioca_cleanup()
+
+
+# @pytest.fixture
+# def asyncio_ioc_override():
+#     ioc = SubprocessIOC("sim_asyncio_ioc_override.py")
+#     yield ioc
+#     ioc.kill()
+#     aioca_cleanup()
+
+
+def _clear_records():
+    # Remove any records created at epicsdbbuilder layer
+    ResetRecords()
+    # And at pythonSoftIoc level
+    # TODO: Remove this hack and use use whatever comes out of
+    # https://github.com/dls-controls/pythonSoftIOC/issues/56
+    RecordLookup._RecordDirectory.clear()
+
+
+@pytest.fixture(autouse=True)
+def clear_records():
+    """Deletes all records before and after every test"""
+    _clear_records()
+    yield
+    _clear_records()
+
+
+@pytest.fixture(autouse=True)
+def enable_code_coverage():
+    """Ensure code coverage works as expected for `multiprocesses` tests.
+    As its harmless for other types of test, we always run this fixture."""
+    try:
+        from pytest_cov.embed import cleanup_on_sigterm
+    except ImportError:
+        pass
+    else:
+        cleanup_on_sigterm()
+
+
+def select_and_recv(conn, expected_char=None):
+    """Wait for the given Connection to have data to receive, and return it.
+    If a character is provided check its correct before returning it."""
+    # Must use cothread's select if cothread is present, otherwise we'd block
+    # processing on all cothread processing. But we don't want to use it
+    # unless we have to, as importing cothread can cause issues with forking.
+    if "cothread" in sys.modules:
+        from cothread import select
+
+        rrdy, _, _ = select([conn], [], [], TIMEOUT)
+    else:
+        # Would use select.select(), but Windows doesn't accept Pipe handles
+        # as selectable objects.
+        if conn.poll(TIMEOUT):
+            rrdy = True
+
+    if rrdy:
+        val = conn.recv()
+    else:
+        pytest.fail("Did not receive expected char before TIMEOUT expired")
+
+    if expected_char:
+        assert val == expected_char, "Expected character did not match"
+
+    return val

tests/test_asyncio_pfc_wrapper.py

@@ -0,0 +1,50 @@
+import os
+from pathlib import Path
+
+# require('pygelf')
+# Import the basic framework components.
+from softioc import asyncio_dispatcher, builder, softioc
+
+from pmacfiltercontrol.pmacFilterControlWrapper import (
+    Wrapper as pmacFilterControlWrapper,
+)
+
+# A couple of identification PVs, assumes this file is the name of the IOC
+device_name = "pytest_pfcw"
+builder.SetDeviceName(device_name)
+builder.stringIn("WHOAMI", initial_value="Test Fast Attenutator Control")
+builder.stringIn("HOSTNAME", VAL=os.uname()[1])
+
+filter_set_total = 2
+filters_per_set = 2
+
+autosave_pos_file = f"{Path.cwd()}/tests/test_autosave.txt"
+
+dispatcher = asyncio_dispatcher.AsyncioDispatcher()
+
+wrapper = pmacFilterControlWrapper(
+    "127.0.0.1",
+    9000,
+    9001,
+    builder=builder,
+    device_name=device_name,
+    filter_set_total=filter_set_total,
+    filters_per_set=filters_per_set,
+    detector="BLXXI-EA-EXCBR-01",
+    motors="BLXXI-OP-FILT-01",
+    autosave_file_path=autosave_pos_file,
+    hdf_file_path=f"{Path.cwd()}/tests/",
+)
+
+# dispatcher(wrapper.run_forever)
+
+# setup_logging(default_level=logging.DEBUG)
+
+# Now get the IOC started
+builder.LoadDatabase()
+softioc.iocInit(dispatcher)
+
+# wrapper.set_device_info()
+
+# Leave the iocsh running
+softioc.interactive_ioc(globals())

tests/test_autosave.txt

@@ -0,0 +1,21 @@
+pytest_pfcw:FILTER_SET:1:IN:1 100.0
+pytest_pfcw:FILTER_SET:1:IN:2 100.0
+pytest_pfcw:FILTER_SET:2:IN:1 100.0
+pytest_pfcw:FILTER_SET:2:IN:2 100.0
+pytest_pfcw:FILTER_SET:1:OUT:1 0.0
+pytest_pfcw:FILTER_SET:1:OUT:2 0.0
+pytest_pfcw:FILTER_SET:2:OUT:1 0.0
+pytest_pfcw:FILTER_SET:2:OUT:2 0.0
+pytest_pfcw:SHUTTER:OPEN 0.0
+pytest_pfcw:SHUTTER:CLOSED 500.0
+pytest_pfcw:HIGH:THRESHOLD:EXTREME 10.0
+pytest_pfcw:HIGH:THRESHOLD:UPPER 5.0
+pytest_pfcw:HIGH:THRESHOLD:LOWER 2.0
+pytest_pfcw:LOW:THRESHOLD:UPPER 2.0
+pytest_pfcw:LOW:THRESHOLD:LOWER 5.0
+High3 10000.0
+High2 500.0
+High1 300.0
+Low1 500.0
+Low2 10000.0
+pytest_pfcw:FILTER_SET 1