GRM-10: refactor: deduplicate CLI, remove dead code, move validation to business layer

This commit is contained in:
Emil Simeonov
2026-06-19 00:57:26 +02:00
parent 4c08606d0c
commit b3ac955515
5 changed files with 151 additions and 164 deletions
+104 -118
View File
@@ -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."))
-42
View File
@@ -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}",
@@ -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)"))