Add coverage to Multiprocess tests

Author: AlexanderWells-diamond

tests/conftest.py

@@ -82,3 +82,14 @@ def clear_records():
     _clear_records()
     yield
     _clear_records()
+
+def enable_code_coverage():
+    """Ensure code coverage works as expected for `multiprocesses` tests.
+    This method should be called just before any `multiprocess.Process` call
+    is made."""
+    try:
+        from pytest_cov.embed import cleanup_on_sigterm
+    except ImportError:
+        pass
+    else:
+        cleanup_on_sigterm()

tests/test_record_values.py

@@ -7,7 +7,7 @@
 from enum import Enum
 from math import isnan, inf, nan
 
-from conftest import requires_cothread
+from conftest import requires_cothread, enable_code_coverage
 
 from softioc import asyncio_dispatcher, builder, softioc
 from softioc.pythonSoftIoc import RecordWrapper
@@ -368,6 +368,8 @@ def run_test_function(
 
     parent_conn, child_conn = multiprocessing.Pipe()
 
+    enable_code_coverage()
+
     ioc_process = multiprocessing.Process(
         target=run_ioc,
         args=(record_configurations, child_conn, set_enum, get_enum),
@@ -776,6 +778,9 @@ def test_value_none_rejected_set_after_init(self, record_func_reject_none):
         """Test that setting \"None\" using .set() after IOC init raises an
         exception"""
         queue = multiprocessing.Queue()
+
+        enable_code_coverage()
+
         process = multiprocessing.Process(
             target=self.none_value_test_func,
             args=(record_func_reject_none, queue),

tests/test_records.py

@@ -4,7 +4,7 @@
 import pytest
 import asyncio
 
-from conftest import requires_cothread, _clear_records
+from conftest import requires_cothread, _clear_records, enable_code_coverage
 
 from softioc import asyncio_dispatcher, builder, softioc
 
@@ -181,6 +181,9 @@ def validate_test_runner(
             validate_pass: bool):
 
         queue = multiprocessing.Queue()
+
+        enable_code_coverage()
+
         process = multiprocessing.Process(
             target=self.validate_ioc_test_func,
             args=(creation_func, queue, validate_pass),