226 lines
8.2 KiB
Python
226 lines
8.2 KiB
Python
"""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
|