#!/usr/bin/env python3 """6/20 部署与运维验证 — 真实 boot 起来, 跑 8 项 production launch checklist. 不依赖 systemd (开发机可能没装). 用 subprocess 启动 admin, 验证: 1. GAOKAO_ENV=dev → /health 200 2. /health checks.db_writable / disk_writable / settings_valid 全部 True 3. /api/auth/login 错误密码 401 4. /api/auth/login 正确密码 200 5. /api/orders 无 token 401 6. /api/orders 错误 token 401 7. /api/orders 有效 token 200 [] 8. /openapi.json 200 + 含 /health + /api/orders + /api/auth/login 9. ops-alerts 文件路径可写 10. /api/admin/ops-alerts 200 (admin 鉴权) 11. /admin/dashboard 200 (含 footer 链接) 输出: reports/deploy_ops_2026_06_20.json + 状态表格 """ from __future__ import annotations import http.client import json import os import subprocess import sys import time from pathlib import Path REPO = Path(__file__).resolve().parent.parent PY = REPO / ".venv" / "bin" / "python" LOG = Path("/tmp/deploy-ops-admin.log") RESULTS = REPO / "reports" / "deploy_ops_2026_06_20.json" PORT = int(os.environ.get("GAOKAO_DEPLOY_PORT", "19093")) def _http( method: str, path: str, *, body: dict | None = None, token: str | None = None, port: int = PORT, ) -> tuple[int, dict, str]: conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10) headers = {"Content-Type": "application/json"} if body else {} if token: headers["Authorization"] = f"Bearer {token}" headers["Connection"] = "close" payload = json.dumps(body).encode() if body else b"" conn.request(method, path, body=payload, headers=headers) resp = conn.getresponse() raw = resp.getheaders() body_text = resp.read().decode("utf-8", "replace") conn.close() return resp.status, {k.lower(): v for k, v in raw}, body_text def _start_admin() -> subprocess.Popen: env = os.environ.copy() env.update({ "GAOKAO_ENV": "dev", "GAOKAO_DB_PATH": str(REPO / f"data/orders/deploy-admin-{PORT}.db"), "GAOKAO_ORDERS_DB_PATH": str(REPO / f"data/orders-deploy-{PORT}.db"), "GAOKAO_SHARE_DB_PATH": str(REPO / f"data/share-deploy-{PORT}.db"), "GAOKAO_SHARE_REPORT_DIR": str(REPO / f"data/share-reports-deploy-{PORT}"), "GAOKAO_OPS_ALERT_LOG": str(REPO / f"data/alerts/deploy-ops-{PORT}.jsonl"), "GAOKAO_RETENTION_DAYS": "180", "GAOKAO_JWT_SECRET": "x" * 32, "GAOKAO_PORTAL_TOKEN_SECRET": "y" * 32, "GAOKAO_ADMIN_USER": "admin", "GAOKAO_ADMIN_PASS": "DeployOpsTest1!", "GAOKAO_PAYMENT_WEBHOOK_SECRET": "P" + "r" * 31 + "!", "GAOKAO_PAYMENT_PROVIDER": "mock", "GAOKAO_ADMIN_BIND": f"127.0.0.1:{PORT}", }) for d in [ REPO / "data/orders", REPO / "data", REPO / f"data/share-reports-deploy-{PORT}", REPO / "data/alerts", ]: d.mkdir(parents=True, exist_ok=True) log = open(LOG, "w") proc = subprocess.Popen( [ str(PY), "-m", "uvicorn", "admin.app:create_app", "--factory", "--host", "127.0.0.1", "--port", str(PORT), "--log-level", "warning", ], cwd=REPO, env=env, stdout=log, stderr=subprocess.STDOUT, ) for _ in range(60): try: s, _, b = _http("GET", "/health") if s == 200 and b.strip().startswith("{"): return proc except Exception: pass time.sleep(0.5) proc.terminate() raise RuntimeError("admin failed to start") def _stop_admin(proc: subprocess.Popen) -> None: proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() def main() -> int: results: list[dict] = [] overall_ok = True def _check(name: str, ok: bool, detail: str = "") -> None: nonlocal overall_ok if not ok: overall_ok = False results.append({"check": name, "ok": ok, "detail": detail}) marker = "✅" if ok else "❌" print(f" {marker} {name:32s} {detail[:120]}") print(f"[deploy-ops] starting admin on :{PORT}") proc = _start_admin() try: # 1. /health = 200 s, _, body = _http("GET", "/health") h = json.loads(body) if s == 200 else {} _check("/health = 200", s == 200 and h.get("status") == "ok", f"status={s}") # 2. checks 三件套 c = h.get("checks", {}) _check( "checks.db_writable=True", c.get("db_writable") is True, f"db={c.get('db_writable')}", ) _check( "checks.disk_writable=True", c.get("disk_writable") is True, f"disk={c.get('disk_writable')}", ) _check( "checks.settings_valid=True", c.get("settings_valid") is True, f"settings={c.get('settings_valid')}", ) # 3. login wrong s, _, _ = _http( "POST", "/api/auth/login", body={"username": "admin", "password": "wrongpass"}, ) _check("login_wrong_password = 401", s == 401, f"status={s}") # 4. login correct s, _, body = _http( "POST", "/api/auth/login", body={"username": "admin", "password": "DeployOpsTest1!"}, ) token = json.loads(body).get("access_token", "") if s == 200 else "" _check("login_correct = 200", s == 200 and bool(token), f"status={s}") # 5. orders no token s, _, _ = _http("GET", "/api/orders") _check("orders_no_token = 401", s == 401, f"status={s}") # 6. orders bad token s, _, _ = _http("GET", "/api/orders", token="invalid.token.here") _check("orders_bad_token = 401", s in (401, 422), f"status={s}") # 7. orders valid token s, _, body = _http("GET", "/api/orders", token=token) _check( "orders_valid_token = 200", s == 200 and isinstance(json.loads(body), list), f"status={s}", ) # 8. openapi s, _, body = _http("GET", "/openapi.json") if s == 200: paths = set(json.loads(body).get("paths", {}).keys()) needed = {"/health", "/api/auth/login", "/api/orders"} _check( "openapi 含必需路径", needed.issubset(paths), f"missing={needed - paths}", ) else: _check("openapi = 200", False, f"status={s}") # 9. admin ops alerts (真实路径: /api/admin/notifications/ops-alerts) s, _, body = _http("GET", "/api/admin/notifications/ops-alerts", token=token) _check("ops_alerts = 200", s == 200, f"status={s}") # 10. admin dashboard 含 footer s, _, body = _http("GET", "/admin/dashboard", token=token) has_footer = "隐私政策" in body and "/privacy" in body _check( "admin_dashboard_footer 链接", s == 200 and has_footer, f"status={s} footer={has_footer}", ) finally: _stop_admin(proc) RESULTS.parent.mkdir(parents=True, exist_ok=True) RESULTS.write_text( json.dumps( { "date": "2026-06-20", "port": PORT, "overall_ok": overall_ok, "checks": results, }, indent=2, ensure_ascii=False, ) ) print(f"\n[deploy-ops] overall {'PASS' if overall_ok else 'FAIL'} → {RESULTS}") return 0 if overall_ok else 1 if __name__ == "__main__": sys.exit(main())