346 lines
9.1 KiB
Python
346 lines
9.1 KiB
Python
"""Click CLI for Gitea Runner Manager."""
|
|
|
|
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
|
|
|
|
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=__version__)
|
|
def cli() -> None:
|
|
pass
|
|
|
|
|
|
@cli.command(help=_("Install and configure a Gitea Runner on a remote host."))
|
|
@click.argument("host")
|
|
@click.option(
|
|
"--user",
|
|
"-u",
|
|
default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()),
|
|
help=_("SSH user"),
|
|
)
|
|
@click.option("--key", "-k", default=lambda: os.getenv("GITEA_RUNNER_KEY"), help=_("Path to SSH private key"))
|
|
@click.option("--name", "-n", help=_("Gitea Runner name (default: host)"))
|
|
@click.option(
|
|
"--token",
|
|
"-t",
|
|
default=lambda: os.getenv("GITEA_REGISTRATION_TOKEN"),
|
|
help=_("Registration token (env: GITEA_REGISTRATION_TOKEN)"),
|
|
)
|
|
@click.option(
|
|
"--mode",
|
|
"-m",
|
|
type=click.Choice(["docker", "binary"]),
|
|
default="docker",
|
|
help=_("Gitea Runner deployment mode (default: docker)"),
|
|
)
|
|
@click.option(
|
|
"--admin-token",
|
|
"-a",
|
|
default=lambda: os.getenv("GITEA_ADMIN_TOKEN"),
|
|
help=_("Gitea admin API token for integration test (env: GITEA_ADMIN_TOKEN)"),
|
|
)
|
|
@click.option(
|
|
"--integration-retries",
|
|
"-r",
|
|
type=int,
|
|
default=lambda: int(os.getenv("GITEA_INTEGRATION_RETRIES", "3")),
|
|
help=_("Integration test API retries (default: 3, env: GITEA_INTEGRATION_RETRIES)"),
|
|
)
|
|
@click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password"))
|
|
def install(
|
|
host: str,
|
|
user: str,
|
|
key: str | None,
|
|
name: str | None,
|
|
token: str | None,
|
|
mode: str,
|
|
admin_token: str | None,
|
|
integration_retries: int,
|
|
ask_become_pass: bool,
|
|
) -> None:
|
|
manager = RunnerManager()
|
|
try:
|
|
manager.install(
|
|
host=host,
|
|
user=user,
|
|
key=key,
|
|
name=name,
|
|
token=token,
|
|
gitea_url=os.getenv("GITEA_URL", ""),
|
|
mode=mode,
|
|
admin_token=admin_token,
|
|
integration_retries=integration_retries,
|
|
ask_become_pass=ask_become_pass,
|
|
)
|
|
except GRMError as e:
|
|
raise click.ClickException(_("Installation failed: {error}", error=e)) from e
|
|
|
|
|
|
@cli.command(help=_("Update the Gitea Runner binary on a remote host."))
|
|
@click.argument("host")
|
|
@click.option(
|
|
"--user",
|
|
"-u",
|
|
default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()),
|
|
help=_("SSH user"),
|
|
)
|
|
@click.option("--key", "-k", default=lambda: os.getenv("GITEA_RUNNER_KEY"), help=_("Path to SSH private key"))
|
|
@click.option("--version", "-v", help=_("Specific Gitea Runner version"))
|
|
@click.option(
|
|
"--mode",
|
|
"-m",
|
|
type=click.Choice(["docker", "binary"]),
|
|
default="docker",
|
|
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,
|
|
key: str | None,
|
|
version: str | None,
|
|
mode: str,
|
|
ask_become_pass: bool,
|
|
) -> None:
|
|
manager = RunnerManager()
|
|
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")
|
|
@_runner_options
|
|
@click.option(
|
|
"--mode",
|
|
"-m",
|
|
type=click.Choice(["docker", "binary"]),
|
|
help=_("Override deployment mode from registry"),
|
|
)
|
|
@_handle_errors("Start failed: {error}")
|
|
def start(
|
|
runner_name: str,
|
|
host: str | None,
|
|
user: str | None,
|
|
key: str | None,
|
|
mode: str | None,
|
|
ask_become_pass: bool,
|
|
) -> None:
|
|
manager = RunnerManager()
|
|
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")
|
|
@_runner_options
|
|
@_handle_errors("Stop failed: {error}")
|
|
def stop(
|
|
runner_name: str,
|
|
host: str | None,
|
|
user: str | None,
|
|
key: str | None,
|
|
ask_become_pass: bool,
|
|
) -> None:
|
|
manager = RunnerManager()
|
|
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")
|
|
@_runner_options
|
|
@_handle_errors("Enable failed: {error}")
|
|
def enable(
|
|
runner_name: str,
|
|
host: str | None,
|
|
user: str | None,
|
|
key: str | None,
|
|
ask_become_pass: bool,
|
|
) -> None:
|
|
manager = RunnerManager()
|
|
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")
|
|
@_runner_options
|
|
@click.option(
|
|
"--token",
|
|
"-t",
|
|
default=lambda: os.getenv("GITEA_REGISTRATION_TOKEN"),
|
|
help=_("Registration token (env: GITEA_REGISTRATION_TOKEN)"),
|
|
)
|
|
@click.option(
|
|
"--mode",
|
|
"-m",
|
|
type=click.Choice(["docker", "binary"]),
|
|
help=_("Override deployment mode from registry"),
|
|
)
|
|
@_handle_errors("Disable failed: {error}")
|
|
def disable(
|
|
runner_name: str,
|
|
host: str | None,
|
|
user: str | None,
|
|
key: str | None,
|
|
token: str | None,
|
|
mode: str | None,
|
|
ask_become_pass: bool,
|
|
) -> None:
|
|
manager = RunnerManager()
|
|
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")
|
|
@_runner_options
|
|
@click.option(
|
|
"--mode",
|
|
"-m",
|
|
type=click.Choice(["docker", "binary"]),
|
|
help=_("Override deployment mode from registry"),
|
|
)
|
|
@_handle_errors("Status check failed: {error}")
|
|
def status(
|
|
runner_name: str,
|
|
host: str | None,
|
|
user: str | None,
|
|
key: str | None,
|
|
mode: str | None,
|
|
ask_become_pass: bool,
|
|
) -> None:
|
|
manager = RunnerManager()
|
|
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")
|
|
@_runner_options
|
|
@click.option(
|
|
"--token",
|
|
"-t",
|
|
default=lambda: os.getenv("GITEA_REGISTRATION_TOKEN"),
|
|
help=_("Registration token (env: GITEA_REGISTRATION_TOKEN)"),
|
|
)
|
|
@click.option(
|
|
"--mode",
|
|
"-m",
|
|
type=click.Choice(["docker", "binary"]),
|
|
help=_("Override deployment mode from registry"),
|
|
)
|
|
@_handle_errors("Remove failed: {error}")
|
|
def remove(
|
|
runner_name: str,
|
|
host: str | None,
|
|
user: str | None,
|
|
key: str | None,
|
|
token: str | None,
|
|
mode: str | None,
|
|
ask_become_pass: bool,
|
|
) -> None:
|
|
manager = RunnerManager()
|
|
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()
|
|
runners = manager.list_runners()
|
|
|
|
if not runners:
|
|
click.echo(_("No runners registered. Use 'grm install' to add one."))
|
|
return
|
|
|
|
click.echo(f"{_('NAME'):<20} {_('HOST'):<18} {_('USER'):<12} {_('MODE'):<8} {_('STATUS')}")
|
|
click.echo("-" * 70)
|
|
for r in runners:
|
|
click.echo(f"{r['name']:<20} {r['host']:<18} {r['user']:<12} {r['mode']:<8} {r['status']}")
|