oblachno-oss/grm
1
0
Fork 0
Code Issues Pull Requests Actions 14 Packages Projects Releases Wiki Activity

GRM-14: feat: use click.echo() for user-facing messages with dual logging

Browse Source 
- Create ui.py with say() helper that routes messages to both click.echo()
  (console/stdout, user-facing) and logging.getLogger('grm') (file audit trail)
- Update executor.py: replace logger.info() with say() for start, status, done
  messages; use say(level=ERROR, err=True) before raising AnsibleError
- Update report.py: replace logger.info() with say() for operation report lines
- Update all unit tests to patch say() instead of using capsys or get_logger
- Add test_ui.py with coverage for say() calling both click.echo and logging
- 117 tests, 100% coverage, pyright clean, ruff clean
This commit is contained in:
Emil Simeonov
2026-06-19 01:30:14 +02:00
parent 3b3e002f7f
commit 3f7507500b
6 changed files with 93 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gitea_runner_manager/executor.py
+7 -5
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import logging
import os
import re
import subprocess
@@ -11,6 +12,7 @@ from pathlib import Path
from .exceptions import AnsibleError
from .i18n import _
from .logging_config import get_logger
from .ui import say
class AnsibleExecutor:
@@ -25,8 +27,8 @@ class AnsibleExecutor:
log_file = self._prepare_log(cmd)
desc = description or _("Running Ansible playbook")
self._logger.info(desc)
self._logger.info(_("Full log: {log_file}", log_file=log_file))
say(f"[GRM] {desc}")
say(f"[GRM] {_('Full log: {log_file}', log_file=log_file)}")
returncode = self._stream(cmd, log_file)
@@ -36,14 +38,14 @@ class AnsibleExecutor:
code=returncode,
log_file=log_file,
)
self._logger.error(msg)
say(f"[GRM] {msg}", level=logging.ERROR, err=True)
raise AnsibleError(msg)
status = self._extract_status(log_file)
if status:
self._logger.info(status)
say(f"[GRM] {status}")
self._logger.info(_("Done. See full log: {log_file}", log_file=log_file))
say(f"[GRM] {_('Done. See full log: {log_file}', log_file=log_file)}")
def _prepare_log(self, cmd: list[str]) -> Path:
"""Create a log file with header."""
src/gitea_runner_manager/report.py
+3 -4
View File
@@ -6,7 +6,7 @@ from collections.abc import Generator
from contextlib import contextmanager
from .i18n import _
from .logging_config import get_logger
from .ui import say
class Step:
@@ -57,15 +57,14 @@ def track_steps() -> Generator[StepTracker, None, None]:
def _print_report(steps: list[Step]) -> None:
"""Print a translated operation report to stdout."""
logger = get_logger()
icons = {
"completed": "✓",
"failed": "✗",
"pending": "○",
"in_progress": "◌",
}
logger.info(_("=== Operation Report ==="))
say(f"[GRM] {_('=== Operation Report ===')}")
for step in steps:
icon = icons.get(step.status, "?")
status_label = _(step.status)
logger.info(f" {icon} {step.name} ({status_label})")
say(f"[GRM] {icon} {step.name} ({status_label})")
src/gitea_runner_manager/ui.py
+18
View File
@@ -0,0 +1,18 @@
"""User-facing output utilities for GRM."""
from __future__ import annotations
import logging
import click
def say(msg: str, level: int = logging.INFO, err: bool = False) -> None:
"""Output a message to the user and also log it for auditing.
Console output goes via ``click.echo`` (handles encoding, CliRunner,
Windows colorama). The same message is also sent to the ``grm`` logger
so it appears in the persistent log file.
"""
click.echo(msg, err=err)
logging.getLogger("grm").log(level, msg)
tests/unit/test_executor.py
+15 -19
View File
@@ -1,5 +1,6 @@
"""Unit tests for AnsibleExecutor."""
import logging
import os
from pathlib import Path
from unittest.mock import MagicMock, mock_open, patch
@@ -21,38 +22,36 @@ def _mock_popen_process(returncode: int = 0) -> MagicMock:
class TestAnsibleExecutorRun:
def test_run_success(self, tmp_path: Path) -> None:
mock_logger = MagicMock()
with patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger):
with patch("gitea_runner_manager.executor.say") as mock_say:
executor = AnsibleExecutor(log_dir=tmp_path)
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open()):
mock_popen.return_value = _mock_popen_process(returncode=0)
executor.run(["echo", "hello"], description="Test run")
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("Test run" in msg for msg in messages)
assert any("Done" in msg for msg in messages)
def test_run_failure(self, tmp_path: Path) -> None:
mock_logger = MagicMock()
with patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger):
with patch("gitea_runner_manager.executor.say") as mock_say:
executor = AnsibleExecutor(log_dir=tmp_path)
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open()):
mock_popen.return_value = _mock_popen_process(returncode=1)
with pytest.raises(AnsibleError, match="exit code 1"):
executor.run(["false"], description="Test run")
assert mock_logger.error.called
assert "exit code 1" in mock_logger.error.call_args[0][0]
error_calls = [call for call in mock_say.call_args_list if call.kwargs.get("level") == logging.ERROR]
assert len(error_calls) == 1
assert "exit code 1" in error_calls[0].args[0]
def test_run_default_description(self, tmp_path: Path) -> None:
mock_logger = MagicMock()
with patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger):
with patch("gitea_runner_manager.executor.say") as mock_say:
executor = AnsibleExecutor(log_dir=tmp_path)
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open()):
mock_popen.return_value = _mock_popen_process(returncode=0)
executor.run(["echo", "hello"])
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("Running Ansible playbook" in msg for msg in messages)
def test_run_with_stdout_lines(self, tmp_path: Path) -> None:
@@ -70,20 +69,19 @@ class TestAnsibleExecutorRun:
handle.write.assert_any_call("line2\n")
def test_run_extracts_status(self, tmp_path: Path) -> None:
mock_logger = MagicMock()
log_content = (
"TASK [gitea-runner : Report runner status]\n"
"ok: [127.0.0.1] => {\n"
' "msg": "Runner \'127.0.0.1\' is installed and running."\n'
"}\n"
)
with patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger):
with patch("gitea_runner_manager.executor.say") as mock_say:
executor = AnsibleExecutor(log_dir=tmp_path)
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open(read_data=log_content)):
mock_popen.return_value = _mock_popen_process(returncode=0)
executor.run(["echo", "hello"], description="Test run")
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("Runner '127.0.0.1' is installed and running" in msg for msg in messages)
@@ -141,32 +139,30 @@ class TestAnsibleExecutorPrepareLog:
class TestAnsibleExecutorTranslation:
def test_run_translated(self, tmp_path: Path) -> None:
mock_logger = MagicMock()
with (
patch.dict(os.environ, {"GRM_LANG": "de"}),
patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger),
patch("gitea_runner_manager.executor.say") as mock_say,
):
executor = AnsibleExecutor(log_dir=tmp_path)
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open()):
mock_popen.return_value = _mock_popen_process(returncode=0)
executor.run(["echo", "hello"]) # use default description to test translation
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("Ansible-Playbook wird ausgeführt" in msg for msg in messages)
assert any("Fertig" in msg for msg in messages)
def test_unsupported_lang_fallback(self, tmp_path: Path) -> None:
mock_logger = MagicMock()
with (
patch.dict(os.environ, {"GRM_LANG": "xx"}),
patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger),
patch("gitea_runner_manager.executor.say") as mock_say,
):
executor = AnsibleExecutor(log_dir=tmp_path)
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open()):
mock_popen.return_value = _mock_popen_process(returncode=0)
executor.run(["echo", "hello"])
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("Running Ansible playbook" in msg for msg in messages) # English fallback
tests/unit/test_report.py
+13 -18
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import pytest
@@ -46,15 +46,14 @@ class TestStepTracker:
class TestTrackSteps:
def test_success(self) -> None:
mock_logger = MagicMock()
with patch("gitea_runner_manager.report.get_logger", return_value=mock_logger):
with patch("gitea_runner_manager.report.say") as mock_say:
with track_steps() as tracker:
tracker.begin("step1")
tracker.done()
tracker.begin("step2")
tracker.done()
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("Operation Report" in msg for msg in messages)
assert any("step1" in msg for msg in messages)
assert any("step2" in msg for msg in messages)
@@ -62,16 +61,15 @@ class TestTrackSteps:
assert any("completed" in msg for msg in messages)
def test_failure_marks_step(self) -> None:
mock_logger = MagicMock()
with pytest.raises(AnsibleError, match="fail"):
with patch("gitea_runner_manager.report.get_logger", return_value=mock_logger):
with patch("gitea_runner_manager.report.say") as mock_say:
with pytest.raises(AnsibleError, match="fail"):
with track_steps() as tracker:
tracker.begin("step1")
tracker.done()
tracker.begin("step2")
raise AnsibleError("fail")
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("step1" in msg for msg in messages)
assert any("step2" in msg for msg in messages)
assert any("✓" in msg for msg in messages)
@@ -80,34 +78,31 @@ class TestTrackSteps:
assert any("failed" in msg for msg in messages)
def test_pending_shown_on_failure(self) -> None:
mock_logger = MagicMock()
with pytest.raises(AnsibleError, match="fail"):
with patch("gitea_runner_manager.report.get_logger", return_value=mock_logger):
with patch("gitea_runner_manager.report.say") as mock_say:
with pytest.raises(AnsibleError, match="fail"):
with track_steps() as tracker:
tracker.begin("step1")
raise AnsibleError("fail")
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("✗" in msg for msg in messages)
assert any("failed" in msg for msg in messages)
def test_empty_report(self) -> None:
mock_logger = MagicMock()
with patch("gitea_runner_manager.report.get_logger", return_value=mock_logger):
with patch("gitea_runner_manager.report.say") as mock_say:
with track_steps():
pass
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("Operation Report" in msg for msg in messages)
def test_translated_report(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GRM_LANG", "bg")
mock_logger = MagicMock()
with patch("gitea_runner_manager.report.get_logger", return_value=mock_logger):
with patch("gitea_runner_manager.report.say") as mock_say:
with track_steps() as tracker:
tracker.begin("step1")
tracker.done()
messages = [call.args[0] for call in mock_logger.info.call_args_list]
messages = [call.args[0] for call in mock_say.call_args_list]
assert any("Отчет за операцията" in msg for msg in messages)
assert any("завършено" in msg for msg in messages)
tests/unit/test_ui.py
+37
View File
@@ -0,0 +1,37 @@
"""Unit tests for user-facing output utilities."""
from __future__ import annotations
import logging
from unittest.mock import patch
from gitea_runner_manager.ui import say
class TestSay:
def test_echoes_to_console(self) -> None:
with patch("gitea_runner_manager.ui.click.echo") as mock_echo:
say("hello")
mock_echo.assert_called_once_with("hello", err=False)
def test_logs_at_info_level(self) -> None:
with (
patch("gitea_runner_manager.ui.click.echo"),
patch("gitea_runner_manager.ui.logging.getLogger") as mock_get_logger,
):
mock_logger = mock_get_logger.return_value
say("hello")
mock_logger.log.assert_called_once_with(logging.INFO, "hello")
def test_passes_level_and_err(self) -> None:
with (
patch("gitea_runner_manager.ui.click.echo") as mock_echo,
patch("gitea_runner_manager.ui.logging.getLogger") as mock_get_logger,
):
mock_logger = mock_get_logger.return_value
say("error msg", level=logging.ERROR, err=True)
mock_echo.assert_called_once_with("error msg", err=True)
mock_logger.log.assert_called_once_with(logging.ERROR, "error msg")