115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
"""
|
||
进程插件（external 模式）端到端测试
|
||
|
||
前置条件:
|
||
1. 网关正常运行（uv run serve）
|
||
2. demo 进程插件（workers/demo/，load_mode = "external"）
|
||
|
||
本脚本从 demo 插件项目根（= 打包 zip 根）拉起/停止独立 worker 进程
|
||
（worker_serve --type demo），验证:
|
||
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/demo/hello?name=chongming"
|
||
POLL_TIMEOUT = 30.0
|
||
|
||
# 插件项目根（= 容器工作目录 = zip 根）：worker 在项目根下读配置/import 代码
|
||
REPO = Path(__file__).parent.parent
|
||
PROJECT = REPO / "workers" / "demo"
|
||
|
||
# 优先使用仓库 venv 的 chongming-worker 控制台脚本（避免在插件项目内新建 venv），
|
||
# 否则退回 uv run --project（SDK 入口脚本，与容器 start.sh 一致）
|
||
_WORKER_CMD = [str(Path(sys.executable).parent / "chongming-worker"), "--type", "demo"]
|
||
if not Path(_WORKER_CMD[0]).exists():
|
||
_WORKER_CMD = ["uv", "run", "--project", str(REPO), "chongming-worker", "--type", "demo"]
|
||
|
||
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 进程（CWD = 插件项目根，与容器一致）
|
||
proc = subprocess.Popen(
|
||
_WORKER_CMD,
|
||
cwd=str(PROJECT),
|
||
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,
|
||
cwd=str(PROJECT),
|
||
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(f"提示: 请先启动网关 (uv run serve)；worker 将从 {PROJECT} 拉起\n")
|
||
sys.exit(run())
|