cc-ci/runner/harness/abra.py

"""Thin, robust wrappers around the `abra` CLI for the CI harness (plan §4.3).

Bakes in the known abra gotchas (re-verify per installed abra version, currently 0.13.0-beta):
  - `abra app undeploy` / `abra app volume remove` do NOT accept `--chaos` → never pass it.
  - plumb a `timeout` through secret generate/insert/remove calls.
  - `abra app ls -S -m` returns nested {server: {apps: [...]}} — parse the inner structure.
  - run non-interactively with `-n` (`--no-input`) everywhere.
"""
from __future__ import annotations

import json
import subprocess
from typing import Optional

ABRA = "abra"


class AbraError(RuntimeError):
    pass


def _run_pty(args: list[str], timeout: int = 900, check: bool = True) -> subprocess.CompletedProcess:
    """Run abra under a pseudo-TTY (via util-linux `script`). Needed for commands that exec into
    a container interactively (backup create / restore: 'the input device is not a TTY')."""
    cmd = "abra " + " ".join(args)
    proc = subprocess.run(
        ["script", "-qec", cmd, "/dev/null"],
        capture_output=True, text=True, timeout=timeout,
    )
    if check and proc.returncode != 0:
        raise AbraError(f"[pty] {cmd} failed ({proc.returncode}):\n{proc.stdout}\n{proc.stderr}")
    return proc


def _run(args: list[str], timeout: int = 300, check: bool = True) -> subprocess.CompletedProcess:
    proc = subprocess.run(
        [ABRA, *args],
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    if check and proc.returncode != 0:
        raise AbraError(f"abra {' '.join(args)} failed ({proc.returncode}):\n{proc.stdout}\n{proc.stderr}")
    return proc


def app_new(recipe: str, domain: str, server: str = "default", version: Optional[str] = None,
            secrets: bool = False) -> None:
    args = ["app", "new", recipe]
    args += ["-s", server, "-D", domain, "-o", "-n"]
    if version:
        # pin to a published version tag (e.g. upgrade's previous-version deploy) — a clean tag
        # checkout, which is incompatible with chaos.
        args.append(version)
    else:
        # -C (chaos): deploy the recipe AT THE CURRENT CHECKOUT (the PR head under test).
        args.append("-C")
    if secrets:
        args.append("-S")
    _run(args)


def env_set(domain: str, key: str, value: str) -> None:
    """Set a key in the app's .env (abra has no setter; edit the file directly)."""
    import os
    import re
    path = os.path.expanduser(f"~/.abra/servers/default/{domain}.env")
    with open(path) as fh:
        lines = fh.read().splitlines()
    out, seen = [], False
    pat = re.compile(rf"^\s*#?\s*{re.escape(key)}=")
    for ln in lines:
        if pat.match(ln):
            out.append(f"{key}={value}")
            seen = True
        else:
            out.append(ln)
    if not seen:
        out.append(f"{key}={value}")
    with open(path, "w") as fh:
        fh.write("\n".join(out) + "\n")


def secret_generate(domain: str, timeout: int = 300) -> None:
    # -m avoids the TTY/table (ioctl) path; output (which contains the generated values) is
    # captured by _run and never logged. -C -o keep the recipe at the PR checkout (without -o it
    # re-resolves to a version tag, dropping the PR's files incl. tests/). check=False: recipes with
    # no secrets are a no-op.
    _run(["app", "secret", "generate", domain, "--all", "-m", "-C", "-o", "-n"],
         timeout=timeout, check=False)


def deploy(domain: str, chaos: bool = True, timeout: int = 900) -> None:
    args = ["app", "deploy", domain, "-o", "-n"]
    if chaos:
        args.append("-C")
    _run(args, timeout=timeout)


def upgrade(domain: str, version: Optional[str] = None, timeout: int = 900) -> None:
    args = ["app", "upgrade", domain]
    if version:
        args.append(version)
    # -f no prompt, -D skip public-DNS checks (our per-run domains route via the gateway).
    # (upgrade has no --chaos flag.)
    args += ["-f", "-D", "-n"]
    _run(args, timeout=timeout)


def backup_create(domain: str, timeout: int = 900) -> None:
    # -C -o: use the current recipe checkout, no remote fetch — like every other recipe-touching
    # call (DECISIONS.md). Without -o, abra tries to fetch recipe tags from the (possibly private)
    # remote and fails "authentication required: Unauthorized".
    _run_pty(["app", "backup", "create", domain, "-n", "-C", "-o"], timeout=timeout)


def restore(domain: str, timeout: int = 900) -> None:
    _run_pty(["app", "restore", domain, "-n", "-C", "-o"], timeout=timeout)


def recipe_versions(recipe: str) -> list[str]:
    """Published versions of a recipe, oldest→newest (from the recipe git tags)."""
    import os
    import subprocess
    path = os.path.expanduser(f"~/.abra/recipes/{recipe}")
    proc = subprocess.run(["git", "-C", path, "tag", "--sort=creatordate"],
                          capture_output=True, text=True)
    return [t for t in proc.stdout.split("\n") if t.strip()]


def undeploy(domain: str, timeout: int = 600) -> None:
    # NB: no --chaos here (unsupported).
    _run(["app", "undeploy", domain, "-n"], timeout=timeout, check=False)


def volume_remove(domain: str, timeout: int = 300) -> None:
    # NB: no --chaos here (unsupported); -f to skip prompts.
    _run(["app", "volume", "remove", domain, "-f", "-n"], timeout=timeout, check=False)


def secret_remove_all(domain: str, timeout: int = 300) -> None:
    _run(["app", "secret", "remove", domain, "--all", "-n"], timeout=timeout, check=False)


def app_config_remove(domain: str, server: str = "default") -> None:
    """Delete the app's .env config so a re-run can recreate it (teardown completeness)."""
    import os
    path = os.path.expanduser(f"~/.abra/servers/{server}/{domain}.env")
    try:
        os.remove(path)
    except FileNotFoundError:
        pass


def app_ls(server: str = "default") -> list[dict]:
    """Parse `abra app ls -S -m` nested {server: {apps: [...]}} structure."""
    proc = _run(["app", "ls", "-S", "-m", "-n"], check=False)
    try:
        data = json.loads(proc.stdout)
    except (ValueError, json.JSONDecodeError):
        return []
    node = data.get(server) or {}
    return node.get("apps", []) if isinstance(node, dict) else []