- added utility function more or less equivalent to subprocess.run,

  but which works with asyncio, so that waiting on subprocesses doesn't
  block the event loop;

.github/workflows/tests.yml

@@ -27,7 +27,7 @@ jobs:
 
     - name: install dependencies
       run: |
-        python3 -m pip install pytest hypothesis
+        python3 -m pip install pytest pytest-asyncio hypothesis
         python3 -m pip install .
 
     - name: run tests

src/coverup/utils.py

@@ -1,5 +1,6 @@
 from pathlib import Path
 import typing as T
+import subprocess
 
 
 class TemporaryOverwrite:
@@ -66,3 +67,27 @@ def lines_branches_do(lines: T.Set[int], neg_lines: T.Set[int], branches: T.Set[
 
     s += " does" if len(lines)+len(relevant_branches) == 1 else " do"
     return s
+
+
+async def subprocess_run(args: str, check: bool = False, timeout: T.Optional[int] = None) -> subprocess.CompletedProcess:
+    """Provides an asynchronous version of subprocess.run"""
+    import asyncio
+
+    process = await asyncio.create_subprocess_exec(*args, stdout=asyncio.subprocess.PIPE,
+                                                   stderr=asyncio.subprocess.STDOUT)
+
+    try:
+        if timeout is not None:
+            output, _ = await asyncio.wait_for(process.communicate(), timeout=timeout)
+        else:
+            output, _ = await process.communicate()
+
+    except asyncio.TimeoutError:
+        process.terminate()
+        await process.wait()
+        raise subprocess.TimeoutExpired(args, timeout, output=process.stdout) from None
+
+    if check and process.returncode != 0:
+        raise subprocess.CalledProcessError(process.returncode, args, output=process.stdout)
+
+    return subprocess.CompletedProcess(args=args, returncode=process.returncode, stdout=output)

tests/test_utils.py

@@ -1,7 +1,8 @@
 import pytest
 from hypothesis import given, strategies as st
 from pathlib import Path
-import coverup.utils as utils
+from coverup import utils
+import subprocess
 
 def test_format_ranges():
     assert "" == utils.format_ranges(set(), set())
@@ -27,3 +28,28 @@ def test_lines_branches_do():
 
     # if a line doesn't execute, neither do the branches that touch it...
     assert "lines 123-125 do" == utils.lines_branches_do({123,124,125}, set(), {(123,124), (10,125)})
+
+
+@pytest.mark.parametrize('check', [False, True])
+@pytest.mark.asyncio
+async def test_subprocess_run(check):
+    p = await utils.subprocess_run(['/bin/echo', 'hi!'], check=check)
+    assert p.stdout == b"hi!\n"
+
+
+@pytest.mark.asyncio
+async def test_subprocess_run_fails_checked():
+    with pytest.raises(subprocess.CalledProcessError) as e:
+        await utils.subprocess_run(['/usr/bin/false'], check=True)
+
+
+@pytest.mark.asyncio
+async def test_subprocess_run_fails_not_checked():
+    p = await utils.subprocess_run(['/usr/bin/false'])
+    assert p.returncode != 0
+
+
+@pytest.mark.asyncio
+async def test_subprocess_run_timeout():
+    with pytest.raises(subprocess.TimeoutExpired) as e:
+        await utils.subprocess_run(['/bin/sleep', '2'], timeout=1)