Source code for constellation_utils.cli

"""`constellation` CLI — currently just `constellation doctor` for the MVP.

Usage:
    constellation doctor [--profile testing|production]

Reads the active R2 credentials via the secrets backend and prints a
green/red status. Designed for fast feedback during onboarding and as
a smoke step in rig deploy scripts.
"""

from __future__ import annotations

import os
import sys

import tyro

from constellation_utils.secrets import ConstellationAuthError, cloudflare, r2
from constellation_utils.secrets._backends import select_backend
from constellation_utils.secrets._config import current_profile, load_profile

_GREEN = "\033[32m"
_RED = "\033[31m"
_DIM = "\033[2m"
_RESET = "\033[0m"


def _supports_color() -> bool:
    return sys.stdout.isatty() and os.environ.get("NO_COLOR") is None


def _ok(msg: str) -> str:
    return f"{_GREEN}✓{_RESET} {msg}" if _supports_color() else f"[OK]   {msg}"


def _fail(msg: str) -> str:
    return f"{_RED}✗{_RESET} {msg}" if _supports_color() else f"[FAIL] {msg}"


def _dim(msg: str) -> str:
    return f"{_DIM}{msg}{_RESET}" if _supports_color() else msg




def doctor(profile: str | None = None) -> int:
    """Smoke-test that the active profile can read its credentials.

    Args:
        profile: Override CONSTELLATION_PROFILE for this run
            (testing|production). When unset, uses the env var
            (defaults to testing).
    """
    if profile:
        os.environ["CONSTELLATION_PROFILE"] = profile

    try:
        active_profile = current_profile()
    except ValueError as exc:
        print(_fail(f"profile resolution failed: {exc}"))
        return 1

    try:
        backend = select_backend()
    except ConstellationAuthError as exc:
        print(_fail("auth backend selection failed:"))
        for line in str(exc).splitlines():
            print(f"  {line}")
        return 1

    backend_name = backend.__class__.__name__
    print(_dim(f"profile: {active_profile}    backend: {backend_name}"))

    # Force the cache miss in case `r2()` was somehow invoked earlier
    # in the same process (e.g. importing for side effects).
    r2.cache_clear()

    try:
        creds = r2()
    except ConstellationAuthError as exc:
        print(_fail("R2 credentials could not be read:"))
        for line in str(exc).splitlines():
            print(f"  {line}")
        return 1
    except Exception as exc:  # noqa: BLE001 — surface anything unexpected with context
        print(_fail(f"R2 credentials could not be read: {exc.__class__.__name__}: {exc}"))
        return 1

    masked_key = (
        creds.access_key_id[:4] + "…" + creds.access_key_id[-2:]
        if creds.access_key_id
        else "<empty>"
    )
    print(_ok(f"R2 ({active_profile}) reachable"))
    print(_dim(f"  endpoint      : {creds.endpoint}"))
    print(_dim(f"  bucket        : {creds.bucket}"))
    print(_dim(f"  region        : {creds.region}"))
    print(_dim(f"  access_key_id : {masked_key}"))

    # Cloudflare is optional in the active profile — only smoke-test
    # when the YAML maps it. Keeps doctor green for older profiles.
    cfg = load_profile(active_profile)
    if "cloudflare" in cfg:
        cloudflare.cache_clear()
        try:
            cf = cloudflare()
        except ConstellationAuthError as exc:
            print(_fail("Cloudflare credentials could not be read:"))
            for line in str(exc).splitlines():
                print(f"  {line}")
            return 1
        except Exception as exc:  # noqa: BLE001
            print(_fail(
                f"Cloudflare credentials could not be read: "
                f"{exc.__class__.__name__}: {exc}"
            ))
            return 1
        masked_token = (
            cf.api_token[:4] + "…" + cf.api_token[-4:]
            if cf.api_token
            else "<empty>"
        )
        print(_ok(f"Cloudflare ({active_profile}) reachable"))
        print(_dim(f"  account_id : {cf.account_id}"))
        print(_dim(f"  api_token  : {masked_token}"))

    return 0





def main() -> None:
    """Tyro entry point. Dispatches subcommands.

    Currently `doctor` is the only subcommand. Adding more later (e.g.
    `bootstrap`, `audit`) is a one-line change to the dict below.
    """
    rc = tyro.extras.subcommand_cli_from_dict(
        {"doctor": doctor},
        prog="constellation",
        description=__doc__,
    )
    sys.exit(rc if isinstance(rc, int) else 0)



if __name__ == "__main__":
    main()