acme-auto/tests/test_e2e.py

"""
CertCenter 端到端测试
用 Let's Encrypt Staging 环境测试完整的证书签发流程

用法:
  1. 先在 WebUI 中配置好 ACME（使用 Staging 环境 + 阿里云 DNS 凭据）
  2. 添加一台服务器和一个域名
  3. 执行: python -m tests.test_e2e --domain example.com
"""
import sys
import json
import argparse
import requests
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

BASE_URL = "http://127.0.0.1:8000"


def log(step, msg, status="info"):
    icons = {"info": "ℹ️", "ok": "✅", "fail": "❌", "warn": "⚠️"}
    print(f"  {icons.get(status, ' ')} [{step}] {msg}")


def test_api_alive():
    """测试 API 是否存活"""
    try:
        resp = requests.get(f"{BASE_URL}/admin/api/stats", timeout=5)
        return resp.status_code == 200
    except:
        return False


def get_or_create_server(name, platform, token):
    """获取或创建服务器"""
    resp = requests.get(f"{BASE_URL}/admin/api/servers")
    for s in resp.json():
        if s["name"] == name:
            return s["id"]

    resp = requests.post(f"{BASE_URL}/admin/api/servers", json={
        "name": name,
        "platform": platform,
        "token": token,
    })
    return resp.json()["id"]


def get_or_create_domain(server_id, domain, cert_dir, check_cmd, reload_cmd):
    """获取或创建域名"""
    resp = requests.get(f"{BASE_URL}/admin/api/domains")
    for d in resp.json():
        if d["domain"] == domain:
            return d["id"]

    resp = requests.post(f"{BASE_URL}/admin/api/domains", json={
        "server_id": server_id,
        "domain": domain,
        "cert_dir": cert_dir,
        "check_cmd": check_cmd,
        "reload_cmd": reload_cmd,
    })
    return resp.json()["id"]


def configure_acme(email, dns_provider, credentials):
    """配置 ACME"""
    resp = requests.put(f"{BASE_URL}/admin/api/acme/config", json={
        "acme_server": "https://acme-staging-v02.api.letsencrypt.org/directory",
        "email": email,
        "dns_provider": dns_provider,
        "dns_credentials": json.dumps(credentials),
        "renew_days": 30,
    })
    return resp.status_code == 200


def issue_certificate(domain_id):
    """申请证书"""
    resp = requests.post(f"{BASE_URL}/admin/api/acme/issue/{domain_id}", timeout=180)
    return resp.json()


def check_cert_info(domain_id):
    """检查证书信息"""
    resp = requests.get(f"{BASE_URL}/admin/api/cert-info/{domain_id}")
    return resp.json()


def test_script_download(domain, server_name):
    """测试脚本下载"""
    resp = requests.get(f"{BASE_URL}/api/script/{domain}", params={"server_name": server_name})
    return resp.status_code == 200, resp.text


def test_version_api(domain):
    """测试版本接口"""
    resp = requests.get(f"{BASE_URL}/api/version/{domain}")
    return resp.status_code == 200, resp.text


def test_cert_download(domain, token):
    """测试证书下载"""
    headers = {"Authorization": f"Bearer {token}"}
    resp = requests.get(f"{BASE_URL}/api/cert/{domain}/fullchain", headers=headers)
    return resp.status_code == 200


def main():
    parser = argparse.ArgumentParser(description="CertCenter 端到端测试")
    parser.add_argument("--domain", required=True, help="测试域名（需已使用阿里云 DNS）")
    parser.add_argument("--email", default="test@example.com", help="ACME 注册邮箱")
    parser.add_argument("--ak", required=True, help="阿里云 Access Key")
    parser.add_argument("--sk", required=True, help="阿里云 Access Secret")
    args = parser.parse_args()

    print(f"\n{'=' * 50}")
    print(f"  端到端测试: {args.domain}")
    print(f"{'=' * 50}\n")

    # Step 0: 检查 API
    log("0", "检查 API 服务...")
    if not test_api_alive():
        log("0", "API 服务未启动，请先执行 python -m backend.main", "fail")
        return 1
    log("0", "API 服务正常", "ok")

    # Step 1: 配置 ACME
    log("1", "配置 ACME (Staging 环境)...")
    credentials = {"access_key": args.ak, "access_secret": args.sk}
    if configure_acme(args.email, "aliyun", credentials):
        log("1", "ACME 配置完成", "ok")
    else:
        log("1", "ACME 配置失败", "fail")
        return 1

    # Step 2: 创建测试服务器
    log("2", "创建测试服务器...")
    test_token = "tok_e2e_test_" + str(int(time.time()))
    server_id = get_or_create_server("e2e-test", "linux", test_token)
    log("2", f"服务器 ID: {server_id}", "ok")

    # Step 3: 创建测试域名
    log("3", "创建测试域名配置...")
    cert_dir = f"/tmp/cert-e2e-test/{args.domain}"
    domain_id = get_or_create_domain(
        server_id, args.domain,
        cert_dir=cert_dir,
        check_cmd="true",  # 测试环境用 true 代替 nginx -t
        reload_cmd="true",
    )
    log("3", f"域名 ID: {domain_id}", "ok")

    # Step 4: 申请证书
    log("4", "申请证书（这需要 1-2 分钟，等待 DNS 验证）...")
    result = issue_certificate(domain_id)
    if result.get("success"):
        log("4", f"证书申请成功: {result.get('message')}", "ok")
    else:
        log("4", f"证书申请失败: {result.get('message')}", "fail")
        log("4", "请检查：1) 域名 DNS 是否托管在阿里云  2) AK/SK 是否正确  3) 网络是否通畅", "warn")
        return 1

    # Step 5: 检查证书信息
    log("5", "检查证书信息...")
    cert_info = check_cert_info(domain_id)
    if cert_info.get("cert_info"):
        info = cert_info["cert_info"]
        log("5", f"签发者: {info.get('issuer')}", "ok")
        log("5", f"有效期: {info.get('not_before')} ~ {info.get('not_after')}", "ok")
        log("5", f"域名: {info.get('san')}", "ok")
        log("5", f"剩余天数: {cert_info.get('days_left')}", "ok")
    else:
        log("5", "无法读取证书信息", "warn")

    # Step 6: 测试版本接口
    log("6", "测试版本接口...")
    ok, version = test_version_api(args.domain)
    if ok:
        log("6", f"版本号: {version}", "ok")
    else:
        log("6", "版本接口异常", "fail")

    # Step 7: 测试脚本生成
    log("7", "测试脚本生成...")
    ok, script = test_script_download(args.domain, "e2e-test")
    if ok and len(script) > 100:
        log("7", f"脚本生成成功，长度: {len(script)} 字符", "ok")
    else:
        log("7", "脚本生成失败", "fail")

    # Step 8: 测试证书下载
    log("8", "测试证书下载...")
    if test_cert_download(args.domain, test_token):
        log("8", "证书下载成功", "ok")
    else:
        log("8", "证书下载失败", "fail")

    # 完成
    print(f"\n{'=' * 50}")
    print(f"  ✅ 端到端测试完成!")
    print(f"{'=' * 50}")
    print(f"\n  证书位置: {cert_dir}/")
    print(f"  Staging 证书不受浏览器信任，仅用于验证流程。")
    print(f"  确认流程正常后，在 WebUI 中切换到生产环境重新签发。\n")

    return 0


if __name__ == "__main__":
    sys.exit(main())