108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
"""
|
||
混合模式（external 插件）端到端测试
|
||
|
||
前置条件:
|
||
1. 网关正常运行（uv run serve）；example 插件配置为 external：
|
||
config/workers/example/config.toml [workers] 节 load_mode = "external"
|
||
2. 本脚本自己拉起/停止独立 worker 进程（uv run worker_serve --type example）
|
||
|
||
验证:
|
||
1. worker 启动后 → HTTP 路由动态出现且跨进程 RPC 可调
|
||
2. worker 优雅退出 → 路由动态摘除（404）
|
||
3. worker 重启 → 路由恢复
|
||
"""
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
GATEWAY_URL = "http://localhost:8000"
|
||
TEST_URL = f"{GATEWAY_URL}/api/v1/calc/add?a=3&b=4"
|
||
POLL_TIMEOUT = 30.0
|
||
|
||
# 优先使用 venv 控制台脚本（uv sync 后生成），不存在时退回 uv run
|
||
_WORKER_CMD = [str(Path(sys.executable).parent / "worker_serve"), "--type", "example"]
|
||
if not Path(_WORKER_CMD[0]).exists():
|
||
_WORKER_CMD = ["uv", "run", "worker_serve", "--type", "example"]
|
||
|
||
PASS = 0
|
||
FAIL = 0
|
||
|
||
|
||
def http_get(url):
|
||
try:
|
||
with urllib.request.urlopen(url, timeout=3) as resp:
|
||
return resp.status, resp.read().decode("utf-8")
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, ""
|
||
except Exception as e:
|
||
return None, str(e)
|
||
|
||
|
||
def report(name, cond, detail=""):
|
||
global PASS, FAIL
|
||
if cond:
|
||
PASS += 1
|
||
print(f"✅ {name}" + (f" — {detail}" if detail else ""))
|
||
else:
|
||
FAIL += 1
|
||
print(f"❌ {name}" + (f" — {detail}" if detail else ""))
|
||
|
||
|
||
def wait_http(target_status=200, timeout=POLL_TIMEOUT):
|
||
deadline = time.monotonic() + timeout
|
||
last = None
|
||
while time.monotonic() < deadline:
|
||
status, body = http_get(TEST_URL)
|
||
last = (status, body)
|
||
if status == target_status:
|
||
return True, body
|
||
time.sleep(0.5)
|
||
return False, last
|
||
|
||
|
||
def run():
|
||
# 1. 拉起独立 worker 进程
|
||
proc = subprocess.Popen(
|
||
_WORKER_CMD,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
text=True,
|
||
)
|
||
try:
|
||
ok, body = wait_http(target_status=200)
|
||
report("worker 启动后路由动态注册 + 跨进程 RPC", ok, f"body={body}")
|
||
|
||
# 2. 优雅退出 → 路由摘除（404）
|
||
proc.terminate()
|
||
proc.wait(timeout=15)
|
||
ok, _ = wait_http(target_status=404)
|
||
report("worker 退出后路由动态摘除 (404)", ok)
|
||
|
||
# 3. 重启 worker → 路由恢复
|
||
proc2 = subprocess.Popen(
|
||
_WORKER_CMD,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
text=True,
|
||
)
|
||
try:
|
||
ok, body = wait_http(target_status=200)
|
||
report("worker 重启后路由恢复", ok, f"body={body}")
|
||
finally:
|
||
proc2.terminate()
|
||
proc2.wait(timeout=15)
|
||
finally:
|
||
if proc.poll() is None:
|
||
proc.terminate()
|
||
|
||
print("\n" + "=" * 50)
|
||
print(f"测试结果: ✅ {PASS} 通过 ❌ {FAIL} 失败")
|
||
return 0 if FAIL == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("提示: 请先以 load_mode=process 启动网关 (uv run serve)\n")
|
||
sys.exit(run())
|