Add helper (and tests) for boilerplate stuff in shell command handling

src/libpvarki/shell.py

@@ -0,0 +1,29 @@
+"""Shell related helpers"""
+from typing import Tuple
+import asyncio
+import logging
+
+LOGGER = logging.getLogger(__name__)
+
+
+async def call_cmd(cmd: str, timeout: float = 2.5) -> Tuple[int, str, str]:
+    """Do the boilerplate for calling cmd and returning the exit code and output as strings"""
+    LOGGER.debug("Calling create_subprocess_shell(({})".format(cmd))
+    process = await asyncio.create_subprocess_shell(
+        cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    out, err = await asyncio.wait_for(process.communicate(), timeout=timeout)
+    out_str = out.decode("utf-8")
+    err_str = err.decode("utf-8")
+    if err:
+        LOGGER.warning("{} stderr: {}".format(cmd, err_str))
+    LOGGER.info(out_str)
+    assert isinstance(process.returncode, int)  # at this point it is, keep mypy happy
+    if process.returncode != 0:
+        LOGGER.error("{} returned nonzero code: {} (process: {})".format(cmd, process.returncode, process))
+        LOGGER.error("{} stderr: {}".format(cmd, err_str))
+        LOGGER.error("{} stdout: {}".format(cmd, out_str))
+
+    return process.returncode, out_str, err_str

tests/test_shell.py

@@ -0,0 +1,52 @@
+"""Test shell helpers"""
+import logging
+import asyncio
+
+import pytest
+
+from libpvarki.shell import call_cmd
+
+LOGGER = logging.getLogger(__name__)
+
+
+@pytest.mark.asyncio
+async def test_true() -> None:
+    """Test that true exists with code 0"""
+    code, stdout, stderr = await call_cmd("true")
+    assert code == 0
+    assert stdout == ""
+    assert stderr == ""
+
+
+@pytest.mark.asyncio
+async def test_false() -> None:
+    """Test that false exists with nonzero code"""
+    code, stdout, stderr = await call_cmd("false")
+    assert code != 0
+    assert stdout == ""
+    assert stderr == ""
+
+
+@pytest.mark.asyncio
+async def test_stdout() -> None:
+    """Test that echo exists with code 0 and ouputs what we expect to stdout"""
+    code, stdout, stderr = await call_cmd("echo 'hello world'")
+    assert code == 0
+    assert stdout == "hello world\n"
+    assert stderr == ""
+
+
+@pytest.mark.asyncio
+async def test_stderr() -> None:
+    """Test that echo exists with code 0 and ouputs what we expect to stderr"""
+    code, stdout, stderr = await call_cmd("echo 'goodbye world' >&2")
+    assert code == 0
+    assert stdout == ""
+    assert stderr == "goodbye world\n"
+
+
+@pytest.mark.asyncio
+async def test_timeout() -> None:
+    """Test that long sleeps are aborted on timeout"""
+    with pytest.raises(asyncio.TimeoutError):
+        await call_cmd("sleep 5", timeout=1.0)