Merge pull request #16 from ZLLentz/fix-tests

TST: Fix tests segfault

Author: ZLLentz

test/test_psp.py

@@ -20,13 +20,9 @@ def setup_pv(pvname, connect=True):
     return pv
 
 
-def test_server_start(server):
-    pass
-
-
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_connect_and_disconnect(pvname):
+def test_connect_and_disconnect(server, pvname):
     logger.debug('test_create_and_clear_channel %s', pvname)
     pv = setup_pv(pvname)
     assert pv.isconnected
@@ -35,16 +31,17 @@ def test_connect_and_disconnect(pvname):
 
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_get(pvname):
+def test_get(server, pvname):
     logger.debug('test_get_data %s', pvname)
     pv = setup_pv(pvname)
     value = pv.get()
     assert value is not None
+    pv.disconnect()
 
 
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_put_get(pvname):
+def test_put_get(server, pvname):
     logger.debug('test_put_get %s', pvname)
     pv = setup_pv(pvname)
     old_value = pv.get()
@@ -59,11 +56,12 @@ def test_put_get(pvname):
     logger.debug('caput %s %s', pvname, new_value)
     pv.put(new_value, timeout=1.0)
     assert pv.get() == new_value
+    pv.disconnect()
 
 
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_monitor(pvname):
+def test_monitor(server, pvname):
     logger.debug('test_subscribe %s', pvname)
     pv = setup_pv(pvname)
     old_value = pv.get()
@@ -83,22 +81,24 @@ def test_monitor(pvname):
         time.sleep(0.1)
         n += 1
     assert pv.value == new_value
+    pv.disconnect()
 
 
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_misc(pvname):
+def test_misc(server, pvname):
     logger.debug('test_misc %s', pvname)
     pv = setup_pv(pvname)
     assert isinstance(pv.host(), str)
     assert isinstance(pv.state(), int)
     assert isinstance(pv.count, int)
     assert isinstance(pv.type(), str)
     assert isinstance(pv.rwaccess(), int)
+    pv.disconnect()
 
 
 @pytest.mark.timeout(10)
-def test_waveform():
+def test_waveform(server):
     logger.debug('test_waveform')
     pv = setup_pv(pvbase + ":WAVE")
     # Do as a tuple
@@ -111,17 +111,19 @@ def test_waveform():
     val = pv.get()
     assert isinstance(val, np.ndarray)
     assert len(val) == pv.count
+    pv.disconnect()
 
 
 @pytest.mark.timeout(10)
-def test_threads():
+def test_threads(server):
     logger.debug('test_threads')
 
     def some_thread_thing(pvname):
         psp.utils.ensure_context()
         pv = setup_pv(pvname)
         val = pv.get()
         assert isinstance(val, tuple)
+        pv.disconnect()
 
     pvname = pvbase + ":WAVE"
     thread = threading.Thread(target=some_thread_thing, args=(pvname,))

test/test_pyca.py

@@ -87,13 +87,9 @@ def setup_pv(pvname, connect=True):
     return pv
 
 
-def test_server_start(server):
-    pass
-
-
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_create_and_clear_channel(pvname):
+def test_create_and_clear_channel(server, pvname):
     logger.debug('test_create_and_clear_channel %s', pvname)
     pv = setup_pv(pvname)
     assert pv.connect_cb.connected
@@ -106,7 +102,7 @@ def test_create_and_clear_channel(pvname):
 
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_get_data(pvname):
+def test_get_data(server, pvname):
     logger.debug('test_get_data %s', pvname)
     pv = setup_pv(pvname)
     # get time vars
@@ -124,11 +120,12 @@ def test_get_data(pvname):
         assert key in pv.data
     # check that value is not None
     assert pv.data['value'] is not None
+    pv.clear_channel()
 
 
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_put_get(pvname):
+def test_put_get(server, pvname):
     logger.debug('test_put_get %s', pvname)
     pv = setup_pv(pvname)
     pv.get_data(False, -1.0)
@@ -151,11 +148,12 @@ def test_put_get(pvname):
     assert pv.getevt_cb.wait(timeout=1)
     recv_value = pv.data['value']
     assert recv_value == new_value
+    pv.clear_channel()
 
 
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_subscribe(pvname):
+def test_subscribe(server, pvname):
     logger.debug('test_subscribe %s', pvname)
     pv = setup_pv(pvname)
     ev = threading.Event()
@@ -186,22 +184,24 @@ def mon_cb(exception=None):
     assert ev.wait(timeout=1)
     recv_value = pv.data['value']
     assert recv_value == new_value
+    pv.clear_channel()
 
 
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize('pvname', test_pvs)
-def test_misc(pvname):
+def test_misc(server, pvname):
     logger.debug('test_misc %s', pvname)
     pv = setup_pv(pvname)
     assert isinstance(pv.host(), str)
     assert isinstance(pv.state(), int)
     assert isinstance(pv.count(), int)
     assert isinstance(pv.type(), str)
     assert isinstance(pv.rwaccess(), int)
+    pv.clear_channel()
 
 
 @pytest.mark.timeout(10)
-def test_waveform():
+def test_waveform(server):
     logger.debug('test_waveform')
     pv = setup_pv(pvbase + ":WAVE")
     # Do as a tuple
@@ -221,10 +221,11 @@ def test_waveform():
     val = pv.data['value']
     assert isinstance(val, np.ndarray)
     assert len(val) == pv.count()
+    pv.clear_channel()
 
 
 @pytest.mark.timeout(10)
-def test_threads():
+def test_threads(server):
     logger.debug('test_threads')
 
     def some_thread_thing(pvname):
@@ -234,6 +235,7 @@ def some_thread_thing(pvname):
         pyca.flush_io()
         assert pv.getevt_cb.wait(timeout=1)
         assert isinstance(pv.data['value'], tuple)
+        pv.clear_channel()
 
     pvname = pvbase + ":WAVE"
     thread = threading.Thread(target=some_thread_thing, args=(pvname,))