From b3ac955515d8ae6a6fe4103681084a752f5ca9bf Mon Sep 17 00:00:00 2001 From: Emil Simeonov Date: Fri, 19 Jun 2026 00:50:20 +0200 Subject: [PATCH] GRM-10: refactor: deduplicate CLI, remove dead code, move validation to business layer --- src/gitea_runner_manager/cli.py | 222 ++++++++++----------- src/gitea_runner_manager/i18n.py | 42 ---- src/gitea_runner_manager/runner_manager.py | 6 + tests/unit/test_cli.py | 18 ++ tests/unit/test_runner_manager.py | 27 ++- 5 files changed, 151 insertions(+), 164 deletions(-) diff --git a/src/gitea_runner_manager/cli.py b/src/gitea_runner_manager/cli.py index b60daff..94d2bdb 100644 --- a/src/gitea_runner_manager/cli.py +++ b/src/gitea_runner_manager/cli.py @@ -2,11 +2,15 @@ from __future__ import annotations +import functools import os +from collections.abc import Callable +from typing import Any import click from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType] +from . import __version__ from .exceptions import GRMError from .i18n import _ from .runner_manager import RunnerManager @@ -14,8 +18,33 @@ from .runner_manager import RunnerManager load_dotenv(override=True) +def _runner_options(func: Callable[..., Any]) -> Callable[..., Any]: + """Apply common override options for registry-based lifecycle commands.""" + func = click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password"))(func) + func = click.option("--key", "-k", help=_("Override SSH key from registry"))(func) + func = click.option("--user", "-u", help=_("Override user from registry"))(func) + func = click.option("--host", help=_("Override host from registry"))(func) + return func + + +def _handle_errors(msg_key: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]: + """Convert GRMError into a Click exception with a translated message.""" + + def decorator(func: Callable[..., Any]) -> Callable[..., Any]: + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + try: + return func(*args, **kwargs) + except GRMError as e: + raise click.ClickException(_(msg_key, error=e)) from e + + return wrapper + + return decorator + + @click.group(help=_("Gitea Runner Manager — manage Gitea Actions runners.")) -@click.version_option(version="0.1.0") +@click.version_option(version=__version__) def cli() -> None: pass @@ -68,9 +97,6 @@ def install( integration_retries: int, ask_become_pass: bool, ) -> None: - gitea_url = os.getenv("GITEA_URL", "") - if not gitea_url: - raise click.ClickException(_("GITEA_URL must be set (or pass --url)")) manager = RunnerManager() try: manager.install( @@ -79,7 +105,7 @@ def install( key=key, name=name, token=token, - gitea_url=gitea_url, + gitea_url=os.getenv("GITEA_URL", ""), mode=mode, admin_token=admin_token, integration_retries=integration_retries, @@ -107,6 +133,7 @@ def install( help=_("Gitea Runner deployment mode (default: docker)"), ) @click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password")) +@_handle_errors("Update failed: {error}") def update( host: str, user: str, @@ -116,31 +143,26 @@ def update( ask_become_pass: bool, ) -> None: manager = RunnerManager() - try: - manager.update( - host=host, - user=user, - key=key, - version=version, - mode=mode, - ask_become_pass=ask_become_pass, - ) - except GRMError as e: - raise click.ClickException(_("Update failed: {error}", error=e)) from e + manager.update( + host=host, + user=user, + key=key, + version=version, + mode=mode, + ask_become_pass=ask_become_pass, + ) @cli.command(help=_("Start a registered Gitea Runner.")) @click.argument("runner_name") -@click.option("--host", help=_("Override host from registry")) -@click.option("--user", "-u", help=_("Override user from registry")) -@click.option("--key", "-k", help=_("Override SSH key from registry")) +@_runner_options @click.option( "--mode", "-m", type=click.Choice(["docker", "binary"]), help=_("Override deployment mode from registry"), ) -@click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password")) +@_handle_errors("Start failed: {error}") def start( runner_name: str, host: str | None, @@ -150,25 +172,20 @@ def start( ask_become_pass: bool, ) -> None: manager = RunnerManager() - try: - manager.start( - name=runner_name, - host=host, - user=user, - key=key, - mode=mode, - ask_become_pass=ask_become_pass, - ) - except GRMError as e: - raise click.ClickException(_("Start failed: {error}", error=e)) from e + manager.start( + name=runner_name, + host=host, + user=user, + key=key, + mode=mode, + ask_become_pass=ask_become_pass, + ) @cli.command(help=_("Stop a registered Gitea Runner.")) @click.argument("runner_name") -@click.option("--host", help=_("Override host from registry")) -@click.option("--user", "-u", help=_("Override user from registry")) -@click.option("--key", "-k", help=_("Override SSH key from registry")) -@click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password")) +@_runner_options +@_handle_errors("Stop failed: {error}") def stop( runner_name: str, host: str | None, @@ -177,24 +194,19 @@ def stop( ask_become_pass: bool, ) -> None: manager = RunnerManager() - try: - manager.stop( - name=runner_name, - host=host, - user=user, - key=key, - ask_become_pass=ask_become_pass, - ) - except GRMError as e: - raise click.ClickException(_("Stop failed: {error}", error=e)) from e + manager.stop( + name=runner_name, + host=host, + user=user, + key=key, + ask_become_pass=ask_become_pass, + ) @cli.command(help=_("Enable a registered Gitea Runner to start on boot.")) @click.argument("runner_name") -@click.option("--host", help=_("Override host from registry")) -@click.option("--user", "-u", help=_("Override user from registry")) -@click.option("--key", "-k", help=_("Override SSH key from registry")) -@click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password")) +@_runner_options +@_handle_errors("Enable failed: {error}") def enable( runner_name: str, host: str | None, @@ -203,23 +215,18 @@ def enable( ask_become_pass: bool, ) -> None: manager = RunnerManager() - try: - manager.enable( - name=runner_name, - host=host, - user=user, - key=key, - ask_become_pass=ask_become_pass, - ) - except GRMError as e: - raise click.ClickException(_("Enable failed: {error}", error=e)) from e + manager.enable( + name=runner_name, + host=host, + user=user, + key=key, + ask_become_pass=ask_become_pass, + ) @cli.command(help=_("Disable a registered Gitea Runner and deregister it.")) @click.argument("runner_name") -@click.option("--host", help=_("Override host from registry")) -@click.option("--user", "-u", help=_("Override user from registry")) -@click.option("--key", "-k", help=_("Override SSH key from registry")) +@_runner_options @click.option( "--token", "-t", @@ -232,7 +239,7 @@ def enable( type=click.Choice(["docker", "binary"]), help=_("Override deployment mode from registry"), ) -@click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password")) +@_handle_errors("Disable failed: {error}") def disable( runner_name: str, host: str | None, @@ -242,37 +249,29 @@ def disable( mode: str | None, ask_become_pass: bool, ) -> None: - gitea_url = os.getenv("GITEA_URL", "") - if not gitea_url: - raise click.ClickException(_("GITEA_URL must be set (or pass --url)")) manager = RunnerManager() - try: - manager.disable( - name=runner_name, - host=host, - user=user, - key=key, - token=token, - gitea_url=gitea_url, - mode=mode, - ask_become_pass=ask_become_pass, - ) - except GRMError as e: - raise click.ClickException(_("Disable failed: {error}", error=e)) from e + manager.disable( + name=runner_name, + host=host, + user=user, + key=key, + token=token, + gitea_url=os.getenv("GITEA_URL", ""), + mode=mode, + ask_become_pass=ask_become_pass, + ) @cli.command(help=_("Check the status of a registered Gitea Runner.")) @click.argument("runner_name") -@click.option("--host", help=_("Override host from registry")) -@click.option("--user", "-u", help=_("Override user from registry")) -@click.option("--key", "-k", help=_("Override SSH key from registry")) +@_runner_options @click.option( "--mode", "-m", type=click.Choice(["docker", "binary"]), help=_("Override deployment mode from registry"), ) -@click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password")) +@_handle_errors("Status check failed: {error}") def status( runner_name: str, host: str | None, @@ -282,24 +281,19 @@ def status( ask_become_pass: bool, ) -> None: manager = RunnerManager() - try: - manager.status( - name=runner_name, - host=host, - user=user, - key=key, - mode=mode, - ask_become_pass=ask_become_pass, - ) - except GRMError as e: - raise click.ClickException(_("Status check failed: {error}", error=e)) from e + manager.status( + name=runner_name, + host=host, + user=user, + key=key, + mode=mode, + ask_become_pass=ask_become_pass, + ) @cli.command(help=_("Remove a registered Gitea Runner completely.")) @click.argument("runner_name") -@click.option("--host", help=_("Override host from registry")) -@click.option("--user", "-u", help=_("Override user from registry")) -@click.option("--key", "-k", help=_("Override SSH key from registry")) +@_runner_options @click.option( "--token", "-t", @@ -312,7 +306,7 @@ def status( type=click.Choice(["docker", "binary"]), help=_("Override deployment mode from registry"), ) -@click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password")) +@_handle_errors("Remove failed: {error}") def remove( runner_name: str, host: str | None, @@ -322,32 +316,24 @@ def remove( mode: str | None, ask_become_pass: bool, ) -> None: - gitea_url = os.getenv("GITEA_URL", "") - if not gitea_url: - raise click.ClickException(_("GITEA_URL must be set (or pass --url)")) manager = RunnerManager() - try: - manager.remove( - name=runner_name, - host=host, - user=user, - key=key, - token=token, - gitea_url=gitea_url, - mode=mode, - ask_become_pass=ask_become_pass, - ) - except GRMError as e: - raise click.ClickException(_("Remove failed: {error}", error=e)) from e + manager.remove( + name=runner_name, + host=host, + user=user, + key=key, + token=token, + gitea_url=os.getenv("GITEA_URL", ""), + mode=mode, + ask_become_pass=ask_become_pass, + ) @cli.command(name="list", help=_("List all registered runners with live status.")) +@_handle_errors("List failed: {error}") def list_runners() -> None: manager = RunnerManager() - try: - runners = manager.list_runners() - except GRMError as e: - raise click.ClickException(_("List failed: {error}", error=e)) from e + runners = manager.list_runners() if not runners: click.echo(_("No runners registered. Use 'grm install' to add one.")) diff --git a/src/gitea_runner_manager/i18n.py b/src/gitea_runner_manager/i18n.py index e4c2aaa..5afb277 100644 --- a/src/gitea_runner_manager/i18n.py +++ b/src/gitea_runner_manager/i18n.py @@ -163,48 +163,6 @@ TRANSLATIONS: dict[str, dict[str, str]] = { "ru": "Обновить бинарный файл Gitea Runner на удалённом хосте.", "zh": "在远程主机上更新 Gitea Runner 二进制文件。", }, - "Start and configure a Gitea Runner on a remote host.": { - "en": "Start and configure a Gitea Runner on a remote host.", - "bg": "Стартиране и конфигуриране на Gitea Runner на отдалечен хост.", - "de": "Gitea Runner auf einem Remote-Host starten und konfigurieren.", - "ru": "Запустить и настроить Gitea Runner на удалённом хосте.", - "zh": "在远程主机上启动并配置 Gitea Runner。", - }, - "Stop a Gitea Runner on a remote host.": { - "en": "Stop a Gitea Runner on a remote host.", - "bg": "Спиране на Gitea Runner на отдалечен хост.", - "de": "Gitea Runner auf einem Remote-Host stoppen.", - "ru": "Остановить Gitea Runner на удалённом хосте.", - "zh": "在远程主机上停止 Gitea Runner。", - }, - "Enable a Gitea Runner to start on boot.": { - "en": "Enable a Gitea Runner to start on boot.", - "bg": "Активиране на Gitea Runner за стартиране при зареждане.", - "de": "Gitea Runner für den Start beim Booten aktivieren.", - "ru": "Включить автозапуск Gitea Runner при загрузке.", - "zh": "启用 Gitea Runner 开机自启。", - }, - "Disable a Gitea Runner and deregister it.": { - "en": "Disable a Gitea Runner and deregister it.", - "bg": "Деактивиране на Gitea Runner и дерегистрация.", - "de": "Gitea Runner deaktivieren und abmelden.", - "ru": "Отключить Gitea Runner и отменить его регистрацию.", - "zh": "禁用 Gitea Runner 并注销其注册。", - }, - "Check the status of a Gitea Runner.": { - "en": "Check the status of a Gitea Runner.", - "bg": "Проверка на състоянието на Gitea Runner.", - "de": "Status eines Gitea Runner prüfen.", - "ru": "Проверить состояние Gitea Runner.", - "zh": "检查 Gitea Runner 的状态。", - }, - "Remove a Gitea Runner completely.": { - "en": "Remove a Gitea Runner completely.", - "bg": "Пълно премахване на Gitea Runner.", - "de": "Gitea Runner vollständig entfernen.", - "ru": "Полностью удалить Gitea Runner.", - "zh": "完全移除 Gitea Runner。", - }, "Starting Gitea Runner {name} on {host}": { "en": "Starting Gitea Runner {name} on {host}", "bg": "Стартиране на Gitea Runner {name} на {host}", diff --git a/src/gitea_runner_manager/runner_manager.py b/src/gitea_runner_manager/runner_manager.py index dca4fdc..d0bc984 100644 --- a/src/gitea_runner_manager/runner_manager.py +++ b/src/gitea_runner_manager/runner_manager.py @@ -37,6 +37,8 @@ class RunnerManager: """Install a runner on a remote host using Ansible.""" if not name: name = host + if not gitea_url: + raise AnsibleError(_("GITEA_URL must be set (or pass --url)")) if not token: raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)")) @@ -162,6 +164,8 @@ class RunnerManager: ask_become_pass: bool = False, ) -> None: """Disable and deregister a runner instance.""" + if not gitea_url: + raise AnsibleError(_("GITEA_URL must be set (or pass --url)")) 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)")) @@ -198,6 +202,8 @@ class RunnerManager: ask_become_pass: bool = False, ) -> None: """Remove a runner instance completely.""" + if not gitea_url: + raise AnsibleError(_("GITEA_URL must be set (or pass --url)")) 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)")) diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 76b9bd3..6930c0c 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -37,6 +37,12 @@ class TestCLI: @patch("gitea_runner_manager.cli.RunnerManager") def test_install_missing_url(self, mock_manager_class: MagicMock) -> None: + mock_manager = MagicMock() + from gitea_runner_manager.exceptions import AnsibleError + + mock_manager.install.side_effect = AnsibleError("GITEA_URL must be set (or pass --url)") + mock_manager_class.return_value = mock_manager + with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True): runner = CliRunner() result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"]) @@ -310,6 +316,12 @@ class TestCLI: @patch("gitea_runner_manager.cli.RunnerManager") def test_disable_missing_url(self, mock_manager_class: MagicMock) -> None: + mock_manager = MagicMock() + from gitea_runner_manager.exceptions import AnsibleError + + mock_manager.disable.side_effect = AnsibleError("GITEA_URL must be set (or pass --url)") + mock_manager_class.return_value = mock_manager + with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True): runner = CliRunner() result = runner.invoke(cli, ["disable", "r1", "--token", "tok"]) @@ -419,6 +431,12 @@ class TestCLI: @patch("gitea_runner_manager.cli.RunnerManager") def test_remove_missing_url(self, mock_manager_class: MagicMock) -> None: + mock_manager = MagicMock() + from gitea_runner_manager.exceptions import AnsibleError + + mock_manager.remove.side_effect = AnsibleError("GITEA_URL must be set (or pass --url)") + mock_manager_class.return_value = mock_manager + with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True): runner = CliRunner() result = runner.invoke(cli, ["remove", "r1", "--token", "tok"]) diff --git a/tests/unit/test_runner_manager.py b/tests/unit/test_runner_manager.py index 0586062..2ac69c5 100644 --- a/tests/unit/test_runner_manager.py +++ b/tests/unit/test_runner_manager.py @@ -96,6 +96,11 @@ class TestRunnerManager: name="host1", host="host1", user="root", key=None, mode="binary", gitea_url="https://git.example.com" ) + def test_install_missing_gitea_url(self) -> None: + manager = RunnerManager() + with pytest.raises(AnsibleError, match="GITEA_URL must be set"): + manager.install("host", "user", token="tok") + def test_install_missing_token(self) -> None: manager = RunnerManager() with pytest.raises(AnsibleError, match="GITEA_REGISTRATION_TOKEN must be set"): @@ -105,7 +110,7 @@ class TestRunnerManager: manager = RunnerManager() with patch.object(Path, "exists", return_value=False): with pytest.raises(AnsibleError, match="Playbook not found"): - manager.install("host", "user", token="tok") + manager.install("host", "user", token="tok", gitea_url="https://git.example.com") def test_install_with_admin_token(self) -> None: mock_registry = MagicMock() @@ -125,7 +130,7 @@ class TestRunnerManager: manager._executor = mock_executor mock_executor.run.side_effect = AnsibleError("Ansible failed with exit code 1. See full log: /tmp/test.log") with pytest.raises(AnsibleError, match="Ansible failed with exit code 1"): - manager.install("host", "user", token="tok") + manager.install("host", "user", token="tok", gitea_url="https://git.example.com") mock_registry.add.assert_not_called() def test_update(self) -> None: @@ -308,7 +313,14 @@ class TestRunnerManager: mock_registry.get.return_value = {"host": "host", "user": "user"} manager = RunnerManager(registry=mock_registry) with pytest.raises(AnsibleError, match="GITEA_REGISTRATION_TOKEN must be set"): - manager.disable("r1") + manager.disable("r1", gitea_url="https://git.example.com") + + def test_disable_missing_gitea_url(self) -> None: + mock_registry = MagicMock() + mock_registry.get.return_value = {"host": "host", "user": "user"} + manager = RunnerManager(registry=mock_registry) + with pytest.raises(AnsibleError, match="GITEA_URL must be set"): + manager.disable("r1", token="tok") def test_status(self) -> None: mock_registry = MagicMock() @@ -348,7 +360,14 @@ class TestRunnerManager: mock_registry.get.return_value = {"host": "host", "user": "user"} manager = RunnerManager(registry=mock_registry) with pytest.raises(AnsibleError, match="GITEA_REGISTRATION_TOKEN must be set"): - manager.remove("r1") + manager.remove("r1", gitea_url="https://git.example.com") + + def test_remove_missing_gitea_url(self) -> None: + mock_registry = MagicMock() + mock_registry.get.return_value = {"host": "host", "user": "user"} + manager = RunnerManager(registry=mock_registry) + with pytest.raises(AnsibleError, match="GITEA_URL must be set"): + manager.remove("r1", token="tok") def test_remove_registry_deleted_on_failure(self) -> None: mock_registry = MagicMock()