266 lines
8.5 KiB
Python
266 lines
8.5 KiB
Python
"""Shared fixtures and helpers for e2e tests."""
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
import pytest
|
|
|
|
TEST_DOMAIN = "kgfamily.com"
|
|
CONTAINER_NAME = TEST_DOMAIN.replace(".", "_")
|
|
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
APP_SH = os.path.join(REPO_ROOT, "tools", "app.sh")
|
|
|
|
VULTR_API = "https://api.vultr.com/v2"
|
|
GITEA_URL = "https://git.timothykim.net"
|
|
GITEA_ORG = "hantim"
|
|
GARAGE_API = "https://garage.hantim.net"
|
|
UPTIMEROBOT_API = "https://api.uptimerobot.com/v3"
|
|
|
|
|
|
def api(method, url, headers=None, data=None):
|
|
"""HTTP request helper. Returns (status_code, parsed_json_or_None)."""
|
|
hdrs = dict(headers or {})
|
|
body = None
|
|
if data is not None:
|
|
body = json.dumps(data).encode()
|
|
hdrs.setdefault("Content-Type", "application/json")
|
|
req = urllib.request.Request(url, method=method, headers=hdrs, data=body)
|
|
try:
|
|
resp = urllib.request.urlopen(req, timeout=30)
|
|
raw = resp.read().decode()
|
|
return resp.status, json.loads(raw) if raw.strip() else None
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read().decode()
|
|
try:
|
|
return e.code, json.loads(raw) if raw.strip() else None
|
|
except json.JSONDecodeError:
|
|
return e.code, None
|
|
|
|
|
|
def _bws_env():
|
|
"""Return env dict with BWS_ACCESS_TOKEN set from token file if needed."""
|
|
env = os.environ.copy()
|
|
if not env.get("BWS_ACCESS_TOKEN") and os.path.isfile(BWS_TOKEN_FILE):
|
|
env["BWS_ACCESS_TOKEN"] = open(BWS_TOKEN_FILE).read().strip()
|
|
return env
|
|
|
|
|
|
def run_app(command, domain=TEST_DOMAIN, timeout=300):
|
|
"""Run app.sh with given command and domain. Returns CompletedProcess."""
|
|
return subprocess.run(
|
|
[APP_SH, command, domain],
|
|
capture_output=True, text=True, timeout=timeout,
|
|
cwd=REPO_ROOT,
|
|
)
|
|
|
|
|
|
BWS_TOKEN_FILE = os.path.join(
|
|
os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")),
|
|
"hantim", "bws-token",
|
|
)
|
|
|
|
REQUIRED_TOOLS = ["bws", "jq", "dig", "ssh", "curl", "git"]
|
|
|
|
REQUIRED_SECRETS = [
|
|
"hantim-vultr-api-key",
|
|
"hantim-new-app-script",
|
|
"hantim-deploy-ssh-private-key",
|
|
"hantim-garage-admin-token",
|
|
"hantim-garage-media-key-id",
|
|
"hantim-uptimerobot-api-key",
|
|
]
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def preflight():
|
|
"""Check that all required tools and credentials are available."""
|
|
errors = []
|
|
|
|
for tool in REQUIRED_TOOLS:
|
|
if shutil.which(tool) is None:
|
|
errors.append(f"missing CLI tool: {tool}")
|
|
|
|
if not os.environ.get("BWS_ACCESS_TOKEN") and not os.path.isfile(BWS_TOKEN_FILE):
|
|
errors.append(
|
|
f"no bws token: set BWS_ACCESS_TOKEN or create {BWS_TOKEN_FILE}"
|
|
)
|
|
|
|
if errors:
|
|
pytest.exit(
|
|
"Preflight failed:\n " + "\n ".join(errors),
|
|
returncode=1,
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def secrets(preflight):
|
|
"""Fetch all bws secrets once for the session."""
|
|
result = subprocess.run(
|
|
["bws", "secret", "list"],
|
|
capture_output=True, text=True, env=_bws_env(),
|
|
)
|
|
if result.returncode != 0:
|
|
pytest.exit(
|
|
f"bws secret list failed (is the token valid?):\n{result.stderr}",
|
|
returncode=1,
|
|
)
|
|
|
|
all_secrets = json.loads(result.stdout)
|
|
by_key = {s["key"]: s["value"] for s in all_secrets}
|
|
|
|
missing = [k for k in REQUIRED_SECRETS if k not in by_key]
|
|
if missing:
|
|
pytest.exit(
|
|
f"Missing secrets in Bitwarden: {', '.join(missing)}",
|
|
returncode=1,
|
|
)
|
|
|
|
return by_key
|
|
|
|
|
|
def cleanup(secrets, revert_git=False):
|
|
"""Best-effort cleanup of all TEST_DOMAIN resources."""
|
|
errors = []
|
|
|
|
# --- UptimeRobot monitor ---
|
|
try:
|
|
key = secrets["hantim-uptimerobot-api-key"]
|
|
auth = {"Authorization": f"Bearer {key}"}
|
|
status, data = api(
|
|
"GET",
|
|
f"{UPTIMEROBOT_API}/monitors?"
|
|
+ urllib.parse.urlencode({"search": TEST_DOMAIN}),
|
|
headers=auth,
|
|
)
|
|
if status == 200 and data:
|
|
for m in data.get("monitors", []):
|
|
if m.get("url") == f"https://www.{TEST_DOMAIN}":
|
|
api("DELETE", f"{UPTIMEROBOT_API}/monitors/{m['id']}", headers=auth)
|
|
except Exception as e:
|
|
errors.append(f"UptimeRobot: {e}")
|
|
|
|
# --- Garage bucket ---
|
|
try:
|
|
token = secrets["hantim-garage-admin-token"]
|
|
auth = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
|
status, data = api(
|
|
"POST",
|
|
f"{GARAGE_API}/v2/GetBucketInfo",
|
|
headers=auth,
|
|
data={"globalAlias": TEST_DOMAIN},
|
|
)
|
|
if status == 200 and data and data.get("id"):
|
|
bid = data["id"]
|
|
# Deny media key access
|
|
media_key = secrets.get("hantim-garage-media-key-id", "")
|
|
if media_key:
|
|
api(
|
|
"POST",
|
|
f"{GARAGE_API}/v2/DenyBucketKey",
|
|
headers=auth,
|
|
data={
|
|
"bucketId": bid,
|
|
"accessKeyId": media_key,
|
|
"permissions": {"read": True, "write": True, "owner": False},
|
|
},
|
|
)
|
|
# Remove global alias
|
|
api(
|
|
"POST",
|
|
f"{GARAGE_API}/v2/RemoveBucketAlias",
|
|
headers=auth,
|
|
data={"bucketId": bid, "globalAlias": TEST_DOMAIN},
|
|
)
|
|
# Delete bucket (id is a query param, not body)
|
|
api(
|
|
"POST",
|
|
f"{GARAGE_API}/v2/DeleteBucket?"
|
|
+ urllib.parse.urlencode({"id": bid}),
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
except Exception as e:
|
|
errors.append(f"Garage: {e}")
|
|
|
|
# --- Gitea repo ---
|
|
try:
|
|
token = secrets["hantim-new-app-script"]
|
|
api(
|
|
"DELETE",
|
|
f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{TEST_DOMAIN}",
|
|
headers={"Authorization": f"token {token}"},
|
|
)
|
|
except Exception as e:
|
|
errors.append(f"Gitea: {e}")
|
|
|
|
# --- Local files ---
|
|
local_paths = [
|
|
os.path.join(REPO_ROOT, "docker", TEST_DOMAIN),
|
|
os.path.join(REPO_ROOT, ".gitea", "workflows", f"deploy-{TEST_DOMAIN}.yml"),
|
|
os.path.join(REPO_ROOT, "docker", "nginx", "conf.d", f"{TEST_DOMAIN}.conf"),
|
|
]
|
|
|
|
if revert_git:
|
|
# Files may be tracked — check if last commit added them
|
|
try:
|
|
log = subprocess.run(
|
|
["git", "log", "--format=%s", "-1"],
|
|
capture_output=True, text=True, cwd=REPO_ROOT,
|
|
)
|
|
if log.stdout.strip() == f"add {TEST_DOMAIN}":
|
|
removed_any = False
|
|
for f in [
|
|
f"docker/{TEST_DOMAIN}",
|
|
f".gitea/workflows/deploy-{TEST_DOMAIN}.yml",
|
|
f"docker/nginx/conf.d/{TEST_DOMAIN}.conf",
|
|
]:
|
|
full = os.path.join(REPO_ROOT, f)
|
|
if os.path.exists(full):
|
|
subprocess.run(
|
|
["git", "rm", "-rf", f],
|
|
cwd=REPO_ROOT,
|
|
capture_output=True,
|
|
)
|
|
removed_any = True
|
|
if removed_any:
|
|
subprocess.run(
|
|
["git", "commit", "-m", f"remove {TEST_DOMAIN}"],
|
|
cwd=REPO_ROOT, check=True, capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "push"],
|
|
cwd=REPO_ROOT, check=True, capture_output=True,
|
|
)
|
|
except Exception as e:
|
|
errors.append(f"Git revert: {e}")
|
|
|
|
# Always try to remove untracked local files too
|
|
for path in local_paths:
|
|
try:
|
|
if os.path.isdir(path):
|
|
shutil.rmtree(path)
|
|
elif os.path.isfile(path):
|
|
os.remove(path)
|
|
except FileNotFoundError:
|
|
pass
|
|
except Exception as e:
|
|
errors.append(f"Local {path}: {e}")
|
|
|
|
# --- Vultr DNS zone ---
|
|
try:
|
|
key = secrets["hantim-vultr-api-key"]
|
|
api(
|
|
"DELETE",
|
|
f"{VULTR_API}/domains/{TEST_DOMAIN}",
|
|
headers={"Authorization": f"Bearer {key}"},
|
|
)
|
|
except Exception as e:
|
|
errors.append(f"Vultr: {e}")
|
|
|
|
if errors:
|
|
print(f"Cleanup warnings: {'; '.join(errors)}")
|