devx/src/devx/tools/install_tools.py

#!/usr/bin/env python3
"""Install CI/CD development tools that are not Python packages.

Handles installation of:
- actionlint (workflow YAML linter)
- git-cliff (changelog generator)
- act_runner (Gitea Actions local runner, optional)
- tea (Gitea CLI — official command-line tool for Gitea API operations)
- hadolint (Dockerfile linter)
- vale (prose linter for documentation quality)
- promtool (Prometheus rule validator)

Each tool is installed to ``~/.local/bin`` if not already on PATH.
Idempotent: skips tools that are already available.

Usage::

    python3 -m devx.tools.install_tools                    # install all
    python3 -m devx.tools.install_tools --tool actionlint  # install one
    python3 -m devx.tools.install_tools --list             # list status
"""

from __future__ import annotations

import logging
import os
import platform
import shutil
import tarfile
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path

import click
from tenacity import (
    Retrying,
    before_sleep_log,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)

TARGET_DIR = Path.home() / ".local" / "bin"

logger = logging.getLogger(__name__)

# Retry configuration for transient network failures during download.
# GitHub releases occasionally drops connections ("Remote end closed
# connection without response"). Retrying with backoff before falling
# through to the next fallback URL makes the build resilient to
# momentary network blips. 5 attempts with up to 30s between retries
# handles sustained transient outages (observed in CI image builds).
MAX_DOWNLOAD_RETRIES = 5

ACTIONLINT_VERSION = "1.7.12"

GIT_CLIFF_VERSION = "2.13.1"

ACT_RUNNER_VERSION = "0.2.11"

TEA_VERSION = "0.14.2"

HADOLINT_VERSION = "2.14.0"

TOFU_VERSION = "1.12.3"

VALE_VERSION = "3.15.1"

PROMTOOL_VERSION = "3.5.5"


def _arch() -> str:
    """Return the architecture string used by release assets (delegates to shared utility)."""
    from devx.tools._shared import arch_string

    return arch_string()


def _ensure_target_dir() -> Path:
    """Ensure the target directory exists and return it."""
    TARGET_DIR.mkdir(parents=True, exist_ok=True)
    return TARGET_DIR


def _download(url: str, dest: Path, *, _sleep=None) -> None:
    """Download a file from ``url`` to ``dest`` with retry and 60s timeout.

    Retries up to ``MAX_DOWNLOAD_RETRIES`` times on transient network
    errors (``URLError``, ``OSError`` from connection resets) using
    exponential backoff. This handles momentary GitHub releases
    connection drops that were causing CI image builds to fail.

    The ``_sleep`` kwarg is for tests to avoid real sleeping; production
    code should leave it as ``None`` (uses ``time.sleep``).
    """
    retrying = Retrying(
        stop=stop_after_attempt(MAX_DOWNLOAD_RETRIES),
        wait=wait_exponential(multiplier=2, min=2, max=30),
        retry=retry_if_exception_type((urllib.error.URLError, OSError, ConnectionError)),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        sleep=_sleep if _sleep is not None else time.sleep,
        reraise=True,
    )
    retrying(_do_download, url, dest)


def _do_download(url: str, dest: Path) -> None:
    """Single download attempt — called by :func:`_download` retry wrapper."""
    with urllib.request.urlopen(url, timeout=60) as resp, open(dest, "wb") as f:  # nosec B310
        shutil.copyfileobj(resp, f)


def _download_with_fallback(urls: list[str], binary_name: str) -> Path:
    """Try downloading a binary from a list of URLs, falling back on failure.

    Returns the path to the installed binary. Raises if all URLs fail.
    """
    target_dir = _ensure_target_dir()
    dest = target_dir / binary_name
    errors: list[str] = []
    for url in urls:
        try:
            _download(url, dest)
            dest.chmod(0o755)
            return dest
        except Exception as exc:  # noqa: BLE001
            errors.append(f"{url}: {exc}")
            click.echo(f"  {binary_name}: retrying — {exc}")
    raise click.ClickException(f"Failed to download {binary_name} from all URLs: {'; '.join(errors)}")


