M4: harness + green install stage (custom-html + Playwright); guaranteed teardown; M4 CLAIMED
All checks were successful
continuous-integration/drone/push Build is passing
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>
This commit is contained in:
109
runner/harness/abra.py
Normal file
109
runner/harness/abra.py
Normal file
@ -0,0 +1,109 @@
|
||||
"""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 []
|
||||
Reference in New Issue
Block a user