Add Listener/Client synchronization to tests

These tests, especially the asyncio one, were prone to timing issues
where record processing and prints to stdout could result in race
conditions. This is alleviated by using signalling between parent and
child processes to ensure synchronization.

Author: AlexanderWells-diamond

tests/conftest.py

@@ -20,7 +20,10 @@
 WAVEFORM_LENGTH = 40
 
 # Default timeout for many operations across testing
-TIMEOUT = 10  # Seconds
+TIMEOUT = 5  # Seconds
+
+# Address for multiprocessing Listener/Client pair
+ADDRESS = ("localhost", 2345)
 
 def create_random_prefix():
     """Create 12-character random string, for generating unique Device Names"""
@@ -103,13 +106,22 @@ def enable_code_coverage():
     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.
-    This function imports Cothread, and so must NOT be called before any
-    multiprocessing sub-processes are spawned."""
-    from cothread import select
-    rrdy, _, _ = select([conn], [], [], TIMEOUT)
+    If a character is provided check its correct before returning it."""
+    # Must use cothread's select if cothread is prsent, 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:

tests/sim_asyncio_ioc.py

@@ -1,11 +1,12 @@
-from argparse import ArgumentParser
-
 import asyncio
-import time
 import sys
 
+from argparse import ArgumentParser
+from multiprocessing.connection import Client
+
 from softioc import alarm, softioc, builder, asyncio_dispatcher, pvlog
 
+from conftest import ADDRESS, select_and_recv
 
 if __name__ == "__main__":
     # Being run as an IOC, so parse args and set prefix
@@ -16,28 +17,40 @@
 
     import sim_records
 
-    async def callback(value):
-        # Set the ao value, which will trigger on_update for it
-        sim_records.t_ao.set(value)
-        await asyncio.sleep(0.5)
-        print("async update %s (%s)" % (value, sim_records.t_ai.get()))
-        # Make sure it goes as epicsExit will not flush this for us
+    with Client(ADDRESS) as conn:
+
+        async def callback(value):
+            # Set the ao value, which will trigger on_update for it
+            sim_records.t_ao.set(value)
+            print("async update %s (%s)" % (value, sim_records.t_ai.get()))
+            # Make sure it goes as epicsExit will not flush this for us
+            sys.stdout.flush()
+            # Set the ai alarm, but keep the value
+            sim_records.t_ai.set_alarm(int(value) % 4, alarm.STATE_ALARM)
+            # Must give the t_ai record time to process
+            await asyncio.sleep(1)
+            conn.send("C")  # "Complete"
+
+        # Set a different initial value
+        sim_records.t_ai.set(23.45)
+
+        # Create a record to set the alarm
+        t_ao = builder.aOut('ALARM', on_update=callback)
+
+        # Run the IOC
+        builder.LoadDatabase()
+        softioc.iocInit(asyncio_dispatcher.AsyncioDispatcher())
+
+        conn.send("R")  # "Ready"
+
+        # Make sure coverage is written on epicsExit
+        from pytest_cov.embed import cleanup
+        sys._run_exitfuncs = cleanup
+
+        select_and_recv(conn, "D")  # "Done"
+        # Attempt to ensure all buffers flushed - C code (from `import pvlog`)
+        # may not be affected by these calls...
         sys.stdout.flush()
-        # Set the ai alarm, but keep the value
-        sim_records.t_ai.set_alarm(int(value) % 4, alarm.STATE_ALARM)
-
-    # Set a different initial value
-    sim_records.t_ai.set(23.45)
-
-    # Create a record to set the alarm
-    t_ao = builder.aOut('ALARM', on_update=callback)
-
-    # Run the IOC
-    builder.LoadDatabase()
-    softioc.iocInit(asyncio_dispatcher.AsyncioDispatcher())
-    # Wait for some prints to have happened
-    time.sleep(1)
-    # Make sure coverage is written on epicsExit
-    from pytest_cov.embed import cleanup
-    sys._run_exitfuncs = cleanup
-    softioc.interactive_ioc()
+        sys.stderr.flush()
+
+        conn.send("D")  # "Done"

tests/sim_asyncio_ioc_override.py

@@ -1,13 +1,16 @@
-from argparse import ArgumentParser
-
 import asyncio
 import os
 import re
