From 50c92456806d3f6c370cb409b6ed121ad2f46afa Mon Sep 17 00:00:00 2001 From: Maxwell G Date: Wed, 29 Nov 2023 22:00:46 +0000 Subject: [PATCH] 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 --- .../fragments/116-subprocess_util_escape.yaml | 4 +-- src/antsibull_core/subprocess_util.py | 28 +++++++--------- tests/units/test_subprocess_util.py | 33 ++++++++++++------- 3 files changed, 35 insertions(+), 30 deletions(-) diff --git a/changelogs/fragments/116-subprocess_util_escape.yaml b/changelogs/fragments/116-subprocess_util_escape.yaml index 902a60f..b3b1d38 100644 --- a/changelogs/fragments/116-subprocess_util_escape.yaml +++ b/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)." diff --git a/src/antsibull_core/subprocess_util.py b/src/antsibull_core/subprocess_util.py index fa4f61d..56fb91e 100644 --- a/src/antsibull_core/subprocess_util.py +++ b/src/antsibull_core/subprocess_util.py @@ -16,6 +16,7 @@ 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] @@ -23,8 +24,6 @@ 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 diff --git a/tests/units/test_subprocess_util.py b/tests/units/test_subprocess_util.py index 3db67c3..9f6712a 100644 --- a/tests/units/test_subprocess_util.py +++ b/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"