autonomic-bot
38a145fd9c
All checks were successful
continuous-integration/drone/push Build is passing
run_recipe_ci.py + conftest + abra/lifecycle wrappers + Nix python/playwright env. deploy_app forces LETS_ENCRYPT_ENV='' (addresses A1). Short per-run domain scheme for the 64-char swarm name limit. 2 passed; teardown leaves zero orphans. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
"""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(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]
|
|
if version:
|
|
args.append(version)
|
|
args += ["-s", server, "-D", domain, "-n"]
|
|
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:
|
|
_run(["app", "secret", "generate", domain, "--all", "-n"], timeout=timeout, check=False)
|
|
|
|
|
|
def deploy(domain: str, chaos: bool = True, timeout: int = 900) -> None:
|
|
args = ["app", "deploy", domain, "-n"]
|
|
if chaos:
|
|
args.append("-C")
|
|
_run(args, timeout=timeout)
|
|
|
|
|
|
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 []
|