GRM-13: feat: replace print() with stdlib logging module

- Create logging_config.py with get_logger() providing dual handlers:
  - Console handler (stderr) controlled by GRM_LOG_LEVEL env var (default INFO)
  - File handler (~/.local/state/grm/logs/grm.log) capturing everything at DEBUG
- Replace all print() calls in executor.py and report.py with logger.info()/error()
- Add error logging before raising AnsibleError in executor.run()
- Add GRM_LOG_LEVEL to README configuration table and logging documentation
- Update all unit tests to mock logger instead of using capsys
- 114 tests, 100% coverage, pyright clean, ruff clean
This commit is contained in:
Emil Simeonov
2026-06-19 01:24:35 +02:00
parent 815b59c537
commit 3b3e002f7f
7 changed files with 287 additions and 103 deletions
+57 -44
View File
@@ -20,31 +20,40 @@ def _mock_popen_process(returncode: int = 0) -> MagicMock:
class TestAnsibleExecutorRun:
def test_run_success(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
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")
def test_run_success(self, tmp_path: Path) -> None:
mock_logger = MagicMock()
with patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger):
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")
captured = capsys.readouterr()
assert "[GRM] Test run" in captured.out
assert "[GRM] Done" in captured.out
messages = [call.args[0] for call in mock_logger.info.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:
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")
mock_logger = MagicMock()
with patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger):
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")
def test_run_default_description(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
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"])
assert mock_logger.error.called
assert "exit code 1" in mock_logger.error.call_args[0][0]
captured = capsys.readouterr()
assert "[GRM] Running Ansible playbook" in captured.out
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):
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]
assert any("Running Ansible playbook" in msg for msg in messages)
def test_run_with_stdout_lines(self, tmp_path: Path) -> None:
executor = AnsibleExecutor(log_dir=tmp_path)
@@ -60,20 +69,22 @@ class TestAnsibleExecutorRun:
handle.write.assert_any_call("line1\n")
handle.write.assert_any_call("line2\n")
def test_run_extracts_status(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
executor = AnsibleExecutor(log_dir=tmp_path)
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("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")
with patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger):
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")
captured = capsys.readouterr()
assert "Runner '127.0.0.1' is installed and running" in captured.out
messages = [call.args[0] for call in mock_logger.info.call_args_list]
assert any("Runner '127.0.0.1' is installed and running" in msg for msg in messages)
class TestAnsibleExecutorExtractStatus:
@@ -129,32 +140,34 @@ class TestAnsibleExecutorPrepareLog:
class TestAnsibleExecutorTranslation:
def test_run_translated(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
executor = AnsibleExecutor(log_dir=tmp_path)
def test_run_translated(self, tmp_path: Path) -> None:
mock_logger = MagicMock()
with (
patch.dict(os.environ, {"GRM_LANG": "de"}),
patch("subprocess.Popen") as mock_popen,
patch("builtins.open", mock_open()),
patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger),
):
mock_popen.return_value = _mock_popen_process(returncode=0)
executor.run(["echo", "hello"]) # use default description to test translation
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
captured = capsys.readouterr()
assert "Ansible-Playbook wird ausgeführt" in captured.out
assert "Fertig" in captured.out
messages = [call.args[0] for call in mock_logger.info.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, capsys: pytest.CaptureFixture[str]) -> None:
executor = AnsibleExecutor(log_dir=tmp_path)
def test_unsupported_lang_fallback(self, tmp_path: Path) -> None:
mock_logger = MagicMock()
with (
patch.dict(os.environ, {"GRM_LANG": "xx"}),
patch("subprocess.Popen") as mock_popen,
patch("builtins.open", mock_open()),
patch("gitea_runner_manager.executor.get_logger", return_value=mock_logger),
):
mock_popen.return_value = _mock_popen_process(returncode=0)
executor.run(["echo", "hello"])
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"])
captured = capsys.readouterr()
assert "[GRM] Running Ansible playbook" in captured.out # English fallback
messages = [call.args[0] for call in mock_logger.info.call_args_list]
assert any("Running Ansible playbook" in msg for msg in messages) # English fallback
class TestAnsibleExecutorAdHoc:
+88
View File
@@ -0,0 +1,88 @@
"""Unit tests for logging configuration."""
from __future__ import annotations
import logging
from collections.abc import Generator
from pathlib import Path
from unittest.mock import patch
import pytest
from gitea_runner_manager.logging_config import _level_from_env, get_logger
@pytest.fixture(autouse=True)
def _cleanup_loggers() -> Generator[None, None, None]:
"""Remove test loggers after each test to prevent state leakage."""
yield
for name in list(logging.Logger.manager.loggerDict.keys()):
if name.startswith("test_"):
logger = logging.getLogger(name)
logger.handlers.clear()
del logging.Logger.manager.loggerDict[name]
class TestLevelFromEnv:
def test_none_defaults_to_info(self) -> None:
assert _level_from_env(None) == logging.INFO
def test_empty_string_defaults_to_info(self) -> None:
assert _level_from_env("") == logging.INFO
def test_valid_levels(self) -> None:
assert _level_from_env("debug") == logging.DEBUG
assert _level_from_env("DEBUG") == logging.DEBUG
assert _level_from_env("warning") == logging.WARNING
assert _level_from_env("error") == logging.ERROR
assert _level_from_env("critical") == logging.CRITICAL
def test_invalid_fallback_to_info(self) -> None:
assert _level_from_env("foo") == logging.INFO
assert _level_from_env("VERBOSE") == logging.INFO
class TestGetLogger:
def test_returns_logger(self, tmp_path: Path) -> None:
with patch("gitea_runner_manager.logging_config.Path.home", return_value=tmp_path):
logger = get_logger("test_returns_logger")
assert isinstance(logger, logging.Logger)
assert logger.name == "test_returns_logger"
assert len(logger.handlers) == 2 # console + file
def test_caches_same_instance(self, tmp_path: Path) -> None:
with patch("gitea_runner_manager.logging_config.Path.home", return_value=tmp_path):
logger1 = get_logger("test_caches_same")
logger2 = get_logger("test_caches_same")
assert logger1 is logger2
def test_file_handler_writes(self, tmp_path: Path) -> None:
with patch("gitea_runner_manager.logging_config.Path.home", return_value=tmp_path):
logger = get_logger("test_file_writes")
logger.info("hello from test")
log_file = tmp_path / ".local" / "state" / "grm" / "logs" / "grm.log"
assert log_file.exists()
content = log_file.read_text()
assert "hello from test" in content
assert "INFO" in content
def test_respects_grm_log_level(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setenv("GRM_LOG_LEVEL", "DEBUG")
with patch("gitea_runner_manager.logging_config.Path.home", return_value=tmp_path):
logger = get_logger("test_env_level")
console = [h for h in logger.handlers if type(h) is logging.StreamHandler]
assert len(console) == 1
assert console[0].level == logging.DEBUG
def test_grm_log_level_warning(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setenv("GRM_LOG_LEVEL", "WARNING")
with patch("gitea_runner_manager.logging_config.Path.home", return_value=tmp_path):
logger = get_logger("test_env_warning")
console = [h for h in logger.handlers if type(h) is logging.StreamHandler]
assert len(console) == 1
assert console[0].level == logging.WARNING
+58 -47
View File
@@ -2,6 +2,8 @@
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from gitea_runner_manager.exceptions import AnsibleError
@@ -43,60 +45,69 @@ class TestStepTracker:
class TestTrackSteps:
def test_success(self, capsys: pytest.CaptureFixture[str]) -> None:
with track_steps() as tracker:
tracker.begin("step1")
tracker.done()
tracker.begin("step2")
tracker.done()
captured = capsys.readouterr()
assert "Operation Report" in captured.out
assert "step1" in captured.out
assert "step2" in captured.out
assert "" in captured.out
assert "completed" in captured.out
def test_failure_marks_step(self, capsys: pytest.CaptureFixture[str]) -> None:
with pytest.raises(AnsibleError, match="fail"):
def test_success(self) -> None:
mock_logger = MagicMock()
with patch("gitea_runner_manager.report.get_logger", return_value=mock_logger):
with track_steps() as tracker:
tracker.begin("step1")
tracker.done()
tracker.begin("step2")
raise AnsibleError("fail")
tracker.done()
captured = capsys.readouterr()
assert "Operation Report" in captured.out
assert "step1" in captured.out
assert "step2" in captured.out
assert "" in captured.out
assert "" in captured.out
assert "completed" in captured.out
assert "failed" in captured.out
messages = [call.args[0] for call in mock_logger.info.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)
assert any("" in msg for msg in messages)
assert any("completed" in msg for msg in messages)
def test_pending_shown_on_failure(self, capsys: pytest.CaptureFixture[str]) -> None:
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 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]
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)
assert any("" in msg for msg in messages)
assert any("completed" in msg for msg in messages)
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 track_steps() as tracker:
tracker.begin("step1")
raise AnsibleError("fail")
messages = [call.args[0] for call in mock_logger.info.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 track_steps():
pass
messages = [call.args[0] for call in mock_logger.info.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 track_steps() as tracker:
tracker.begin("step1")
raise AnsibleError("fail")
tracker.done()
captured = capsys.readouterr()
assert "" in captured.out
assert "failed" in captured.out
def test_empty_report(self, capsys: pytest.CaptureFixture[str]) -> None:
with track_steps():
pass
captured = capsys.readouterr()
assert "Operation Report" in captured.out
def test_translated_report(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
monkeypatch.setenv("GRM_LANG", "bg")
with track_steps() as tracker:
tracker.begin("step1")
tracker.done()
captured = capsys.readouterr()
assert "Отчет за операцията" in captured.out
assert "завършено" in captured.out
messages = [call.args[0] for call in mock_logger.info.call_args_list]
assert any("Отчет за операцията" in msg for msg in messages)
assert any("завършено" in msg for msg in messages)