+import sys
+
 from pathlib import Path
 from tempfile import NamedTemporaryFile
+from argparse import ArgumentParser
+from multiprocessing.connection import Client
 
 from softioc import softioc, builder, asyncio_dispatcher
 
+from conftest import ADDRESS, select_and_recv
 
 if __name__ == "__main__":
     # Being run as an IOC, so parse args and set prefix
@@ -34,4 +37,18 @@
     builder.LoadDatabase()
     event_loop = asyncio.get_event_loop()
     softioc.iocInit(asyncio_dispatcher.AsyncioDispatcher(event_loop))
-    event_loop.run_forever()
+
+    with Client(ADDRESS) as conn:
+        conn.send("R")  # "Ready"
+
+        # Make sure coverage is written on epicsExit
+        from pytest_cov.embed import cleanup
+        sys._run_exitfuncs = cleanup
+
+        select_and_recv(conn, "D")  # "Done"
+        # Attempt to ensure all buffers flushed - C code (from `import pvlog`)
+        # may not be affected by these calls...
+        sys.stdout.flush()
+        sys.stderr.flush()
+
+        conn.send("D")  # "Done"

tests/sim_cothread_ioc.py

@@ -1,7 +1,10 @@
 from argparse import ArgumentParser
+from multiprocessing.connection import Client
+import sys
 
 from softioc import softioc, builder, pvlog
 
+from conftest import ADDRESS, select_and_recv
 
 if __name__ == "__main__":
     import cothread
@@ -17,4 +20,14 @@
     # Run the IOC
     builder.LoadDatabase()
     softioc.iocInit()
-    cothread.WaitForQuit()
+
+    with Client(ADDRESS) as conn:
+        conn.send("R")  # "Ready"
+
+        select_and_recv(conn, "D")  # "Done"
+        # Attempt to ensure all buffers flushed - C code (from `import pvlog`)
+        # may not be affected by these calls...
+        sys.stdout.flush()
+        sys.stderr.flush()
+
+        conn.send("D")  # "Ready"

tests/test_asyncio.py

@@ -1,44 +1,56 @@
-# Will be ignored on Python2 by conftest.py settings
-
 import pytest
 import sys
 
+from multiprocessing.connection import Listener
+
+from conftest import requires_cothread, ADDRESS, select_and_recv
 
 @pytest.mark.asyncio
 async def test_asyncio_ioc(asyncio_ioc):
     import asyncio
     from aioca import caget, caput, camonitor, CANothing, FORMAT_TIME
 
-    # Start
     pre = asyncio_ioc.pv_prefix
-    assert (await caget(pre + ":UPTIME")).startswith("00:00:0")
-    # WAVEFORM
-    await caput(pre + ":SINN", 4, wait=True)
-    q = asyncio.Queue()
-    m = camonitor(pre + ":SIN", q.put, notify_disconnect=True)
-    assert len(await asyncio.wait_for(q.get(), 1)) == 4
-    # AO
-    ao_val = await caget(pre + ":ALARM", format=FORMAT_TIME)
-    assert ao_val == 0
-    assert ao_val.severity == 3  # INVALID
-    assert ao_val.status == 17  # UDF
-    await caput(pre + ":ALARM", 3, wait=True)
-    await asyncio.sleep(0.1)
-    ai_val = await caget(pre + ":AI", format=FORMAT_TIME)
-    assert ai_val == 23.45
-    assert ai_val.severity == 0
-    assert ai_val.status == 0
-    await asyncio.sleep(0.8)
-    ai_val = await caget(pre + ":AI", format=FORMAT_TIME)
-    assert ai_val == 23.45
-    assert ai_val.severity == 3
-    assert ai_val.status == 7  # STATE_ALARM
-    # Check pvaccess works
-    from p4p.client.asyncio import Context
-    with Context("pva") as ctx:
-        assert await ctx.get(pre + ":AI") == 23.45
-    # Wait for a bit longer for the print output to flush
-    await asyncio.sleep(2)
+
+    with Listener(ADDRESS) as listener, listener.accept() as conn:
+
+        select_and_recv(conn, "R")  # "Ready"
+
+        # Start
+        assert (await caget(pre + ":UPTIME")).startswith("00:00:0")
+        # WAVEFORM
+        await caput(pre + ":SINN", 4, wait=True)
+        q = asyncio.Queue()
+        m = camonitor(pre + ":SIN", q.put, notify_disconnect=True)
+        assert len(await asyncio.wait_for(q.get(), 1)) == 4
+        # AO
+        ao_val = await caget(pre + ":ALARM", format=FORMAT_TIME)
+        assert ao_val == 0
+        assert ao_val.severity == 3  # INVALID
+        assert ao_val.status == 17  # UDF
+
+        ai_val = await caget(pre + ":AI", format=FORMAT_TIME)
+        assert ai_val == 23.45
+        assert ai_val.severity == 0
+        assert ai_val.status == 0
+
+        await caput(pre + ":ALARM", 3, wait=True)
+
+        # Confirm the ALARM callback has completed
+        select_and_recv(conn, "C")  # "Complete"
+
+        ai_val = await caget(pre + ":AI", format=FORMAT_TIME)
+        assert ai_val == 23.45
+        assert ai_val.severity == 3
+        assert ai_val.status == 7  # STATE_ALARM
+        # Check pvaccess works
+        from p4p.client.asyncio import Context
+        with Context("pva") as ctx:
+            assert await ctx.get(pre + ":AI") == 23.45
+
+        conn.send("D")  # "Done"
+        select_and_recv(conn, "D")  # "Done"
+
     # Stop
     out, err = asyncio_ioc.proc.communicate(b"exit\n", timeout=5)
     out = out.decode()
