Gitea-Tools/gitea_config_menu.py

"""Interactive setup menu for the canonical Gitea MCP profile config (#31, #36).

Create / edit / remove / validate profiles, test a profile's authentication,
show the authenticated user, check reviewer eligibility for a PR, and print
thin launcher snippets for Claude / Gemini / Codex — without hand-editing JSON
and without ever putting a raw token in a launcher config.

Run:  ``./scripts/gitea-config-menu``  (or ``python gitea_config.py menu``).

UX (#36): in an interactive TTY the top-level menu and profile chooser take a
single keypress (no Enter); Enter quits the main menu and cancels/back-outs of
any submenu; the screen clears before redraws; results pause for a keypress.
All of that runs through an injectable :class:`MenuIO`, so tests exercise the
logic with scripted keys and no real terminal, keychain, or network.
"""
import sys
import json
import getpass
import urllib.parse

import gitea_config
import gitea_auth


def _host(base_url):
    """Extract the host from a profile base_url (e.g. https://gitea.x → gitea.x)."""
    netloc = urllib.parse.urlparse(base_url or "").netloc
    return netloc or (base_url or "").strip()


def resolve_identity(profile, *, resolve_token=gitea_config.resolve_token,
                     api_request=gitea_auth.api_request):
    """Return the authenticated username for *profile*, or raise ConfigError.

    Resolves the profile's auth reference to a token (keychain/env), calls the
    read-only current-user endpoint, and returns ``login``. Never returns or
    logs the token.
    """
    token = resolve_token(profile)
    if not token:
        raise gitea_config.ConfigError("no token resolved for this profile")
    host = _host(profile.get("base_url"))
    data = api_request("GET", f"https://{host}/api/v1/user", f"token {token}")
    login = (data or {}).get("login")
    if not login:
        raise gitea_config.ConfigError("could not determine authenticated user")
    return login


def check_eligibility(profile, pr_number, owner=None, repo=None, *,
                      resolve_token=gitea_config.resolve_token,
                      api_request=gitea_auth.api_request):
    """Report whether *profile* is reviewer-eligible for a PR. Read-only.

    Eligible ≙ the PR is open AND the authenticated user is NOT the PR author.
    Never approves or merges. Returns a dict with the facts and a verdict.
    """
    owner = owner or profile.get("default_owner")
    repo = repo or profile.get("default_repo")
    if not owner or not repo:
        raise gitea_config.ConfigError(
            "owner/repo required (set default_owner/default_repo or pass them)"
        )
    token = resolve_token(profile)
    if not token:
        raise gitea_config.ConfigError("no token resolved for this profile")
    auth = f"token {token}"
    host = _host(profile.get("base_url"))
    me = (api_request("GET", f"https://{host}/api/v1/user", auth) or {}).get("login")
    pr = api_request(
        "GET", f"https://{host}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}", auth
    ) or {}
    author = (pr.get("user") or {}).get("login")
    state = pr.get("state")
    reasons = []
    if not me:
        reasons.append("authenticated user unknown")
    if state != "open":
        reasons.append(f"PR is not open (state={state})")
    if me and author and me == author:
        reasons.append("authenticated user is the PR author")
    eligible = not reasons
    return {
        "authenticated_user": me,
        "pr_author": author,
        "pr_state": state,
        "eligible": eligible,
        "reasons": reasons or ["open PR and reviewer is not the author"],
    }


def launcher_snippets(profile_name, config_path):
    """Return {client: pretty-JSON snippet} for Claude / Gemini / Codex.

    All three MCP hosts consume the same server entry; the snippet carries only
    command/args + GITEA_MCP_CONFIG/GITEA_MCP_PROFILE — never a secret.
    """
    entry = gitea_config.launcher_entry(profile_name, config_path)
    claude = json.dumps({"mcpServers": entry}, indent=2)
    gemini = json.dumps({"mcpServers": entry}, indent=2)
    codex = json.dumps(entry, indent=2)
    return {"claude": claude, "gemini": gemini, "codex": codex}


# ── Injectable menu IO ──────────────────────────────────────────────────────────

CLEAR_SEQUENCE = "\x1b[2J\x1b[H"


class MenuIO:
    """Terminal IO for the menu, fully injectable for tests.

    - ``read_key(prompt)`` → one keypress in a TTY, or the first character of a
      line otherwise. Enter / EOF return ``""`` (used as cancel/back/quit).
    - ``read_line(prompt)`` → a full line (for free text). EOF returns ``""``.
    - ``out(text)`` → write a line.
    - ``clear()`` → clear the screen, but only when ``clear_enabled``.

    ``is_tty`` gates single-key input and the pause prompt; ``clear_enabled``
    gates screen clearing so automated/non-TTY runs never emit control codes.
    """

    def __init__(self, *, read_key, read_line, write, is_tty=False,
                 clear_enabled=False, clear_fn=None):
        self._read_key = read_key
        self._read_line = read_line
        self._write = write
        self.is_tty = is_tty
        self.clear_enabled = clear_enabled
        self._clear_fn = clear_fn

    def read_key(self, prompt=""):
        return self._read_key(prompt)

    def read_line(self, prompt=""):
        return self._read_line(prompt)

    def out(self, text=""):
        self._write(text)

    def clear(self):
        if self.clear_enabled and self._clear_fn:
            self._clear_fn()


