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.

changelogs/fragments/116-subprocess_util_escape.yaml

@@ -0,0 +1,5 @@
+---
+bugfixes:
+  - "``subprocess_util`` - escape brackets in the subprocess output before
+    logging output with twiggy
+    (https://github.com/ansible-community/antsibull-core/pull/116)."

src/antsibull_core/subprocess_util.py

@@ -18,13 +18,14 @@
 from inspect import isawaitable
 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")
@@ -52,6 +53,18 @@ 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,
@@ -127,15 +140,23 @@ async def async_log_run(
             stdout_logfunc = stdout_loglevel
             stdout_log_prefix = ""
         else:
-            stdout_logfunc = getattr(logger, stdout_loglevel)
+            stdout_logfunc = cast(
+                "Callable[[str], Any]", getattr(logger, stdout_loglevel)
+            )
+            if isinstance(logger, TwiggyLogger):
+                stdout_logfunc = _escape_brackets(stdout_logfunc)
     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)
+            stderr_logfunc = cast(
+                "Callable[[str], Any]", getattr(logger, stderr_loglevel)
+            )
+            if isinstance(logger, TwiggyLogger):
+                stderr_logfunc = _escape_brackets(stderr_logfunc)
     logger.debug(f"Running subprocess: {args!r}")
     kwargs["stdout"] = asyncio.subprocess.PIPE
     kwargs["stderr"] = asyncio.subprocess.PIPE

tests/units/test_subprocess_util.py

@@ -6,8 +6,13 @@
 from unittest.mock import MagicMock, call
 
 import pytest
+import twiggy
 
 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:
@@ -89,3 +94,22 @@ async def add_to_stderr(string: str, /) -> None:
     )
     assert stdout_lines == ["Never", "give"]
     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:
+    try:
+        logging.initialize_app_logging()
+        args = ("echo", BRACKETS_ESCAPE_TEST)
+        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"
+    finally:
+        logging.log.min_level = twiggy.levels.DISABLED