GRM-1: feat: initial implementation of Gitea Runner Manager

2026-06-17 18:15:28 +02:00
commit 170b53ad28
52 changed files with 2191 additions and 0 deletions
@@ -0,0 +1,3 @@
+"""Gitea Runner Manager — lean CLI for managing Gitea Actions runners."""
+
+__version__ = "0.1.0"
@@ -0,0 +1,76 @@
+"""Gitea API client for runner management."""
+
+from __future__ import annotations
+
+import os
+import time
+from typing import Any
+
+import requests
+
+from .exceptions import GiteaAPIError, RunnerNotFoundError
+
+
+class GiteaAPIClient:
+    """Client for interacting with the Gitea API."""
+
+    def __init__(self, base_url: str, token: str) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.token = token
+        self.session = requests.Session()
+        self.session.headers.update({"Authorization": f"token {token}"})
+
+    @classmethod
+    def from_env(cls) -> GiteaAPIClient:
+        """Create a client from environment variables."""
+        base_url = os.getenv("GITEA_URL", "")
+        token = os.getenv("GITEA_TOKEN", "")
+        if not base_url or not token:
+            raise GiteaAPIError("GITEA_URL and GITEA_TOKEN must be set")
+        return cls(base_url, token)
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        params: dict[str, Any] | None = None,
+        json: dict[str, Any] | None = None,
+    ) -> requests.Response:
+        url = f"{self.base_url}/api/v1{path}"
+        response = self.session.request(method, url, params=params, json=json)
+        if not response.ok:
+            raise GiteaAPIError(
+                f"Gitea API error: {response.status_code} {response.text}",
+                status_code=response.status_code,
+            )
+        return response
+
+    def get_runners(self) -> list[dict[str, Any]]:
+        """List all registered runners."""
+        response = self._request("GET", "/admin/runners")
+        data: dict[str, Any] = response.json()
+        return data.get("runners", [])
+
+    def create_registration_token(self) -> str:
+        """Generate a new runner registration token."""
+        response = self._request("POST", "/admin/runners/registration-token")
+        data: dict[str, Any] = response.json()
+        token = data.get("token")
+        if not token:
+            raise GiteaAPIError("No token in response")
+        return token
+
+    def wait_for_runner(self, name: str, timeout: int = 120, interval: int = 10) -> dict[str, Any]:
+        """Wait for a runner to appear and be online."""
+        elapsed = 0
+        effective_interval = max(interval, 1)
+        while elapsed < timeout:
+            runners = self.get_runners()
+            for runner in runners:
+                if runner.get("name") == name:
+                    if runner.get("status") == "online":
+                        return runner
+                    raise RunnerNotFoundError(f"Runner '{name}' is not online")
+            time.sleep(effective_interval)
+            elapsed += effective_interval
+        raise RunnerNotFoundError(f"Runner '{name}' did not appear within {timeout}s")
@@ -0,0 +1,81 @@
+"""Click CLI for Gitea Runner Manager."""
+
+from __future__ import annotations
+
+import builtins
+import os
+from typing import Any
+
+import click
+from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnknownVariableType]
+
+from .api_client import GiteaAPIClient
+from .exceptions import GRMError
+from .runner_manager import RunnerManager
+
+load_dotenv()
+
+
+@click.group()
+@click.version_option(version="0.1.0")
+def cli() -> None:
+    """Gitea Runner Manager — manage Gitea Actions runners."""
+    pass
+
+
+@cli.command()
+def list() -> None:
+    """List all registered runners."""
+    api = GiteaAPIClient.from_env()
+    manager = RunnerManager(api)
+    runners: builtins.list[dict[str, Any]] = manager.list_runners()
+    if not runners:
+        click.echo("No runners found.")
+        return
+    click.echo(f"{'ID':<6} {'Name':<20} {'Status':<10}")
+    click.echo("-" * 40)
+    for runner in runners:
+        click.echo(f"{runner.get('id', 0):<6} {runner.get('name', 'N/A'):<20} {runner.get('status', 'unknown'):<10}")
+
+
+@cli.command()
+def token() -> None:
+    """Generate a new runner registration token."""
+    api = GiteaAPIClient.from_env()
+    manager = RunnerManager(api)
+    try:
+        tok = manager.generate_token()
+        click.echo(tok)
+    except GRMError as e:
+        raise click.ClickException(str(e)) from e
+
+
+@cli.command()
+@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="Runner name (default: host)")
+@click.option("--token", "-t", help="Registration token (auto-generated if omitted)")
+def install(host: str, user: str, key: str | None, name: str | None, token: str | None) -> None:
+    """Install and configure a runner on a remote host."""
+    api = GiteaAPIClient.from_env()
+    manager = RunnerManager(api)
+    try:
+        manager.install(host=host, user=user, key=key, name=name, token=token)
+    except GRMError as e:
+        raise click.ClickException(str(e)) from e
+
+
+@cli.command()
+@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 act_runner version")
+def update(host: str, user: str, key: str | None, version: str | None) -> None:
+    """Update the act_runner binary on a remote host."""
+    api = GiteaAPIClient.from_env()
+    manager = RunnerManager(api)
+    try:
+        manager.update(host=host, user=user, key=key, version=version)
+    except GRMError as e:
+        raise click.ClickException(str(e)) from e
@@ -0,0 +1,27 @@
+"""Custom exceptions for Gitea Runner Manager."""
+
+
+class GRMError(Exception):
+    """Base exception for all Gitea Runner Manager errors."""
+
+    pass
+
+
+class GiteaAPIError(GRMError):
+    """Raised when the Gitea API returns an error."""
+
+    def __init__(self, message: str, status_code: int = 0) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class AnsibleError(GRMError):
+    """Raised when an Ansible command fails."""
+
+    pass
+
+
+class RunnerNotFoundError(GRMError):
+    """Raised when a runner is not found in Gitea."""
+
+    pass
@@ -0,0 +1,93 @@
+"""Core logic for managing Gitea runners."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from pathlib import Path
+from typing import Any
+
+from .api_client import GiteaAPIClient
+from .exceptions import AnsibleError
+
+
+class RunnerManager:
+    """Orchestrates runner installation and updates."""
+
+    def __init__(self, api_client: GiteaAPIClient) -> None:
+        self.api = api_client
+
+    def list_runners(self) -> list[dict[str, Any]]:
+        """List all registered runners."""
+        return self.api.get_runners()
+
+    def generate_token(self) -> str:
+        """Generate a new registration token."""
+        return self.api.create_registration_token()
+
+    def install(
+        self,
+        host: str,
+        user: str,
+        key: str | None = None,
+        name: str | None = None,
+        token: str | None = None,
+    ) -> None:
+        """Install a runner on a remote host using Ansible."""
+        if not name:
+            name = host
+        if not token:
+            token = self.generate_token()
+
+        playbook = Path(__file__).parent.parent.parent / "ansible" / "install-runner.yml"
+        if not playbook.exists():
+            raise AnsibleError(f"Playbook not found: {playbook}")
+
+        cmd = [
+            "ansible-playbook",
+            str(playbook),
+            "-i",
+            f"{host},",
+            "-u",
+            user,
+            "--extra-vars",
+            f"registration_token={token} runner_name={name} gitea_url={self.api.base_url}",
+        ]
+        if key:
+            cmd.extend(["--private-key", key])
+
+        self._run_ansible(cmd)
+
+    def update(
+        self,
+        host: str,
+        user: str,
+        key: str | None = None,
+        version: str | None = None,
+    ) -> None:
+        """Update the act_runner binary on a remote host."""
+        playbook = Path(__file__).parent.parent.parent / "ansible" / "update-runner.yml"
+        if not playbook.exists():
+            raise AnsibleError(f"Playbook not found: {playbook}")
+
+        cmd = [
+            "ansible-playbook",
+            str(playbook),
+            "-i",
+            f"{host},",
+            "-u",
+            user,
+        ]
+        if key:
+            cmd.extend(["--private-key", key])
+        if version:
+            cmd.extend(["--extra-vars", f"act_runner_version={version}"])
+
+        self._run_ansible(cmd)
+
+    def _run_ansible(self, cmd: list[str]) -> None:
+        """Execute an Ansible command, streaming output."""
+        env = os.environ.copy()
+        result = subprocess.run(cmd, env=env, check=False)  # noqa: S603
+        if result.returncode != 0:
+            raise AnsibleError(f"Ansible failed with exit code {result.returncode}")