"""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)}")