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

GRM-12: feat: add translated operation report for success and failure cases

Browse Source 
- Create StepTracker context manager in new report.py module
- Track each step of lifecycle operations (install, update, start, stop, enable,
  disable, status, remove) with begin/done markers
- On success: report lists all completed steps with ✓ icons
- On failure: report shows failed step (✗), completed steps before failure (✓),
  and pending steps that never started (○)
- Add translations for report header, status labels, and registry step names
  in all 5 supported languages (EN/BG/DE/RU/ZH)
- 105 tests, 100% coverage, pyright clean, ruff clean
This commit is contained in:
Emil Simeonov
2026-06-19 01:16:39 +02:00
parent 3b68302274
commit 815b59c537
4 changed files with 297 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gitea_runner_manager/i18n.py
+42
View File
@@ -425,6 +425,48 @@ TRANSLATIONS: dict[str, dict[str, str]] = {
"ru": "неизвестно",
"zh": "未知",
},
"=== Operation Report ===": {
"en": "=== Operation Report ===",
"bg": "=== Отчет за операцията ===",
"de": "=== Operationsbericht ===",
"ru": "=== Отчет об операции ===",
"zh": "=== 操作报告 ===",
},
"completed": {
"en": "completed",
"bg": "завършено",
"de": "abgeschlossen",
"ru": "завершено",
"zh": "已完成",
},
"pending": {
"en": "pending",
"bg": "чакащо",
"de": "ausstehend",
"ru": "в ожидании",
"zh": "待处理",
},
"in_progress": {
"en": "in progress",
"bg": "в процес",
"de": "in Bearbeitung",
"ru": "в процессе",
"zh": "进行中",
},
"Save runner '{name}' to local registry": {
"en": "Save runner '{name}' to local registry",
"bg": "Запазване на runner '{name}' в локалния регистър",
"de": "Runner '{name}' in lokaler Registrierung speichern",
"ru": "Сохранить runner '{name}' в локальном реестре",
"zh": "将 runner '{name}' 保存到本地注册表",
},
"Remove runner '{name}' from local registry": {
"en": "Remove runner '{name}' from local registry",
"bg": "Премахване на runner '{name}' от локалния регистър",
"de": "Runner '{name}' aus lokaler Registrierung entfernen",
"ru": "Удалить runner '{name}' из локального реестра",
"zh": "从本地注册表移除 runner '{name}'",
},
}
src/gitea_runner_manager/report.py
+69
View File
@@ -0,0 +1,69 @@
"""Operation report tracking for GRM lifecycle commands."""
from __future__ import annotations
from collections.abc import Generator
from contextlib import contextmanager
from .i18n import _
class Step:
"""A single tracked step in an operation."""
def __init__(self, name: str) -> None:
self.name = name
self.status = "pending"
class StepTracker:
"""Tracks steps of an operation and prints a translated report on exit."""
def __init__(self) -> None:
self.steps: list[Step] = []
def begin(self, name: str) -> None:
"""Start a new step."""
step = Step(name)
self.steps.append(step)
step.status = "in_progress"
def done(self) -> None:
"""Mark the most recent in-progress step as completed."""
if self.steps and self.steps[-1].status == "in_progress":
self.steps[-1].status = "completed"
@contextmanager
def track_steps() -> Generator[StepTracker, None, None]:
"""Context manager that tracks steps and prints a report on exit.
On exception the last in-progress step is marked as failed.
The report is printed in the ``finally`` block so it always appears.
"""
tracker = StepTracker()
try:
yield tracker
except Exception:
for step in reversed(tracker.steps):
if step.status == "in_progress":
step.status = "failed"
break
raise
finally:
_print_report(tracker.steps)
def _print_report(steps: list[Step]) -> None:
"""Print a translated operation report to stdout."""
icons = {
"completed": "✓",
"failed": "✗",
"pending": "○",
"in_progress": "◌",
}
print(f"[GRM] {_('=== Operation Report ===')}")
for step in steps:
icon = icons.get(step.status, "?")
status_label = _(step.status)
print(f"[GRM] {icon} {step.name} ({status_label})")
src/gitea_runner_manager/runner_manager.py
+84 -33
View File
@@ -8,6 +8,7 @@ from .exceptions import AnsibleError
from .executor import AnsibleExecutor
from .i18n import _
from .registry import RunnerRegistry
from .report import track_steps
class RunnerManager:
@@ -50,17 +51,22 @@ class RunnerManager:
if admin_token:
extra_vars += f" gitea_admin_token={admin_token}"
cmd = self._build_cmd("install-runner.yml", host, user, extra_vars, key, ask_become_pass)
self._executor.run(cmd, description=_("Installing Gitea Runner on {host}", host=host))
with track_steps() as tracker:
tracker.begin(_("Installing Gitea Runner on {host}", host=host))
cmd = self._build_cmd("install-runner.yml", host, user, extra_vars, key, ask_become_pass)
self._executor.run(cmd, description=_("Installing Gitea Runner on {host}", host=host))
tracker.done()
self._registry.add(
name=name,
host=host,
user=user,
key=key,
mode=mode,
gitea_url=gitea_url,
)
tracker.begin(_("Save runner '{name}' to local registry", name=name))
self._registry.add(
name=name,
host=host,
user=user,
key=key,
mode=mode,
gitea_url=gitea_url,
)
tracker.done()
def update(
self,
@@ -76,8 +82,11 @@ class RunnerManager:
if version:
extra_vars += f" gitea_runner_version={version}"
cmd = self._build_cmd("update-runner.yml", host, user, extra_vars, key, ask_become_pass)
self._executor.run(cmd, description=_("Updating Gitea Runner on {host}", host=host))
with track_steps() as tracker:
tracker.begin(_("Updating Gitea Runner on {host}", host=host))
cmd = self._build_cmd("update-runner.yml", host, user, extra_vars, key, ask_become_pass)
self._executor.run(cmd, description=_("Updating Gitea Runner on {host}", host=host))
tracker.done()
def _resolve_runner(
self,
@@ -120,9 +129,14 @@ class RunnerManager:
) -> None:
"""Start a runner instance on a remote host."""
actual_host, actual_user, actual_key, actual_mode = self._resolve_runner(name, host, user, key, mode)
extra_vars = f"runner_name={name} runner_mode={actual_mode}"
cmd = self._build_cmd("start-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(cmd, description=_("Starting Gitea Runner {name} on {host}", name=name, host=actual_host))
with track_steps() as tracker:
tracker.begin(_("Starting Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = f"runner_name={name} runner_mode={actual_mode}"
cmd = self._build_cmd("start-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(
cmd, description=_("Starting Gitea Runner {name} on {host}", name=name, host=actual_host)
)
tracker.done()
def stop(
self,
@@ -134,9 +148,14 @@ class RunnerManager:
) -> None:
"""Stop a runner instance on a remote host."""
actual_host, actual_user, actual_key, _mode = self._resolve_runner(name, host, user, key)
extra_vars = f"runner_name={name}"
cmd = self._build_cmd("stop-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(cmd, description=_("Stopping Gitea Runner {name} on {host}", name=name, host=actual_host))
with track_steps() as tracker:
tracker.begin(_("Stopping Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = f"runner_name={name}"
cmd = self._build_cmd("stop-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(
cmd, description=_("Stopping Gitea Runner {name} on {host}", name=name, host=actual_host)
)
tracker.done()
def enable(
self,
@@ -148,9 +167,16 @@ class RunnerManager:
) -> None:
"""Enable a runner instance to start on boot."""
actual_host, actual_user, actual_key, _mode = self._resolve_runner(name, host, user, key)
extra_vars = f"runner_name={name}"
cmd = self._build_cmd("enable-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(cmd, description=_("Enabling Gitea Runner {name} on {host}", name=name, host=actual_host))
with track_steps() as tracker:
tracker.begin(_("Enabling Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = f"runner_name={name}"
cmd = self._build_cmd(
"enable-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
)
self._executor.run(
cmd, description=_("Enabling Gitea Runner {name} on {host}", name=name, host=actual_host)
)
tracker.done()
def disable(
self,
@@ -169,9 +195,18 @@ class RunnerManager:
actual_host, actual_user, actual_key, actual_mode = self._resolve_runner(name, host, user, key, mode)
if not token:
raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)"))
extra_vars = f"runner_name={name} registration_token={token} gitea_url={gitea_url} runner_mode={actual_mode}"
cmd = self._build_cmd("disable-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(cmd, description=_("Disabling Gitea Runner {name} on {host}", name=name, host=actual_host))
with track_steps() as tracker:
tracker.begin(_("Disabling Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = (
f"runner_name={name} registration_token={token} gitea_url={gitea_url} runner_mode={actual_mode}"
)
cmd = self._build_cmd(
"disable-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
)
self._executor.run(
cmd, description=_("Disabling Gitea Runner {name} on {host}", name=name, host=actual_host)
)
tracker.done()
def status(
self,
@@ -184,11 +219,16 @@ class RunnerManager:
) -> None:
"""Check the status of a runner instance."""
actual_host, actual_user, actual_key, actual_mode = self._resolve_runner(name, host, user, key, mode)
extra_vars = f"runner_name={name} runner_mode={actual_mode}"
cmd = self._build_cmd("status-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(
cmd, description=_("Checking status of Gitea Runner {name} on {host}", name=name, host=actual_host)
)
with track_steps() as tracker:
tracker.begin(_("Checking status of Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = f"runner_name={name} runner_mode={actual_mode}"
cmd = self._build_cmd(
"status-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
)
self._executor.run(
cmd, description=_("Checking status of Gitea Runner {name} on {host}", name=name, host=actual_host)
)
tracker.done()
def remove(
self,
@@ -207,11 +247,22 @@ class RunnerManager:
actual_host, actual_user, actual_key, actual_mode = self._resolve_runner(name, host, user, key, mode)
if not token:
raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)"))
extra_vars = f"runner_name={name} registration_token={token} gitea_url={gitea_url} runner_mode={actual_mode}"
cmd = self._build_cmd("remove-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(cmd, description=_("Removing Gitea Runner {name} from {host}", name=name, host=actual_host))
with track_steps() as tracker:
tracker.begin(_("Removing Gitea Runner {name} from {host}", name=name, host=actual_host))
extra_vars = (
f"runner_name={name} registration_token={token} gitea_url={gitea_url} runner_mode={actual_mode}"
)
cmd = self._build_cmd(
"remove-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
)
self._executor.run(
cmd, description=_("Removing Gitea Runner {name} from {host}", name=name, host=actual_host)
)
tracker.done()
self._registry.remove(name)
tracker.begin(_("Remove runner '{name}' from local registry", name=name))
self._registry.remove(name)
tracker.done()
def list_runners(self) -> list[dict[str, str]]:
"""Return a list of registered runners with live service status."""
tests/unit/test_report.py
+102
View File
@@ -0,0 +1,102 @@
"""Unit tests for the operation report tracker."""
from __future__ import annotations
import pytest
from gitea_runner_manager.exceptions import AnsibleError
from gitea_runner_manager.report import Step, StepTracker, track_steps
class TestStep:
def test_init(self) -> None:
step = Step("test")
assert step.name == "test"
assert step.status == "pending"
class TestStepTracker:
def test_begin(self) -> None:
tracker = StepTracker()
tracker.begin("step1")
assert len(tracker.steps) == 1
assert tracker.steps[0].name == "step1"
assert tracker.steps[0].status == "in_progress"
def test_done(self) -> None:
tracker = StepTracker()
tracker.begin("step1")
tracker.done()
assert tracker.steps[0].status == "completed"
def test_done_no_steps(self) -> None:
tracker = StepTracker()
tracker.done() # should not raise
assert tracker.steps == []
def test_done_last_not_in_progress(self) -> None:
tracker = StepTracker()
tracker.begin("step1")
tracker.done()
tracker.done() # second done is a no-op
assert tracker.steps[0].status == "completed"
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"):
with track_steps() as tracker:
tracker.begin("step1")
tracker.done()
tracker.begin("step2")
raise AnsibleError("fail")
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
def test_pending_shown_on_failure(self, capsys: pytest.CaptureFixture[str]) -> None:
with pytest.raises(AnsibleError, match="fail"):
with track_steps() as tracker:
tracker.begin("step1")
raise AnsibleError("fail")
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