TST: disconnect and clear channels to avoid segfault in test cleanup

Author: ZLLentz

test/test_psp.py

@@ -40,6 +40,7 @@ def test_get(pvname):
     pv = setup_pv(pvname)
     value = pv.get()
     assert value is not None
+    pv.disconnect()
 
 
 @pytest.mark.timeout(10)
@@ -59,6 +60,7 @@ 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)
@@ -83,6 +85,7 @@ def test_monitor(pvname):
         time.sleep(0.1)
         n += 1
     assert pv.value == new_value
+    pv.disconnect()
 
 
 @pytest.mark.timeout(10)
@@ -95,6 +98,7 @@ def test_misc(pvname):
     assert isinstance(pv.count, int)
     assert isinstance(pv.type(), str)
     assert isinstance(pv.rwaccess(), int)
+    pv.disconnect()
 
 
 @pytest.mark.timeout(10)
@@ -111,6 +115,7 @@ def test_waveform():
     val = pv.get()
     assert isinstance(val, np.ndarray)
     assert len(val) == pv.count
+    pv.disconnect()
 
 
 @pytest.mark.timeout(10)
@@ -122,6 +127,7 @@ def some_thread_thing(pvname):
         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

@@ -124,6 +124,7 @@ 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)
@@ -151,6 +152,7 @@ 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)
@@ -186,6 +188,7 @@ 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)
@@ -198,6 +201,7 @@ def test_misc(pvname):
     assert isinstance(pv.count(), int)
     assert isinstance(pv.type(), str)
     assert isinstance(pv.rwaccess(), int)
+    pv.clear_channel()
 
 
 @pytest.mark.timeout(10)
@@ -221,6 +225,7 @@ def test_waveform():
     val = pv.data['value']
     assert isinstance(val, np.ndarray)
     assert len(val) == pv.count()
+    pv.clear_channel()
 
 
 @pytest.mark.timeout(10)
@@ -234,6 +239,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,))