def _download_and_extract_tarball(url: str, binary_name: str, *, fallback_urls: list[str] | None = None) -> Path:
    """Download a tarball, extract the binary, and install it to TARGET_DIR.

    Returns the path to the installed binary. Falls back to ``fallback_urls``
    if the primary ``url`` fails all retries.
    """
    target_dir = _ensure_target_dir()
    dest = target_dir / binary_name
    urls = [url, *(fallback_urls or [])]
    errors: list[str] = []
    for try_url in urls:
        with tempfile.TemporaryDirectory() as tmpdir:
            tarball = Path(tmpdir) / "archive.tar.gz"
            try:
                _download(try_url, tarball)
            except Exception as exc:  # noqa: BLE001
                errors.append(f"{try_url}: {exc}")
                click.echo(f"  {binary_name}: fallback — {exc}")
                continue
            with tarfile.open(tarball, "r:gz") as tar:
                tar.extractall(tmpdir)  # nosec B202
            # Find the binary in the extracted tree
            extracted = Path(tmpdir).rglob(binary_name)
            found = next(extracted, None)
            if found is None:
                errors.append(f"{try_url}: binary not found in archive")
                continue
            shutil.copy2(found, dest)
            dest.chmod(0o755)
            return dest
    raise click.ClickException(f"Failed to download {binary_name} from all URLs: {'; '.join(errors)}")


def _download_binary(url: str, binary_name: str) -> Path:
    """Download a standalone binary and install it to TARGET_DIR.

    Returns the path to the installed binary.
    """
    target_dir = _ensure_target_dir()
    dest = target_dir / binary_name
    _download(url, dest)
    dest.chmod(0o755)
    return dest


def _is_installed(name: str) -> bool:
    """Check if a tool is already on PATH or in TARGET_DIR."""
    if shutil.which(name) is not None:
        return True
    return (TARGET_DIR / name).exists()


def install_actionlint() -> bool:
    """Install actionlint if not already present. Returns True if installed/skipped."""
    if _is_installed("actionlint"):
        click.echo("actionlint: already installed")
        return True
    arch = _arch()
    path = (
        f"rhysd/actionlint/releases/download/v{ACTIONLINT_VERSION}/actionlint_{ACTIONLINT_VERSION}_linux_{arch}.tar.gz"
    )
    url = f"https://github.com/{path}"
    fallback = [f"https://ghproxy.com/{path}"]
    dest = _download_and_extract_tarball(url, "actionlint", fallback_urls=fallback)
    click.echo(f"actionlint: installed to {dest}")
    return True


def install_git_cliff() -> bool:
    """Install git-cliff if not already present. Returns True if installed/skipped."""
    if _is_installed("git-cliff"):
        click.echo("git-cliff: already installed")
        return True
    # git-cliff uses x86_64/arm64 in release asset names (not amd64)
    machine = platform.machine().lower()
    arch = "x86_64" if machine in {"x86_64", "amd64"} else "arm64"
    url = (
        f"https://github.com/orhun/git-cliff/releases/download/"
        f"v{GIT_CLIFF_VERSION}/git-cliff-{GIT_CLIFF_VERSION}-{arch}-unknown-linux-gnu.tar.gz"
    )
    dest = _download_and_extract_tarball(url, "git-cliff")
    click.echo(f"git-cliff: installed to {dest}")
    return True


def install_act_runner() -> bool:
    """Install act_runner if not already present. Returns True if installed/skipped."""
    if _is_installed("act_runner"):
        click.echo("act_runner: already installed")
        return True
    arch = _arch()
    url = (
        f"https://gitea.com/gitea/act_runner/releases/download/"
        f"v{ACT_RUNNER_VERSION}/act_runner-{ACT_RUNNER_VERSION}-linux-{arch}"
    )
    dest = _download_binary(url, "act_runner")
    click.echo(f"act_runner: installed to {dest}")
    return True


def install_tea() -> bool:
    """Install tea (Gitea CLI) if not already present. Returns True if installed/skipped."""
    if _is_installed("tea"):
        click.echo("tea: already installed")
        return True
    arch = _arch()
    # dl.gitea.com is the primary CDN, but it can return 403 from some networks.
    # Fall back to the gitea.com release downloads URL.
    urls = [
        f"https://dl.gitea.com/tea/{TEA_VERSION}/tea-{TEA_VERSION}-linux-{arch}",
        f"https://gitea.com/gitea/tea/releases/download/v{TEA_VERSION}/tea-{TEA_VERSION}-linux-{arch}",
    ]
    dest = _download_with_fallback(urls, "tea")
    click.echo(f"tea: installed to {dest}")
    return True


