"""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 subprocess import time 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']}", } status, data = api( "GET", f"{GARAGE_API}/v2/GetBucketInfo?" + urllib.parse.urlencode({"globalAlias": TEST_DOMAIN}), headers=auth, ) 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" assert data.get("websiteAccess") is True, f"Website access not enabled: {data.get('websiteAccess')}" 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}" # UptimeRobot API may have brief eventual consistency after create auth = {"Authorization": f"Bearer {secrets['hantim-uptimerobot-api-key']}"} target = f"https://www.{TEST_DOMAIN}" for attempt in range(3): status, data = api( "GET", f"{UPTIMEROBOT_API}/monitors", headers=auth, ) assert status == 200, f"UptimeRobot API returned {status}" urls = [m["url"] for m in data.get("data", [])] if target in urls: break time.sleep(2) assert target in urls, ( f"Monitor not found. URLs: {urls}\n" f"app.sh output:\n{result.stdout}" ) 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']}", } status, data = api( "GET", f"{GARAGE_API}/v2/GetBucketInfo?" + urllib.parse.urlencode({"globalAlias": TEST_DOMAIN}), headers=garage_auth, ) assert status == 200, "Garage bucket not found" # --- Site is live --- # Use server IP from verified DNS records to bypass negative cache server_ip = next( r["data"] for r in records if r["type"] == "A" and r["name"] == "" ) curl = subprocess.run( ["curl", "-sf", "--resolve", f"www.{TEST_DOMAIN}:443:{server_ip}", "-o", "/dev/null", "-w", "%{http_code}", f"https://www.{TEST_DOMAIN}"], capture_output=True, text=True, timeout=10, ) assert curl.returncode == 0, f"Site not reachable (HTTP {curl.stdout})" # --- UptimeRobot monitor --- ur_auth = {"Authorization": f"Bearer {secrets['hantim-uptimerobot-api-key']}"} target = f"https://www.{TEST_DOMAIN}" for attempt in range(3): status, data = api( "GET", f"{UPTIMEROBOT_API}/monitors", headers=ur_auth, ) assert status == 200 urls = [m["url"] for m in data.get("data", [])] if target in urls: break time.sleep(2) assert target in urls