Add shell helper

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 1.8.0
+current_version = 1.9.0
 commit = False
 tag = False
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "libpvarki"
-version = "1.8.0"
+version = "1.9.0"
 description = "Common helpers like standard logging init"
 authors = ["Eero af Heurlin <eero.afheurlin@iki.fi>"]
 homepage = "https://github.com/pvarki/python-libpvarki/"

src/libpvarki/__init__.py

@@ -1,2 +1,2 @@
 """ Common helpers like standard logging init """
-__version__ = "1.8.0"  # NOTE Use `bump2version --config-file patch` to bump versions correctly
+__version__ = "1.9.0"  # NOTE Use `bump2version --config-file patch` to bump versions correctly

src/libpvarki/middleware/mtlsheader.py

@@ -9,7 +9,7 @@
 
 
 LOGGER = logging.getLogger(__name__)
-CONFIG = Config(".env")
+CONFIG = Config()  # not supporting .env files anymore because https://github.com/encode/starlette/discussions/2446
 DNDict = Mapping[str, str]
 
 

src/libpvarki/mtlshelp/context.py

@@ -8,7 +8,7 @@
 
 
 LOGGER = logging.getLogger(__name__)
-CONFIG = Config(".env")
+CONFIG = Config()  # not supporting .env files anymore because https://github.com/encode/starlette/discussions/2446
 
 # https://github.com/miguelgrinberg/python-socketio/discussions/1040 was very helpful
 

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, *, stderr_warn: bool = True) -> 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 and stderr_warn:
+        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_libpvarki.py

@@ -4,4 +4,4 @@
 
 def test_version() -> None:
     """Make sure version matches expected"""
-    assert __version__ == "1.8.0"
+    assert __version__ == "1.9.0"

tests/test_shell.py

@@ -0,0 +1,63 @@
+"""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")
+    # FIXME: Capture log output and check for the warning
+    assert code == 0
+    assert stdout == ""
+    assert stderr == "goodbye world\n"
+
+
+@pytest.mark.asyncio
+async def test_stderr_nowarn() -> 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", stderr_warn=False)
+    # FIXME: Capture log output and check that there is no warning
+    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)