e2e test
This commit is contained in:
@@ -0,0 +1,265 @@
|
||||
"""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)}")
|
||||
@@ -0,0 +1,225 @@
|
||||
"""End-to-end tests for tools/app.sh against real services.
|
||||
|
||||
Test domain: kgfamily.com
|
||||
Run: uvx pytest tests/test_app_e2e.py -v -x
|
||||
"""
|
||||
|
||||
import os
|
||||
import ssl
|
||||
import subprocess
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
import pytest
|
||||
|
||||
from conftest import (
|
||||
TEST_DOMAIN,
|
||||
CONTAINER_NAME,
|
||||
REPO_ROOT,
|
||||
VULTR_API,
|
||||
GITEA_URL,
|
||||
GITEA_ORG,
|
||||
GARAGE_API,
|
||||
UPTIMEROBOT_API,
|
||||
api,
|
||||
run_app,
|
||||
cleanup,
|
||||
)
|
||||
|
||||
|
||||
class TestAppSubcommands:
|
||||
"""Test each app.sh subcommand individually, in order."""
|
||||
|
||||
@pytest.fixture(autouse=True, scope="class")
|
||||
def _cleanup(self, secrets):
|
||||
cleanup(secrets, revert_git=True)
|
||||
yield
|
||||
cleanup(secrets, revert_git=False)
|
||||
|
||||
def test_dns(self, secrets):
|
||||
result = run_app("dns")
|
||||
assert result.returncode == 0, f"app.sh dns failed:\n{result.stdout}\n{result.stderr}"
|
||||
|
||||
auth = {"Authorization": f"Bearer {secrets['hantim-vultr-api-key']}"}
|
||||
status, data = api("GET", f"{VULTR_API}/domains/{TEST_DOMAIN}/records", headers=auth)
|
||||
assert status == 200, f"Vultr API returned {status}"
|
||||
|
||||
records = data["records"]
|
||||
bare_a = [r for r in records if r["type"] == "A" and r["name"] == ""]
|
||||
www_a = [r for r in records if r["type"] == "A" and r["name"] == "www"]
|
||||
|
||||
assert len(bare_a) == 1, f"Expected 1 bare A record, got {len(bare_a)}"
|
||||
assert len(www_a) == 1, f"Expected 1 www A record, got {len(www_a)}"
|
||||
assert bare_a[0]["data"] == www_a[0]["data"], "Bare and www should point to same IP"
|
||||
|
||||
def test_repo(self, secrets):
|
||||
result = run_app("repo")
|
||||
assert result.returncode == 0, f"app.sh repo failed:\n{result.stdout}\n{result.stderr}"
|
||||
|
||||
auth = {"Authorization": f"token {secrets['hantim-new-app-script']}"}
|
||||
status, data = api("GET", f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{TEST_DOMAIN}", headers=auth)
|
||||
assert status == 200, f"Repo not found (HTTP {status})"
|
||||
assert data["name"] == TEST_DOMAIN
|
||||
|
||||
def test_files(self):
|
||||
result = run_app("files")
|
||||
assert result.returncode == 0, f"app.sh files failed:\n{result.stdout}\n{result.stderr}"
|
||||
|
||||
# compose.yml
|
||||
compose_path = os.path.join(REPO_ROOT, "docker", TEST_DOMAIN, "compose.yml")
|
||||
assert os.path.isfile(compose_path)
|
||||
compose = open(compose_path).read()
|
||||
assert f"image: git.timothykim.net/hantim/{TEST_DOMAIN}:latest" in compose
|
||||
assert f"container_name: {CONTAINER_NAME}" in compose
|
||||
assert "shared" in compose
|
||||
|
||||
# deploy workflow
|
||||
workflow_path = os.path.join(
|
||||
REPO_ROOT, ".gitea", "workflows", f"deploy-{TEST_DOMAIN}.yml"
|
||||
)
|
||||
assert os.path.isfile(workflow_path)
|
||||
workflow = open(workflow_path).read()
|
||||
assert f"docker/{TEST_DOMAIN}/**" in workflow
|
||||
assert f"deploy-{TEST_DOMAIN}" in workflow
|
||||
|
||||
# nginx conf
|
||||
conf_path = os.path.join(
|
||||
REPO_ROOT, "docker", "nginx", "conf.d", f"{TEST_DOMAIN}.conf"
|
||||
)
|
||||
assert os.path.isfile(conf_path)
|
||||
conf = open(conf_path).read()
|
||||
assert f"server_name {TEST_DOMAIN} www.{TEST_DOMAIN}" in conf
|
||||
assert f"server_name www.{TEST_DOMAIN}" in conf
|
||||
assert "security-headers.inc" in conf
|
||||
assert f"proxy_pass http://{CONTAINER_NAME}:80" in conf
|
||||
|
||||
def test_cert(self):
|
||||
result = run_app("cert", timeout=180)
|
||||
assert result.returncode == 0, f"app.sh cert failed:\n{result.stdout}\n{result.stderr}"
|
||||
|
||||
def test_garage(self, secrets):
|
||||
result = run_app("garage")
|
||||
assert result.returncode == 0, f"app.sh garage failed:\n{result.stdout}\n{result.stderr}"
|
||||
|
||||
auth = {
|
||||
"Authorization": f"Bearer {secrets['hantim-garage-admin-token']}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
status, data = api(
|
||||
"POST",
|
||||
f"{GARAGE_API}/v2/GetBucketInfo",
|
||||
headers=auth,
|
||||
data={"globalAlias": TEST_DOMAIN},
|
||||
)
|
||||
assert status == 200, f"Bucket not found (HTTP {status})"
|
||||
assert data["id"], "Bucket has no ID"
|
||||
assert TEST_DOMAIN in data.get("globalAliases", []), "Global alias not set"
|
||||
website = data.get("websiteAccess", {})
|
||||
assert website.get("enabled") is True, f"Website access not enabled: {website}"
|
||||
|
||||
def test_build(self, secrets):
|
||||
result = run_app("build", timeout=180)
|
||||
assert result.returncode == 0, f"app.sh build failed:\n{result.stdout}\n{result.stderr}"
|
||||
|
||||
def test_monitor(self, secrets):
|
||||
result = run_app("monitor")
|
||||
assert result.returncode == 0, f"app.sh monitor failed:\n{result.stdout}\n{result.stderr}"
|
||||
|
||||
auth = {"Authorization": f"Bearer {secrets['hantim-uptimerobot-api-key']}"}
|
||||
status, data = api(
|
||||
"GET",
|
||||
f"{UPTIMEROBOT_API}/monitors?"
|
||||
+ urllib.parse.urlencode({"search": TEST_DOMAIN}),
|
||||
headers=auth,
|
||||
)
|
||||
assert status == 200, f"UptimeRobot API returned {status}"
|
||||
urls = [m["url"] for m in data.get("monitors", [])]
|
||||
assert f"https://www.{TEST_DOMAIN}" in urls, f"Monitor not found. URLs: {urls}"
|
||||
|
||||
|
||||
class TestAppAll:
|
||||
"""Test the full app.sh all flow end-to-end."""
|
||||
|
||||
@pytest.fixture(autouse=True, scope="class")
|
||||
def _cleanup(self, secrets):
|
||||
cleanup(secrets, revert_git=True)
|
||||
yield
|
||||
cleanup(secrets, revert_git=True)
|
||||
|
||||
def test_all(self, secrets):
|
||||
result = run_app("all", timeout=900)
|
||||
assert result.returncode == 0, f"app.sh all failed:\n{result.stdout}\n{result.stderr}"
|
||||
|
||||
# --- DNS ---
|
||||
auth = {"Authorization": f"Bearer {secrets['hantim-vultr-api-key']}"}
|
||||
status, data = api(
|
||||
"GET", f"{VULTR_API}/domains/{TEST_DOMAIN}/records", headers=auth
|
||||
)
|
||||
assert status == 200
|
||||
records = data["records"]
|
||||
assert any(r["type"] == "A" and r["name"] == "" for r in records)
|
||||
assert any(r["type"] == "A" and r["name"] == "www" for r in records)
|
||||
|
||||
# --- Gitea repo ---
|
||||
gitea_auth = {"Authorization": f"token {secrets['hantim-new-app-script']}"}
|
||||
status, _ = api(
|
||||
"GET",
|
||||
f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{TEST_DOMAIN}",
|
||||
headers=gitea_auth,
|
||||
)
|
||||
assert status == 200, "Gitea repo not found"
|
||||
|
||||
# --- Local files ---
|
||||
assert os.path.isfile(
|
||||
os.path.join(REPO_ROOT, "docker", TEST_DOMAIN, "compose.yml")
|
||||
)
|
||||
assert os.path.isfile(
|
||||
os.path.join(REPO_ROOT, ".gitea", "workflows", f"deploy-{TEST_DOMAIN}.yml")
|
||||
)
|
||||
assert os.path.isfile(
|
||||
os.path.join(REPO_ROOT, "docker", "nginx", "conf.d", f"{TEST_DOMAIN}.conf")
|
||||
)
|
||||
|
||||
# --- Git commit was pushed ---
|
||||
log = subprocess.run(
|
||||
["git", "log", "--format=%s", "-1"],
|
||||
capture_output=True, text=True, cwd=REPO_ROOT,
|
||||
)
|
||||
assert log.stdout.strip() == f"add {TEST_DOMAIN}"
|
||||
|
||||
# --- Garage bucket ---
|
||||
garage_auth = {
|
||||
"Authorization": f"Bearer {secrets['hantim-garage-admin-token']}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
status, data = api(
|
||||
"POST",
|
||||
f"{GARAGE_API}/v2/GetBucketInfo",
|
||||
headers=garage_auth,
|
||||
data={"globalAlias": TEST_DOMAIN},
|
||||
)
|
||||
assert status == 200, "Garage bucket not found"
|
||||
|
||||
# --- Site is live ---
|
||||
ctx = ssl.create_default_context()
|
||||
req = urllib.request.Request(f"https://www.{TEST_DOMAIN}", method="HEAD")
|
||||
try:
|
||||
resp = urllib.request.urlopen(req, context=ctx, timeout=10)
|
||||
assert resp.status in (200, 301, 302)
|
||||
except urllib.error.HTTPError:
|
||||
pass # any HTTP response means the site is up
|
||||
except urllib.error.URLError as e:
|
||||
pytest.fail(f"Site not reachable: {e}")
|
||||
|
||||
# --- UptimeRobot monitor ---
|
||||
ur_auth = {"Authorization": f"Bearer {secrets['hantim-uptimerobot-api-key']}"}
|
||||
status, data = api(
|
||||
"GET",
|
||||
f"{UPTIMEROBOT_API}/monitors?"
|
||||
+ urllib.parse.urlencode({"search": TEST_DOMAIN}),
|
||||
headers=ur_auth,
|
||||
)
|
||||
assert status == 200
|
||||
urls = [m["url"] for m in data.get("monitors", [])]
|
||||
assert f"https://www.{TEST_DOMAIN}" in urls
|
||||
Reference in New Issue
Block a user