117 lines
5.3 KiB
Python
117 lines
5.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""mcpctl 热重载前置探活单测（2026-09-25 用户反馈）。
|
||
|
||
用户场景：mcp↔wsl 开关对调后，WSL/OC 没在跑 → 报「热重载失败：… WinError 10061
|
||
连接拒绝」。期望：没有可重载的 OC 就跳过重载、按成功返回（下次冷启动自动加载），
|
||
只有「OC 在跑但重载出错」才算失败。
|
||
"""
|
||
import os
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import mcpctl # noqa: E402
|
||
|
||
FAIL: list[str] = []
|
||
|
||
|
||
def check(name: str, cond: bool, extra: str = "") -> None:
|
||
print(("PASS " if cond else "FAIL ") + name + (" | " + extra if extra else ""))
|
||
if not cond:
|
||
FAIL.append(name)
|
||
|
||
|
||
class _A: # 最小 args 替身（只用到 _live_oc/do_reload 需要的字段）
|
||
distro = "Ubuntu-24.04"
|
||
password = ""
|
||
port = 4096
|
||
timeout = 2.0
|
||
|
||
|
||
args = _A()
|
||
_orig = (mcpctl._wsl_running, mcpctl._find_live_host, mcpctl._post)
|
||
_orig_cand = mcpctl._candidate_hosts
|
||
|
||
|
||
def _no_net(*_a, **_k):
|
||
raise AssertionError("这一步不该做网络探测")
|
||
|
||
|
||
try:
|
||
# ---------- 1. WSL 目标 + VM 未运行 -> 直接 skip，绝不探测 ----------
|
||
mcpctl._wsl_running = lambda d: False
|
||
mcpctl._find_live_host = _no_net
|
||
h, pwd, skip = mcpctl._live_oc("wsl", args)
|
||
check("WSL 未运行 -> skip 且不做探测",
|
||
h == "" and bool(skip) and "WSL 未启动" in skip, str(skip))
|
||
|
||
# ---------- 2. WSL 在跑但 OC 没在跑 -> skip 带探测原因（不再是「失败」） ----------
|
||
mcpctl._wsl_running = lambda d: True
|
||
mcpctl._find_live_host = (
|
||
lambda t, a, pw: ("", "/root/.config/opencode", "http://172.29.95.185:4096 连不上")
|
||
)
|
||
h, pwd, skip = mcpctl._live_oc("wsl", args)
|
||
check("OC 未运行 -> skip 带原因", h == "" and bool(skip) and "连不上" in skip, str(skip))
|
||
|
||
# ---------- 3. OC 在跑 -> 不 skip，host 直接复用 ----------
|
||
mcpctl._find_live_host = (
|
||
lambda t, a, pw: ("http://127.0.0.1:4096", "/root/.config/opencode", "")
|
||
)
|
||
h, pwd, skip = mcpctl._live_oc("wsl", args)
|
||
check("OC 运行中 -> 不 skip 且拿到 host",
|
||
skip is None and h.startswith("http"), str((h, skip)))
|
||
|
||
# ---------- 4. do_reload 复用 host -> 不再探测，直接 reload 成功 ----------
|
||
mcpctl._find_live_host = _no_net
|
||
mcpctl._post = lambda url, pw, timeout, data=b"{}", method="POST": (200, "{}")
|
||
ok, msg = mcpctl.do_reload("wsl", args, host="http://127.0.0.1:4096", password="x")
|
||
check("do_reload 复用 host -> 热重载成功", ok and "热重载成功" in msg, msg)
|
||
|
||
# ---------- 5. OC 在跑但 reload 返回错误 -> 仍算真实失败（不能吞） ----------
|
||
mcpctl._post = lambda url, pw, timeout, data=b"{}", method="POST": (0, "boom")
|
||
ok, msg = mcpctl.do_reload("wsl", args, host="http://127.0.0.1:4096", password="x")
|
||
check("OC 在跑但重载连不上 -> 仍报失败", (not ok) and "连不上" in msg, msg)
|
||
|
||
# ---------- 6. 本机目标 OC 没在跑 -> skip（本机目标不看 WSL 状态） ----------
|
||
mcpctl._find_live_host = (
|
||
lambda t, a, pw: ("", "C:/cfg", "没有找到运行中的 OC（4096 未监听，端口发现也未命中）")
|
||
)
|
||
h, pwd, skip = mcpctl._live_oc("local", args)
|
||
check("local OC 未运行 -> skip", h == "" and bool(skip) and "没有找到" in skip, str(skip))
|
||
|
||
# ---------- 7. 探活措辞（2026-09-25 Round 9）：OC 没启动要说「未启动」 ----------
|
||
# 真实 _find_live_host + 假 _post：10061 拒绝不许再写成吓人的「连不上」
|
||
mcpctl._find_live_host = _orig[1]
|
||
mcpctl._candidate_hosts = lambda t, a: ["http://172.29.95.185:4096"]
|
||
mcpctl._post = lambda *ar, **kw: (
|
||
0, "<urlopen error [WinError 10061] 由于目标计算机积极拒绝，无法连接。>")
|
||
h, _d, why = mcpctl._find_live_host("wsl", args, "pw")
|
||
check("10061 拒绝 -> 措辞「OC 未启动」且不含「连不上」",
|
||
h == "" and "OC 未启动" in why and "连不上" not in why, why)
|
||
|
||
# ---------- 8. 超时 -> 「连接超时（OC 未启动或不可达）」 ----------
|
||
mcpctl._post = lambda *ar, **kw: (
|
||
0, "<urlopen error [WinError 10060] 连接方在一段时间后没有正确答复>")
|
||
h, _d, why = mcpctl._find_live_host("wsl", args, "pw")
|
||
check("10060 超时 -> 措辞「连接超时」", h == "" and "超时" in why, why)
|
||
|
||
# ---------- 9. 未知网络错误 -> 保留原文（真实错误不能吞） ----------
|
||
mcpctl._post = lambda *ar, **kw: (0, "weird backend exploded")
|
||
h, _d, why = mcpctl._find_live_host("wsl", args, "pw")
|
||
check("未知错误 -> 保留「连不上」原文", h == "" and "连不上" in why, why)
|
||
|
||
# ---------- 10. skip 端到端：_live_oc 拼出的整句都不该像报错 ----------
|
||
mcpctl._wsl_running = lambda d: True
|
||
mcpctl._post = lambda *ar, **kw: (
|
||
0, "<urlopen error [WinError 10061] 由于目标计算机积极拒绝，无法连接。>")
|
||
h, pwd, skip = mcpctl._live_oc("wsl", args)
|
||
check("skip 整句 = 状态描述（含「OC 未启动」，不含「连不上」）",
|
||
h == "" and bool(skip) and "OC 未启动" in skip and "连不上" not in skip,
|
||
str(skip))
|
||
finally:
|
||
mcpctl._wsl_running, mcpctl._find_live_host, mcpctl._post = _orig
|
||
mcpctl._candidate_hosts = _orig_cand
|
||
|
||
print()
|
||
print(("ALL PASS" if not FAIL else "FAILURES: " + ", ".join(FAIL)))
|
||
sys.exit(1 if FAIL else 0)
|