Move test fixtures around to avoid imports

Author: coretl

tests/conftest.py

@@ -1,22 +1,25 @@
+import atexit
 import os
 import random
 import string
 import subprocess
 import sys
 
+import pytest
+
 if sys.version_info < (3,):
     # Python2 has no asyncio, so ignore these tests
     collect_ignore = [
-        "test_asyncio.py", "sim_asyncio_ioc.py", "sim_asyncio_ioc_overide.py"
+        "test_asyncio.py", "sim_asyncio_ioc.py", "sim_asyncio_ioc_override.py"
     ]
 
-PV_PREFIX = "".join(random.choice(string.ascii_uppercase) for _ in range(12))
-
-
 class SubprocessIOC:
     def __init__(self, ioc_py):
+        self.pv_prefix = "".join(
+            random.choice(string.ascii_uppercase) for _ in range(12)
+        )
         sim_ioc = os.path.join(os.path.dirname(__file__), ioc_py)
-        cmd = [sys.executable, sim_ioc, PV_PREFIX]
+        cmd = [sys.executable, sim_ioc, self.pv_prefix]
         self.proc = subprocess.Popen(
             cmd, stdin=subprocess.PIPE,
             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
@@ -28,3 +31,37 @@ def kill(self):
             out, err = self.proc.communicate()
             print(out.decode())
             print(err.decode())
+
+
+@pytest.fixture
+def cothread_ioc():
+    ioc = SubprocessIOC("sim_cothread_ioc.py")
+    yield ioc
+    ioc.kill()
+
+
+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():
+    ioc = SubprocessIOC("sim_asyncio_ioc.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()

tests/test_asyncio.py

@@ -1,28 +1,7 @@
 # Will be ignored on Python2 by conftest.py settings
 
-import atexit
 import signal
 import pytest
-from tests.conftest import SubprocessIOC, PV_PREFIX
-
-
-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():
-    ioc = SubprocessIOC("sim_asyncio_ioc.py")
-    yield ioc.proc
-    ioc.kill()
-    aioca_cleanup()
 
 
 @pytest.mark.asyncio
@@ -31,73 +10,67 @@ async def test_asyncio_ioc(asyncio_ioc):
     from aioca import caget, caput, camonitor, CANothing, _catools, FORMAT_TIME
 
     # Start
-    assert (await caget(PV_PREFIX + ":UPTIME")).startswith("00:00:0")
+    pre = asyncio_ioc.pv_prefix
+    assert (await caget(pre + ":UPTIME")).startswith("00:00:0")
     # WAVEFORM
-    await caput(PV_PREFIX + ":SINN", 4, wait=True)
+    await caput(pre + ":SINN", 4, wait=True)
     q = asyncio.Queue()
-    m = camonitor(PV_PREFIX + ":SIN", q.put, notify_disconnect=True)
+    m = camonitor(pre + ":SIN", q.put, notify_disconnect=True)
     assert len(await asyncio.wait_for(q.get(), 1)) == 4
     # AO
-    ao_val = await caget(PV_PREFIX + ":ALARM", format=FORMAT_TIME)
+    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(PV_PREFIX + ":ALARM", 3, wait=True)
+    await caput(pre + ":ALARM", 3, wait=True)
     await asyncio.sleep(0.1)
-    ai_val = await caget(PV_PREFIX + ":AI", format=FORMAT_TIME)
+    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(PV_PREFIX + ":AI", format=FORMAT_TIME)
+    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(PV_PREFIX + ":AI") == 23.45
+        assert await ctx.get(pre + ":AI") == 23.45
     # Wait for a bit longer for the print output to flush
     await asyncio.sleep(2)
     # Stop
-    out, err = asyncio_ioc.communicate(b"exit\n", timeout=5)
+    out, err = asyncio_ioc.proc.communicate(b"exit\n", timeout=5)
     out = out.decode()
     err = err.decode()
     # Disconnect
     assert isinstance(await asyncio.wait_for(q.get(), 10), CANothing)
     m.close()
     # check closed and output
-    assert "%s:SINN.VAL 1024 -> 4" % PV_PREFIX in out
+    assert "%s:SINN.VAL 1024 -> 4" % pre in out
     assert 'update_sin_wf 4' in out
-    assert "%s:ALARM.VAL 0 -> 3" % PV_PREFIX in out
-    assert 'on_update %s:AO : 3.0' % PV_PREFIX 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
 
 
-@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
+    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
-    asyncio_ioc_override.send_signal(signal.SIGINT)
+    asyncio_ioc_override.proc.send_signal(signal.SIGINT)
     # check closed and output
-    out, err = asyncio_ioc_override.communicate()
+    out, err = asyncio_ioc_override.proc.communicate()
     out = out.decode()
     err = err.decode()
     # check closed and output

tests/test_cothread.py

@@ -1,54 +1,48 @@
 import sys
 import signal
-from tests.conftest import SubprocessIOC, PV_PREFIX
 import pytest
 
 
 if sys.platform.startswith("win"):
     pytest.skip("Cothread doesn't work on windows", allow_module_level=True)
 
 
-@pytest.fixture
-def cothread_ioc():
-    ioc = SubprocessIOC("sim_cothread_ioc.py")
-    yield ioc.proc
-    ioc.kill()
-
 
 def test_cothread_ioc(cothread_ioc):
     import cothread
     from cothread.catools import ca_nothing, caget, caput, camonitor
 
+    pre = cothread_ioc.pv_prefix
     # Start
-    assert caget(PV_PREFIX + ":UPTIME").startswith("00:00:0")
+    assert caget(pre + ":UPTIME").startswith("00:00:0")
     # WAVEFORM
-    caput(PV_PREFIX + ":SINN", 4, wait=True)
+    caput(pre + ":SINN", 4, wait=True)
     q = cothread.EventQueue()
-    m = camonitor(PV_PREFIX + ":SIN", q.Signal, notify_disconnect=True)
+    m = camonitor(pre + ":SIN", q.Signal, notify_disconnect=True)
     assert len(q.Wait(1)) == 4
     # STRINGOUT
-    assert caget(PV_PREFIX + ":STRINGOUT") == "watevah"
-    caput(PV_PREFIX + ":STRINGOUT", "something", wait=True)
-    assert caget(PV_PREFIX + ":STRINGOUT") == "something"
+    assert caget(pre + ":STRINGOUT") == "watevah"
+    caput(pre + ":STRINGOUT", "something", wait=True)
+    assert caget(pre + ":STRINGOUT") == "something"
     # Check pvaccess works
     from p4p.client.cothread import Context
     with Context("pva") as ctx:
-        assert ctx.get(PV_PREFIX + ":STRINGOUT") == "something"
+        assert ctx.get(pre + ":STRINGOUT") == "something"
     # Wait for a bit longer for the print output to flush
     cothread.Sleep(2)
     # Stop
-    cothread_ioc.send_signal(signal.SIGINT)
+    cothread_ioc.proc.send_signal(signal.SIGINT)
     # Disconnect
     assert isinstance(q.Wait(10), ca_nothing)
     m.close()
     # check closed and output
-    out, err = cothread_ioc.communicate()
+    out, err = cothread_ioc.proc.communicate()
     out = out.decode()
     err = err.decode()
     # check closed and output
-    assert "%s:SINN.VAL 1024 -> 4" % PV_PREFIX in out
+    assert "%s:SINN.VAL 1024 -> 4" % pre in out
     assert 'update_sin_wf 4' in out
-    assert "%s:STRINGOUT.VAL watevah -> something" % PV_PREFIX in out
+    assert "%s:STRINGOUT.VAL watevah -> something" % pre in out
     assert 'on_update \'something\'' in out
     assert 'Starting iocInit' in err
     assert 'iocRun: All initialization complete' in err