def _read_key_tty():
    """Read a single keypress from a real TTY without requiring Enter.

    Returns ``""`` for Enter, raises KeyboardInterrupt on Ctrl-C. macOS/Linux
    only (termios/tty); callers use a line-input fallback when not a TTY.
    """
    import termios
    import tty
    fd = sys.stdin.fileno()
    old = termios.tcgetattr(fd)
    try:
        tty.setraw(fd)
        ch = sys.stdin.read(1)
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, old)
    if ch in ("\r", "\n", ""):
        return ""
    if ch == "\x03":
        raise KeyboardInterrupt
    return ch


def default_io():
    """Build the real MenuIO: single-key in a TTY, line-input otherwise."""
    is_tty = bool(getattr(sys.stdin, "isatty", lambda: False)()) and \
        bool(getattr(sys.stdout, "isatty", lambda: False)())

    def write(text=""):
        print(text)

    def read_key(prompt=""):
        if prompt:
            sys.stdout.write(prompt)
            sys.stdout.flush()
        if is_tty:
            key = _read_key_tty()
            print()  # move off the prompt line
            return key
        line = sys.stdin.readline()
        if line == "":
            return ""  # EOF → back/quit
        return line.rstrip("\r\n")[:1]

    def read_line(prompt=""):
        try:
            return input(prompt)
        except EOFError:
            return ""

    def clear():
        sys.stdout.write(CLEAR_SEQUENCE)
        sys.stdout.flush()

    return MenuIO(read_key=read_key, read_line=read_line, write=write,
                  is_tty=is_tty, clear_enabled=is_tty, clear_fn=clear)


# ── UX helpers ──────────────────────────────────────────────────────────────────

_MENU_LINES = [
    "Gitea MCP profile setup",
    "  1) list profiles",
    "  2) add profile",
    "  3) edit profile",
    "  4) remove profile",
    "  5) validate config",
    "  6) test profile authentication",
    "  7) show authenticated user",
    "  8) generate launcher snippets (Claude/Gemini/Codex)",
    "  9) check reviewer eligibility for a PR",
    "  (Enter or 0) quit",
]


def clear_screen(io):
    """Clear the terminal (no-op unless the IO has clearing enabled)."""
    io.clear()


def pause_for_key(io):
    """Wait for a keypress in a TTY; return immediately otherwise (never hangs)."""
    if io.is_tty:
        io.out("Press any key to continue")
        io.read_key()


def choose_menu_option(io):
    """Clear, draw the main menu, and return the pressed key ('' = quit)."""
    clear_screen(io)
    for line in _MENU_LINES:
        io.out(line)
    return io.read_key("choice: ")


def choose_profile(io, config, purpose="profile"):
    """Show a numbered profile list and return the chosen name, or None.

    Enter cancels (returns None). Digits 1-9 select from the list. 'm' opens an
    explicit manual-entry path. Empty config is handled gracefully.
    """
    names = sorted((config.get("profiles") or {}))
    clear_screen(io)
    if not names:
        io.out("(no profiles yet — add one first)")
        pause_for_key(io)
        return None
    io.out(f"Select {purpose} (Enter to cancel):")
    for i, name in enumerate(names[:9], start=1):
        io.out(f"  {i}) {name}")
    io.out("  m) type a name manually")
    key = io.read_key("choose: ")
    if key == "":
        return None
    if key == "m":
        typed = io.read_line("profile name: ").strip()
        return typed or None
    if key.isdigit():
        idx = int(key)
        if 1 <= idx <= min(len(names), 9):
            return names[idx - 1]
    io.out("invalid selection")
    pause_for_key(io)
    return None


def _prompt_profile_fields(io, secret_fn, keychain_set, name):
    """Prompt for a profile's fields for *name*. Returns a profile dict or None.

    A blank Base URL cancels. Tokens are read with the secret prompt (no echo)
    and stored in the keychain; they are never placed in the returned profile.
    """
    base_url = io.read_line("Base URL (e.g. https://gitea.prgs.cc; Enter cancels): ").strip()
    if not base_url:
        return None
    username = io.read_line("Gitea username: ").strip()
    default_owner = io.read_line("Default owner (optional): ").strip()
    default_repo = io.read_line("Default repo (optional): ").strip()
    execution_profile = io.read_line("Execution profile (optional): ").strip()
    auth_type = io.read_line("Auth type [keychain/env]: ").strip().lower()

    if auth_type == "keychain":
        default_id = f"{name}-gitea-token"
        item_id = io.read_line(f"Keychain item id [{default_id}]: ").strip() or default_id
        store = io.read_line("Store a token in the keychain now? [y/N]: ").strip().lower()
        if store == "y":
            token = secret_fn("Gitea token (hidden): ")
            keychain_set(item_id, token)  # token never echoed
            io.out(f"Stored token in keychain item '{item_id}'.")
        auth = gitea_config.keychain_auth(item_id)
    elif auth_type == "env":
        default_var = f"GITEA_TOKEN_{name.upper().replace('-', '_').replace('.', '_')}"
        var = io.read_line(f"Env var name [{default_var}]: ").strip() or default_var
        io.out(f"Set {var} in the environment; its value is never stored here.")
        auth = gitea_config.env_auth(var)
    else:
        raise gitea_config.ConfigError("auth type must be 'keychain' or 'env'")

    return gitea_config.build_profile(
        base_url=base_url, auth=auth, username=username,
        default_owner=default_owner, default_repo=default_repo,
        execution_profile=execution_profile,
    )