def install_hadolint() -> bool:
    """Install hadolint if not already present. Returns True if installed/skipped."""
    if _is_installed("hadolint"):
        click.echo("hadolint: already installed")
        return True
    machine = platform.machine().lower()
    arch = "x86_64" if machine in {"x86_64", "amd64"} else "arm64"
    url = f"https://github.com/hadolint/hadolint/releases/download/v{HADOLINT_VERSION}/hadolint-Linux-{arch}"
    dest = _download_binary(url, "hadolint")
    click.echo(f"hadolint: installed to {dest}")
    return True


def install_tofu() -> bool:
    """Install OpenTofu if not already present. Returns True if installed/skipped.

    Downloads the official release tarball from GitHub and extracts the
    ``tofu`` binary to ``~/.local/bin``.
    """
    if _is_installed("tofu"):
        click.echo("tofu: already installed")
        return True
    arch = _arch()
    os_name = platform.system().lower()
    url = (
        f"https://github.com/opentofu/opentofu/releases/download/"
        f"v{TOFU_VERSION}/tofu_{TOFU_VERSION}_{os_name}_{arch}.tar.gz"
    )
    dest = _download_and_extract_tarball(url, "tofu")
    click.echo(f"tofu: installed to {dest}")
    return True


def install_vale() -> bool:
    """Install Vale (prose linter) if not already present. Returns True if installed/skipped."""
    if _is_installed("vale"):
        click.echo("vale: already installed")
        return True
    machine = platform.machine().lower()
    arch = "64-bit" if machine in {"x86_64", "amd64"} else "arm64"
    path = f"errata-ai/vale/releases/download/v{VALE_VERSION}/vale_{VALE_VERSION}_Linux_{arch}.tar.gz"
    url = f"https://github.com/{path}"
    fallback = [f"https://ghproxy.com/{path}"]
    dest = _download_and_extract_tarball(url, "vale", fallback_urls=fallback)
    click.echo(f"vale: installed to {dest}")
    return True


def install_promtool() -> bool:
    """Install promtool (Prometheus rule validator) if not already present.

    Downloads the official Prometheus release tarball from GitHub and
    extracts the ``promtool`` binary to ``~/.local/bin``.
    """
    if _is_installed("promtool"):
        click.echo("promtool: already installed")
        return True
    arch = _arch()
    url = (
        f"https://github.com/prometheus/prometheus/releases/download/"
        f"v{PROMTOOL_VERSION}/prometheus-{PROMTOOL_VERSION}.linux-{arch}.tar.gz"
    )
    dest = _download_and_extract_tarball(url, "promtool")
    click.echo(f"promtool: installed to {dest}")
    return True


TOOL_NAMES = ["actionlint", "git-cliff", "act_runner", "tea", "hadolint", "tofu", "vale", "promtool"]


def _install_tool(name: str) -> bool:
    """Install a single tool by name."""
    if name == "actionlint":
        return install_actionlint()
    if name == "git-cliff":
        return install_git_cliff()
    if name == "act_runner":
        return install_act_runner()
    if name == "tea":
        return install_tea()
    if name == "hadolint":
        return install_hadolint()
    if name == "tofu":
        return install_tofu()
    if name == "vale":
        return install_vale()
    if name == "promtool":
        return install_promtool()
    raise click.ClickException(f"Unknown tool: {name}")


def list_tools() -> None:
    """Print the installation status of all tools."""
    for name in TOOL_NAMES:
        status = "installed" if _is_installed(name) else "not installed"
        click.echo(f"  {name}: {status}")


@click.command()
@click.option(
    "--tool",
    "tools",
    multiple=True,
    type=click.Choice(TOOL_NAMES),
    help="Install specific tool(s). Can be repeated.",
)
@click.option("--list", "list_status", is_flag=True, help="List tool installation status.")
def main(tools: tuple[str, ...], list_status: bool) -> None:
    """Install CI/CD development tools to ~/.local/bin."""
    if list_status:
        list_tools()
        return

    tools_to_install = list(tools) if tools else TOOL_NAMES
    failed: list[str] = []
    for name in tools_to_install:
        try:
            _install_tool(name)
        except Exception as exc:
            click.echo(f"  {name}: FAILED — {exc}", err=True)
            failed.append(name)

    if failed:
        raise click.ClickException(f"Failed to install: {', '.join(failed)}")

    # Remind user to add ~/.local/bin to PATH if not already there
    path_env = os.environ.get("PATH", "")
    if str(TARGET_DIR) not in path_env:
        click.echo(f"\nAdd {TARGET_DIR} to your PATH to use these tools.")


if __name__ == "__main__":  # pragma: no cover
    main()  # pragma: no cover