@@ -47,14 +59,19 @@ async def test_asyncio_ioc(asyncio_ioc):
     assert isinstance(await asyncio.wait_for(q.get(), 10), CANothing)
     m.close()
     # check closed and output
-    assert "%s:SINN.VAL 1024 -> 4" % pre in out
-    assert 'update_sin_wf 4' in out
-    assert "%s:ALARM.VAL 0 -> 3" % pre in out
-    assert 'on_update %s:AO : 3.0' % pre in out
-    assert 'async update 3.0 (23.45)' in out
-    assert 'Starting iocInit' in err
-    assert 'iocRun: All initialization complete' in err
-    assert '(InteractiveConsole)' in err
+    try:
+        assert "%s:SINN.VAL 1024 -> 4" % pre in out
+        assert 'update_sin_wf 4' in out
+        assert "%s:ALARM.VAL 0 -> 3" % pre in out
+        assert 'on_update %s:AO : 3.0' % pre in out
+        assert 'async update 3.0 (23.45)' in out
+        assert 'Starting iocInit' in err
+        assert 'iocRun: All initialization complete' in err
+    except Exception:
+        # Useful printing for when things go wrong!
+        print("Out:", out)
+        print("Err:", err)
+        raise
 
 
 @pytest.mark.asyncio
@@ -64,20 +81,34 @@ async def test_asyncio_ioc(asyncio_ioc):
 async def test_asyncio_ioc_override(asyncio_ioc_override):
     from aioca import caget, caput
 
-    # Gain bo
-    pre = asyncio_ioc_override.pv_prefix
-    assert (await caget(pre + ":GAIN")) == 0
-    await caput(pre + ":GAIN", "On", wait=True)
-    assert (await caget(pre + ":GAIN")) == 1
+    with Listener(ADDRESS) as listener, listener.accept() as conn:
+
+        select_and_recv(conn, "R")  # "Ready"
+
+        # Gain bo
+        pre = asyncio_ioc_override.pv_prefix
+        assert (await caget(pre + ":GAIN")) == 0
+        await caput(pre + ":GAIN", "On", wait=True)
+        assert (await caget(pre + ":GAIN")) == 1
+
+        # Stop
+        await caput(pre + ":SYSRESET", 1)
+
+        conn.send("D")  # "Done"
+        select_and_recv(conn, "D")  # "Done"
 
-    # Stop
-    await caput(pre + ":SYSRESET", 1)
     # check closed and output
     out, err = asyncio_ioc_override.proc.communicate(timeout=5)
     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
-    assert 'IOC reboot started' in err
+    try:
+        assert '1' in out
+        assert 'Starting iocInit' in err
+        assert 'iocRun: All initialization complete' in err
+        assert 'IOC reboot started' in err
+    except Exception:
+        # Useful printing for when things go wrong!
+        print("Out:", out)
+        print("Err:", err)
+        raise