def main(argv=None, *, io=None, secret_fn=None, config_path=None,
         resolve_token=gitea_config.resolve_token,
         api_request=gitea_auth.api_request,
         keychain_set=gitea_config.keychain_set):
    """Run the interactive menu loop. Returns a process exit code."""
    io = io or default_io()
    secret_fn = secret_fn or getpass.getpass
    path = config_path or gitea_config.config_path() or gitea_config.DEFAULT_CONFIG_PATH

    try:
        config = gitea_config.load_config(path) or gitea_config.empty_config()
    except gitea_config.ConfigError as exc:
        io.out(f"config error: {exc}")
        config = gitea_config.empty_config()

    while True:
        key = choose_menu_option(io)
        if key in ("", "0"):  # Enter or 0 → quit
            io.out("bye")
            return 0
        try:
            if key == "1":
                names = sorted(config.get("profiles") or {})
                clear_screen(io)
                io.out("profiles: " + (", ".join(names) if names else "(none)"))
                pause_for_key(io)
            elif key == "2":
                name = io.read_line("New profile name (Enter to cancel): ").strip()
                if not name:
                    continue
                if not gitea_config.is_valid_profile_name(name):
                    io.out(f"invalid profile name {name!r}")
                    pause_for_key(io)
                    continue
                profile = _prompt_profile_fields(io, secret_fn, keychain_set, name)
                if profile is None:
                    continue
                config = gitea_config.add_profile(config, name, profile)
                saved = gitea_config.save_config(config, path)
                io.out(f"added '{name}'; wrote {saved}")
                pause_for_key(io)
            elif key == "3":
                name = choose_profile(io, config, "profile to edit")
                if not name:
                    continue
                profile = _prompt_profile_fields(io, secret_fn, keychain_set, name)
                if profile is None:
                    continue
                config = gitea_config.upsert_profile(config, name, profile)
                saved = gitea_config.save_config(config, path)
                io.out(f"saved '{name}'; wrote {saved}")
                pause_for_key(io)
            elif key == "4":
                name = choose_profile(io, config, "profile to remove")
                if not name:
                    continue
                config = gitea_config.remove_profile(config, name)
                saved = gitea_config.save_config(config, path)
                io.out(f"removed '{name}'; wrote {saved}")
                pause_for_key(io)
            elif key == "5":
                problems = gitea_config.validate_config(config)
                clear_screen(io)
                io.out("valid" if not problems
                       else "problems:\n  " + "\n  ".join(problems))
                pause_for_key(io)
            elif key in ("6", "7"):
                name = choose_profile(io, config, "profile")
                if not name:
                    continue
                profile = gitea_config.select_profile(config, name)
                who = resolve_identity(profile, resolve_token=resolve_token,
                                       api_request=api_request)
                io.out(f"authenticated as: {who}")
                pause_for_key(io)
            elif key == "8":
                name = choose_profile(io, config, "profile")
                if not name:
                    continue
                clear_screen(io)
                for client, snippet in launcher_snippets(name, path).items():
                    io.out(f"--- {client} ---\n{snippet}")
                pause_for_key(io)
            elif key == "9":
                name = choose_profile(io, config, "profile")
                if not name:
                    continue
                pr_number = io.read_line("PR number (Enter to cancel): ").strip()
                if not pr_number:
                    continue
                profile = gitea_config.select_profile(config, name)
                result = check_eligibility(
                    profile, pr_number, resolve_token=resolve_token,
                    api_request=api_request)
                io.out(f"authenticated user: {result['authenticated_user']}")
                io.out(f"PR author:          {result['pr_author']}")
                io.out("ELIGIBLE" if result["eligible"] else "INELIGIBLE")
                for r in result["reasons"]:
                    io.out(f"  - {r}")
                pause_for_key(io)
            else:
                io.out("unknown choice")
                pause_for_key(io)
        except gitea_config.ConfigError as exc:
            io.out(f"error: {exc}")
            pause_for_key(io)
        except Exception as exc:  # noqa: BLE001 - keep the menu alive, no secrets
            io.out(f"error: {type(exc).__name__}")
            pause_for_key(io)


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main(sys.argv[1:]))