72 lines
1.8 KiB
Python
72 lines
1.8 KiB
Python
"""验证就绪探针:数据库不可用时 /health/ready 返回 503 而非 200。
|
|
|
|
运行方式(在宿主机上):
|
|
python tests/test_health_ready.py
|
|
|
|
脚本经本地 8000 端口直调 API,通过启停 postgres 容器验证:
|
|
- 健康时返回 HTTP 200
|
|
- postgres 停止后返回 HTTP 503(不再误报 200)
|
|
- postgres 恢复后回到 200
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
|
|
BASE = "http://localhost:8000"
|
|
|
|
|
|
def api_status() -> tuple[int, dict]:
|
|
try:
|
|
with urllib.request.urlopen(f"{BASE}/health/ready", timeout=5) as r:
|
|
return r.status, json.loads(r.read())
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, json.loads(e.read())
|
|
|
|
|
|
def compose(*args: str) -> None:
|
|
subprocess.run(["docker", "compose", *args], check=False, capture_output=True)
|
|
|
|
|
|
def wait_for(target: int, timeout: int = 30) -> bool:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
try:
|
|
code, _ = api_status()
|
|
if code == target:
|
|
return True
|
|
except Exception:
|
|
pass
|
|
time.sleep(1)
|
|
return False
|
|
|
|
|
|
def main() -> int:
|
|
code, _ = api_status()
|
|
if code != 200:
|
|
print(f"FAIL: 初始状态期望 200,得到 {code}"); return 1
|
|
print(f"PASS: 健康时 HTTP 200")
|
|
|
|
compose("stop", "postgres")
|
|
try:
|
|
if not wait_for(503, timeout=30):
|
|
print("FAIL: postgres 停止后未返回 503"); return 1
|
|
print("PASS: postgres 停止后 HTTP 503(就绪探针正确拒绝)")
|
|
finally:
|
|
compose("start", "postgres")
|
|
|
|
if not wait_for(200, timeout=30):
|
|
print("FAIL: postgres 恢复后未回到 200"); return 1
|
|
print("PASS: postgres 恢复后 HTTP 200")
|
|
|
|
print("ALL PASS")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|