test(refactor): clean up run_in_subprocess

numba_cuda/numba/cuda/tests/cudapy/test_warning.py

@@ -41,9 +41,7 @@ def kernel():
         performance_warning = "float16 relies on LTO for performance"
         expected_warning_count = 0 if driver._have_nvjitlink() else 1
         _, err = run_in_subprocess(fp16_kernel_invocation)
-        self.assertEqual(
-            err.decode().count(performance_warning), expected_warning_count
-        )
+        self.assertEqual(err.count(performance_warning), expected_warning_count)
 
     def test_inefficient_launch_configuration(self):
         @cuda.jit

numba_cuda/numba/cuda/tests/nocuda/test_import.py

@@ -51,12 +51,18 @@ def test_no_impl_import(self):
             "numba.cuda.np.npyimpl",
         )
 
-        code = "import sys; from numba import cuda; print(list(sys.modules))"
+        code = """\
+import sys
+
+from numba import cuda
+
+for mod in sys.modules:
+    print(mod)"""
 
         out, _ = run_in_subprocess(code)
-        modlist = set(eval(out.strip()))
+        modlist = out.splitlines()
         unexpected = set(banlist) & set(modlist)
-        self.assertFalse(unexpected, "some modules unexpectedly imported")
+        assert not unexpected
 
 
 if __name__ == "__main__":

numba_cuda/numba/cuda/tests/support.py

@@ -193,30 +193,28 @@ def override_config(name, value):
         setattr(config, name, old_value)
 
 
-def run_in_subprocess(code, flags=None, env=None, timeout=30):
+def run_in_subprocess(code, flags=(), env=None, timeout=30):
     """Run a snippet of Python code in a subprocess with flags, if any are
     given. 'env' is passed to subprocess.Popen(). 'timeout' is passed to
     popen.communicate().
 
     Returns the stdout and stderr of the subprocess after its termination.
     """
-    if flags is None:
-        flags = []
-    cmd = (
-        [
-            sys.executable,
-        ]
-        + flags
-        + ["-c", code]
-    )
-    popen = subprocess.Popen(
-        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env
-    )
-    out, err = popen.communicate(timeout=timeout)
-    if popen.returncode != 0:
-        msg = "process failed with code %s: stderr follows\n%s\n"
-        raise AssertionError(msg % (popen.returncode, err.decode()))
-    return out, err
+    try:
+        proc = subprocess.run(
+            [sys.executable, *flags, "-c", code],
+            env=env,
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+    except subprocess.CalledProcessError as e:
+        raise AssertionError(
+            f"process failed with code {e.returncode:d}: stderr:\n{e.stderr}\n"
+        ) from e
+    else:
+        return proc.stdout, proc.stderr
 
 
 def captured_stdout():