subprocess_util: fix formatting for log callbacks (#116)

* subprocess_util: escape brackets in output for twiggy

Twiggy attempts to preform `str.format`-style formatting on log entries.
Brackets in the output cause Exceptions or other unpredictable results.

* subprocess_util: make stdout/stderr log handling DRY

* subprocess_util: refactor _escape_brackets logic

We can simply pass the value to the logger as a string formatting
argument instead of trying to perform naive manual escaping.

Suggested-by: Felix Fontein <felix@fontein.de>

changelogs/fragments/116-subprocess_util_escape.yaml

@@ -0,0 +1,5 @@
+---
+bugfixes:
+  - "``subprocess_util.log_run`` - use proper string formatting when passing
+    command output to the logger
+    (https://github.com/ansible-community/antsibull-core/pull/116)."

src/antsibull_core/subprocess_util.py

@@ -16,15 +16,15 @@
 from collections.abc import Awaitable, Callable, Sequence
 from functools import partial
 from inspect import isawaitable
+from logging import Logger as StdLogger
 from typing import TYPE_CHECKING, Any, TypeVar, cast
 
+from twiggy.logger import Logger as TwiggyLogger  # type: ignore[import]
+
 from antsibull_core.logging import log
 
 if TYPE_CHECKING:
-    from logging import Logger as StdLogger
-
     from _typeshed import StrOrBytesPath
-    from twiggy.logger import Logger as TwiggyLogger  # type: ignore[import]
     from typing_extensions import ParamSpec, TypeAlias
 
     _T = TypeVar("_T")
@@ -87,6 +87,31 @@ async def _stream_log(
     return "".join(lines)
 
 
+def _get_log_func_and_prefix(
+    name: str, loglevel: str | OutputCallbackType | None, logger: Any
+) -> tuple[OutputCallbackType | None, str]:
+    logfunc: Callable[[str], Any] | None = None
+    log_prefix = f"{name}: "
+    if loglevel:
+        if callable(loglevel):
+            logfunc = loglevel
+            log_prefix = ""
+        else:
+            # fmt: off
+            func = getattr(logger, loglevel)
+            if isinstance(logger, TwiggyLogger):
+                def logfunc(string: str, /):
+                    func("{0}", string)
+            elif isinstance(logger, StdLogger):
+                def logfunc(string: str, /):
+                    func("%s", string)
+            else:
+                logfunc = func
+            # fmt: on
+
+    return logfunc, log_prefix
+
+
 async def async_log_run(
     args: Sequence[StrOrBytesPath],
     logger: TwiggyLogger | StdLogger | None = None,
@@ -120,22 +145,12 @@ async def async_log_run(
         How to handle UTF-8 decoding errors. Default is ``strict``.
     """
     logger = logger or mlog
-    stdout_logfunc: Callable[[str], Any] | None = None
-    stdout_log_prefix = "stdout: "
-    if stdout_loglevel:
-        if callable(stdout_loglevel):
-            stdout_logfunc = stdout_loglevel
-            stdout_log_prefix = ""
-        else:
-            stdout_logfunc = getattr(logger, stdout_loglevel)
-    stderr_logfunc: Callable[[str], Any] | None = None
-    stderr_log_prefix = "stderr: "
-    if stderr_loglevel:
-        if callable(stderr_loglevel):
-            stderr_logfunc = stderr_loglevel
-            stderr_log_prefix = ""
-        else:
-            stderr_logfunc = getattr(logger, stderr_loglevel)
+    stdout_logfunc, stdout_log_prefix = _get_log_func_and_prefix(
+        "stdout", stdout_loglevel, logger
+    )
+    stderr_logfunc, stderr_log_prefix = _get_log_func_and_prefix(
+        "stderr", stderr_loglevel, logger
+    )
     logger.debug(f"Running subprocess: {args!r}")
     kwargs["stdout"] = asyncio.subprocess.PIPE
     kwargs["stderr"] = asyncio.subprocess.PIPE

tests/units/test_subprocess_util.py

@@ -3,11 +3,14 @@
 # GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or
 # https://www.gnu.org/licenses/gpl-3.0.txt)
 
+import logging as stdlog
 from unittest.mock import MagicMock, call
 
 import pytest
+import twiggy
 
 import antsibull_core.subprocess_util
+from antsibull_core import logging
 
 
 def test_log_run() -> None:
@@ -89,3 +92,33 @@ async def add_to_stderr(string: str, /) -> None:
     )
     assert stdout_lines == ["Never", "give"]
     assert stderr_lines == ["gonna"]
+
+
+def test_log_run_brackets_twiggy(capsys: pytest.CaptureFixture) -> None:
+    try:
+        logging.initialize_app_logging()
+        msg = "{abc} {x} }{}"
+        args = ("echo", msg)
+        antsibull_core.subprocess_util.log_run(
+            args, logger=logging.log, stdout_loglevel="error"
+        )
+        _, stderr = capsys.readouterr()
+        assert stderr == f"ERROR:antsibull|stdout: {msg}\n"
+    finally:
+        logging.log.min_level = twiggy.levels.DISABLED
+
+
+def test_log_run_percent_logging(capsys: pytest.CaptureFixture) -> None:
+    logger = stdlog.Logger("test_logger")
+    logger.setLevel(stdlog.WARNING)
+    ch = stdlog.StreamHandler()
+    ch.setLevel(stdlog.WARNING)
+    formatter = stdlog.Formatter("%(levelname)s:%(name)s|%(message)s")
+    ch.setFormatter(formatter)
+    logger.addHandler(ch)
+
+    msg = "%s %(abc)s % %%"
+    args = ("echo", msg)
+    antsibull_core.subprocess_util.log_run(args, logger=logger, stdout_loglevel="error")
+    _, stderr = capsys.readouterr()
+    assert stderr == f"ERROR:test_logger|stdout: {msg}\n"