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 <[email protected]>

changelogs/fragments/116-subprocess_util_escape.yaml

@@ -1,5 +1,5 @@
 ---
 bugfixes:
-  - "``subprocess_util`` - escape brackets in the subprocess output before
-    logging output with twiggy
+  - "``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,14 @@
 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 typing_extensions import ParamSpec, TypeAlias
 
@@ -53,18 +52,6 @@ async def _sync_or_async(
     return cast("_T", out)
 
 
-def _escape_brackets(func: Callable[[str], Any]) -> Callable[[str], Any]:
-    """
-    Return a function that takes a string, escapes brackets, and then calls
-    `func`
-    """
-
-    def inner(string: str, /):
-        func(string.replace("{", "{{").replace("}", "}}"))
-
-    return inner
-
-
 async def _stream_log(
     name: str,
     callback: OutputCallbackType | None,
@@ -110,9 +97,18 @@ def _get_log_func_and_prefix(
             logfunc = loglevel
             log_prefix = ""
         else:
-            logfunc = cast("Callable[[str], Any]", getattr(logger, loglevel))
+            # fmt: off
+            func = getattr(logger, loglevel)
             if isinstance(logger, TwiggyLogger):
-                logfunc = _escape_brackets(logfunc)
+                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
 
 

tests/units/test_subprocess_util.py

@@ -3,6 +3,7 @@
 # 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
@@ -11,9 +12,6 @@
 import antsibull_core.subprocess_util
 from antsibull_core import logging
 
-BRACKETS_ESCAPE_TEST = "{abc} {x} }{}"
-BRACKET_ESCAPE_TEST_ESCAPED = "{{abc}} {{x}} }}{{}}"
-
 
 def test_log_run() -> None:
     logger = MagicMock()
@@ -96,20 +94,31 @@ async def add_to_stderr(string: str, /) -> None:
     assert stderr_lines == ["gonna"]
 
 
-def test__escape_brackets() -> None:
-    d: list[str] = []
-    antsibull_core.subprocess_util._escape_brackets(d.append)(BRACKETS_ESCAPE_TEST)
-    assert d == [BRACKET_ESCAPE_TEST_ESCAPED]
-
-
-def test_log_run_brackets_escape(capsys: pytest.CaptureFixture) -> None:
+def test_log_run_brackets_twiggy(capsys: pytest.CaptureFixture) -> None:
     try:
         logging.initialize_app_logging()
-        args = ("echo", BRACKETS_ESCAPE_TEST)
+        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: {BRACKETS_ESCAPE_TEST}\n"
+        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"