重构：模块化 RRPP Lite 并更新入口
This commit is contained in:
RoyirinRosmontis
@@ -0,0 +1,10 @@
|
||||
class Qt: pass
|
||||
class QTimer: pass
|
||||
|
||||
class _Signal:
|
||||
def connect(self, *a, **k): pass
|
||||
def emit(self, *a, **k): pass
|
||||
|
||||
def pyqtSignal(*a, **k):
|
||||
return _Signal()
|
||||
class QRectF: pass
|
||||
@@ -0,0 +1,6 @@
|
||||
class QFont: pass
|
||||
class QIcon: pass
|
||||
class QPixmap: pass
|
||||
class QImage: pass
|
||||
class QColor: pass
|
||||
class QPainter: pass
|
||||
@@ -0,0 +1,28 @@
|
||||
class QApplication: pass
|
||||
class QWidget: pass
|
||||
class QLabel:
|
||||
def __init__(self, text=""):
|
||||
self.text = text
|
||||
def setStyleSheet(self, *a): pass
|
||||
def setToolTip(self, *a): pass
|
||||
def setFixedSize(self, *a): pass
|
||||
def setFixedHeight(self, *a): pass
|
||||
def setAlignment(self, *a): pass
|
||||
def setPixmap(self, *a): pass
|
||||
def setText(self, *a): pass
|
||||
def setCursor(self, *a): pass
|
||||
def unsetCursor(self): pass
|
||||
def hide(self): pass
|
||||
def show(self): pass
|
||||
class QPushButton: pass
|
||||
class QVBoxLayout: pass
|
||||
class QHBoxLayout: pass
|
||||
class QFrame: pass
|
||||
class QMessageBox: pass
|
||||
class QTextEdit: pass
|
||||
class QCheckBox: pass
|
||||
class QMenu: pass
|
||||
class QDialog: Accepted = 1
|
||||
class QLineEdit: Password = 2
|
||||
class QListWidget: pass
|
||||
class QListWidgetItem: pass
|
||||
@@ -0,0 +1,2 @@
|
||||
# PyQt5 桩模块：仅供 WSL 下 import main 做路由逻辑测试，
|
||||
# 决策函数（_decide_direct 等）不依赖任何 Qt 对象。
|
||||
@@ -16,15 +16,19 @@ sys.path = [str(LITE)] + [p for p in sys.path if Path(p).resolve() != Path(__fil
|
||||
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
|
||||
import proxy_lite as pp
|
||||
import main as pp
|
||||
from PyQt5.QtWidgets import QApplication
|
||||
|
||||
# 构造窗口时不写用户数据目录（避免污染），构造后再指向真实 nodes.json
|
||||
# 构造窗口时不触发用户数据初始化或远程同步；本测试只验证静态菜单和本地节点状态。
|
||||
pp._user_init_local = lambda: None
|
||||
pp._user_init_remote = lambda *args, **kwargs: None
|
||||
pp._nodes_sync_remote = lambda *args, **kwargs: pp._NODES_SYNC_DONE.set()
|
||||
pp._NODES_SYNC_DONE.set()
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
w = pp.MainWindow()
|
||||
w.timer.stop() # 停掉状态轮询，纯静态检查
|
||||
failures = []
|
||||
|
||||
pp._USER_DIR = LITE / "test" # 真实 nodes.json 所在目录
|
||||
nodes = pp._reload_nodes()
|
||||
@@ -44,6 +48,8 @@ for act in menu.actions():
|
||||
if not act.isEnabled():
|
||||
act.trigger()
|
||||
print(f"\n点击禁用项后节点下标: {before} -> {pp._selected_index()} (应不变)")
|
||||
if pp._selected_index() != before:
|
||||
failures.append("点击不可用节点后发生了切换")
|
||||
|
||||
# ── 上次选中的节点已不可用：启动时自动回退到第一个可达节点 ──────────
|
||||
print("\n启动校验（_verify_selected_node）:")
|
||||
@@ -54,19 +60,28 @@ else:
|
||||
import json
|
||||
_bak = (pp._USER_DIR / "settings.json").read_bytes() if (pp._USER_DIR / "settings.json").is_file() else None
|
||||
bad_idx = bad[0]
|
||||
pp._SELECTED_NODE = bad_idx # 模拟上次选了不可用节点
|
||||
# 模拟上次持久化选中了不可用节点：内存下标与 current_node 必须一致，
|
||||
# 否则 _verify_selected_node 会先按真实保存的域名重新匹配，测试并非回退分支。
|
||||
pp._settings_save({**pp._settings_load(), "current_node": nodes[bad_idx].get("domain") or ""})
|
||||
pp._SELECTED_NODE = bad_idx
|
||||
switched = pp._verify_selected_node()
|
||||
expect = pp._first_reachable_index(nodes)
|
||||
ok = switched and pp._selected_index() == expect
|
||||
print(f" 上次选中 {nodes[bad_idx].get('domain')}(下标 {bad_idx}) 不可用 -> "
|
||||
f"回退到下标 {pp._selected_index()} ({nodes[pp._selected_index()].get('domain')}) "
|
||||
f"{'PASS' if ok else 'FAIL'}")
|
||||
if not ok:
|
||||
failures.append("不可用节点未回退到首个可达节点")
|
||||
saved = json.loads((pp._USER_DIR / "settings.json").read_text("utf-8")).get("current_node") if (pp._USER_DIR / "settings.json").is_file() else ""
|
||||
ok2 = saved == str(nodes[expect].get("domain") or "")
|
||||
print(f" 回退结果已写入设置 current_node={saved!r} {'PASS' if ok2 else 'FAIL'}")
|
||||
if not ok2:
|
||||
failures.append("节点回退未写回 current_node")
|
||||
# 再校验一次：已经是可达节点，不应再切换
|
||||
ok3 = pp._verify_selected_node() is False
|
||||
print(f" 再次校验不重复切换 {'PASS' if ok3 else 'FAIL'}")
|
||||
if not ok3:
|
||||
failures.append("已回退节点被重复切换")
|
||||
if _bak is None:
|
||||
(pp._USER_DIR / "settings.json").unlink(missing_ok=True)
|
||||
else:
|
||||
@@ -87,13 +102,19 @@ else:
|
||||
ok = switched and ordered[pp._selected_index()].get("domain") == dom_old
|
||||
print(f" current_node={dom_old} 调序后跟随域名 -> 下标 {pp._selected_index()} "
|
||||
f"{'PASS' if ok else 'FAIL'}")
|
||||
if not ok:
|
||||
failures.append("节点调序后没有按域名重新匹配")
|
||||
# 已是正确域名时再跑一次：不应重复切换
|
||||
ok2 = pp._rematch_saved_domain() is False
|
||||
print(f" 再次重匹配不重复切换 {'PASS' if ok2 else 'FAIL'}")
|
||||
if not ok2:
|
||||
failures.append("节点调序重匹配重复切换")
|
||||
# 域名已不在列表（节点被移除）：不切换，保持现状
|
||||
pp._settings_save({**pp._settings_load(), "current_node": "gone.example.com"})
|
||||
ok3 = pp._rematch_saved_domain() is False
|
||||
print(f" 节点被移除时不动 {'PASS' if ok3 else 'FAIL'}")
|
||||
if not ok3:
|
||||
failures.append("已移除节点仍发生切换")
|
||||
pp._NODES = None # 还原缓存，后续用例重读文件
|
||||
# 还原 settings.json，避免测试污染
|
||||
if _bak is None:
|
||||
@@ -101,4 +122,6 @@ else:
|
||||
else:
|
||||
(pp._USER_DIR / "settings.json").write_bytes(_bak)
|
||||
|
||||
if failures:
|
||||
raise SystemExit("FAIL: " + "; ".join(failures))
|
||||
print("OK")
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
[
|
||||
{
|
||||
"domain": "example.com",
|
||||
"port": 443,
|
||||
"country": "TEST",
|
||||
"remark": "普通测试节点"
|
||||
},
|
||||
{
|
||||
"domain": "example.org",
|
||||
"port": 443,
|
||||
"country": "TEST",
|
||||
"remark": "备用测试节点"
|
||||
},
|
||||
{
|
||||
"domain": "ipv6.test.invalid",
|
||||
"port": 443,
|
||||
"country": "TEST",
|
||||
"ipv6_only": true,
|
||||
"remark": "IPv6 测试节点"
|
||||
}
|
||||
]
|
||||
@@ -0,0 +1,151 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""连通性探测：用 test/nodes.json 的真实节点，走 _tls_start 转发逻辑实测。
|
||||
|
||||
仅需标准库；不启动 GUI（main 的 GUI 在 __main__ 保护下）。
|
||||
用法: python3 test/probe_nodes.py [JP US ...] 不带参数测全部
|
||||
"""
|
||||
import os, sys, time, json, ssl, socket, urllib.request, urllib.error
|
||||
from pathlib import Path
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
LITE = HERE.parent
|
||||
sys.path.insert(0, str(HERE)) # PyQt5 桩
|
||||
sys.path.insert(0, str(LITE))
|
||||
|
||||
for k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"):
|
||||
os.environ.pop(k, None)
|
||||
# 不能设 NO_PROXY=*：urllib 会把 127.0.0.1 也当直连，显式 ProxyHandler 被绕过
|
||||
os.environ["NO_PROXY"] = ""
|
||||
|
||||
import main as pp
|
||||
|
||||
NODES = json.loads((HERE / "nodes.json").read_text("utf-8"))
|
||||
want = [a.upper() for a in sys.argv[1:]]
|
||||
if want:
|
||||
NODES = [n for n in NODES if str(n.get("country", "")).upper() in want]
|
||||
|
||||
TIMEOUT = 12
|
||||
|
||||
|
||||
def free_port():
|
||||
s = socket.socket(); s.bind(("127.0.0.1", 0))
|
||||
p = s.getsockname()[1]; s.close(); return p
|
||||
|
||||
|
||||
def http_method(node):
|
||||
return (node.get("methods") or {}).get("http") or {}
|
||||
|
||||
|
||||
def as_flat_node(node):
|
||||
"""把 nodes.json 的 methods.http 结构摊平成代码 _node_val() 期望的形状。
|
||||
仅用于探测：暴露「代码读不到真实节点字段」这一问题。"""
|
||||
m = http_method(node)
|
||||
return dict(node, port=int(m.get("port")), username=m.get("username", ""),
|
||||
password=m.get("password", ""))
|
||||
|
||||
|
||||
def probe_connect(host, port, sni=None, connect_timeout=8):
|
||||
"""直连探测：DNS -> TCP -> TLS -> 发 CONNECT 127.0.0.1:9 看上游代理响应。"""
|
||||
res = {"dns": None, "tcp": None, "tls": None, "proxy": None, "err": None}
|
||||
t0 = time.time()
|
||||
try:
|
||||
infos = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM)
|
||||
res["dns"] = sorted({i[4][0] for i in infos})
|
||||
except Exception as e:
|
||||
res["err"] = f"DNS: {e}"
|
||||
return res
|
||||
addr = res["dns"][0]
|
||||
af = socket.AF_INET6 if ":" in addr else socket.AF_INET
|
||||
try:
|
||||
s = socket.socket(af, socket.SOCK_STREAM)
|
||||
s.settimeout(connect_timeout)
|
||||
s.connect((addr, port))
|
||||
res["tcp"] = round(time.time() - t0, 2)
|
||||
except Exception as e:
|
||||
res["err"] = f"TCP: {e}"
|
||||
return res
|
||||
try:
|
||||
ctx = ssl.create_default_context()
|
||||
s = ctx.wrap_socket(s, server_hostname=sni or host)
|
||||
res["tls"] = s.version()
|
||||
except Exception as e:
|
||||
res["err"] = f"TLS: {e}"
|
||||
try: s.close()
|
||||
except Exception: pass
|
||||
return res
|
||||
try:
|
||||
s.sendall(b"CONNECT 127.0.0.1:9 HTTP/1.1\r\nHost: 127.0.0.1:9\r\n\r\n")
|
||||
res["proxy"] = s.recv(200).split(b"\r\n", 1)[0].decode("latin-1", "replace")
|
||||
except Exception as e:
|
||||
res["err"] = f"CONNECT: {e}"
|
||||
try: s.close()
|
||||
except Exception: pass
|
||||
return res
|
||||
|
||||
|
||||
def probe_via_forwarder(node):
|
||||
"""真实转发器链路：_tls_start -> 经 127.0.0.1 代理请求，取出口 IP 与归属地。"""
|
||||
m = http_method(node)
|
||||
if not m:
|
||||
return "（该节点无 http 方法，跳过）"
|
||||
old_node, old_port = pp._current_node, pp._TLS_LOCAL_PORT
|
||||
port = free_port()
|
||||
try:
|
||||
# _proxy_auth_basic() 内部走 _current_node()，固定成本次待测节点
|
||||
pp._current_node = lambda: node
|
||||
pp._TLS_LOCAL_PORT = port
|
||||
pp._tls_stop()
|
||||
pp._tls_start(node["domain"], int(m["port"]), use_tls=True,
|
||||
bypass_lan=False, smart=False, quiet=True)
|
||||
ph = urllib.request.ProxyHandler({"http": f"127.0.0.1:{port}",
|
||||
"https": f"127.0.0.1:{port}"})
|
||||
op = urllib.request.build_opener(ph)
|
||||
t0 = time.time()
|
||||
# 明文 HTTP 代理（走 GET 绝对 URI）与 CONNECT 隧道各测一条
|
||||
with op.open("http://ip-api.com/json/?fields=query,country,regionName,isp",
|
||||
timeout=TIMEOUT) as r:
|
||||
info = json.loads(r.read(512).decode("utf-8", "replace"))
|
||||
with op.open("https://ifconfig.me/ip", timeout=TIMEOUT) as r:
|
||||
https_ip = r.read(64).decode().strip()
|
||||
dt = round(time.time() - t0, 2)
|
||||
return (f"OK {dt}s 出口IP={info.get('query')} "
|
||||
f"[{info.get('country')} {info.get('regionName')} / {info.get('isp')}] "
|
||||
f"HTTPS隧道出口={https_ip}")
|
||||
except urllib.error.HTTPError as e:
|
||||
return f"HTTP {e.code}: {e.read(120).decode('utf-8', 'replace').strip()[:100]}"
|
||||
except Exception as e:
|
||||
return f"失败: {type(e).__name__}: {e}"
|
||||
finally:
|
||||
pp._tls_stop()
|
||||
pp._current_node, pp._TLS_LOCAL_PORT = old_node, old_port
|
||||
|
||||
|
||||
print(f"节点数: {len(NODES)}")
|
||||
n0 = NODES[0]
|
||||
print(f"代码 _endpoint_domain()={pp._endpoint_domain()!r} "
|
||||
f"_endpoint_port()={pp._endpoint_port()!r} _endpoint_country()={pp._endpoint_country()!r}")
|
||||
print(f"代码 _proxy_auth_basic()={pp._proxy_auth_basic()!r}")
|
||||
print(f"nodes.json[0] 顶层键 = {list(n0.keys())}")
|
||||
print(f"nodes.json[0] domain={n0.get('domain')!r} methods.http.port="
|
||||
f"{http_method(n0).get('port')!r}")
|
||||
print()
|
||||
|
||||
for n in NODES:
|
||||
c = n.get("country", "?")
|
||||
m = http_method(n)
|
||||
print("=" * 78)
|
||||
print(f"[{c}] {n.get('domain')}:{m.get('port')} provider={n.get('provider')} "
|
||||
f"traffic={n.get('traffic')} level={n.get('available_level')}")
|
||||
if not m:
|
||||
print(" 该节点无 http 方法，跳过")
|
||||
continue
|
||||
r = probe_connect(n["domain"], int(m["port"]))
|
||||
print(f" DNS : {r['dns']}")
|
||||
print(f" TCP : {r['tcp']}s" if r["tcp"] is not None else " TCP : -")
|
||||
print(f" TLS : {r['tls']}" if r.get("tls") else " TLS : -")
|
||||
print(f" CONNECT: {r['proxy']}" if r.get("proxy") else " CONNECT: -")
|
||||
if r.get("err"):
|
||||
print(f" 错误 : {r['err'].splitlines()[0][:200]}")
|
||||
print(f" 经转发器: {probe_via_forwarder(n)}")
|
||||
print(f" 经转发器(摊平节点): {probe_via_forwarder(as_flat_node(n))}")
|
||||
print()
|
||||
@@ -0,0 +1,114 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""拆分后 rrpp_core 的独立回归测试（不依赖 Qt 或真实代理连接）。"""
|
||||
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
LITE_DIR = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(LITE_DIR))
|
||||
|
||||
from rrpp_core.forwarder_primitives import (
|
||||
force_http_close,
|
||||
inject_proxy_auth,
|
||||
parse_connect_authority,
|
||||
parse_http_absolute,
|
||||
)
|
||||
from rrpp_core.node_api import NodeAuthError, b64url
|
||||
from rrpp_core.node_secrets import NodeSecretsStore
|
||||
from rrpp_core.network_primitives import is_private_host, is_private_ipv4_int, ttl_cache
|
||||
from rrpp_core.pac_data import atomic_write, read_version, validate_document
|
||||
from rrpp_core.routing_primitives import (
|
||||
build_suffix_trie,
|
||||
merge_ip_ranges,
|
||||
parse_domain_rules,
|
||||
suffix_trie_matches,
|
||||
)
|
||||
from rrpp_core.settings_store import JsonSettingsStore
|
||||
from rrpp_core.traffic import TrafficStats, format_bytes
|
||||
|
||||
|
||||
class CoreModuleTests(unittest.TestCase):
|
||||
def test_domain_rules_and_trie(self):
|
||||
suffixes, exact, regexes, invalid = parse_domain_rules(
|
||||
["example.com", "full:only.example", "regexp:^edge-[0-9]+\\.example$", "regexp:["]
|
||||
)
|
||||
self.assertEqual(invalid, 1)
|
||||
self.assertIn("example.com", suffixes)
|
||||
self.assertIn("only.example", exact)
|
||||
self.assertTrue(regexes[0].search("edge-42.example"))
|
||||
trie = build_suffix_trie(suffixes)
|
||||
self.assertTrue(suffix_trie_matches("cdn.example.com", trie))
|
||||
self.assertFalse(suffix_trie_matches("example.net", trie))
|
||||
self.assertEqual(
|
||||
merge_ip_ranges(["10.0.0.0/25", "10.0.0.128/25"], 4),
|
||||
[(167772160, 167772415)],
|
||||
)
|
||||
|
||||
def test_forwarder_primitives(self):
|
||||
self.assertEqual(
|
||||
parse_connect_authority("CONNECT [::1]:443 HTTP/1.1"), ("::1", 443)
|
||||
)
|
||||
self.assertEqual(
|
||||
parse_http_absolute("GET https://example.com/path HTTP/1.1"),
|
||||
("example.com", 443),
|
||||
)
|
||||
request = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
|
||||
injected = inject_proxy_auth(request, "token")
|
||||
self.assertIn(b"Authorization: Basic token\r\n", injected)
|
||||
self.assertEqual(inject_proxy_auth(injected, "different"), injected)
|
||||
self.assertIn(b"Connection: close\r\n", force_http_close(request))
|
||||
|
||||
def test_settings_pac_and_node_secrets_storage(self):
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
directory = Path(temporary)
|
||||
settings = JsonSettingsStore(lambda: directory / "settings.json")
|
||||
self.assertEqual(settings.load(), {})
|
||||
settings.save({"local_port": 1080})
|
||||
self.assertEqual(settings.load(), {"local_port": 1080})
|
||||
|
||||
payload = {"version": "1", "domains": ["example.com"]}
|
||||
atomic_write(directory, "pac_domains", payload)
|
||||
self.assertEqual(read_version(directory / "pac_domains.json"), ("1", ""))
|
||||
self.assertTrue(validate_document("pac_domains", payload))
|
||||
|
||||
secrets = NodeSecretsStore(lambda: directory)
|
||||
self.assertEqual(secrets.decrypt_bytes(secrets.encrypt_text("round-trip")), "round-trip")
|
||||
secrets.save_auth("test-user", "test-password")
|
||||
self.assertEqual(
|
||||
secrets.load_auth(), {"username": "test-user", "password": "test-password"}
|
||||
)
|
||||
|
||||
def test_traffic_and_node_protocol_helpers(self):
|
||||
times = iter((10.0, 12.0))
|
||||
stats = TrafficStats(clock=lambda: next(times))
|
||||
stats.reset()
|
||||
stats.add(up=20, down=10)
|
||||
self.assertEqual(stats.rate(), (10, 5, 20, 10))
|
||||
self.assertEqual(format_bytes(1536), "1.5 KB")
|
||||
self.assertEqual(b64url("a/b="), "YS9iPQ")
|
||||
error = NodeAuthError("denied", 403)
|
||||
self.assertEqual((str(error), error.code), ("denied", 403))
|
||||
|
||||
def test_network_primitives_and_ttl_cache(self):
|
||||
self.assertTrue(is_private_ipv4_int(0x0A010203))
|
||||
self.assertTrue(is_private_host("::1"))
|
||||
self.assertFalse(is_private_host("8.8.8.8"))
|
||||
calls = []
|
||||
|
||||
@ttl_cache(300)
|
||||
def probe(value):
|
||||
calls.append(value)
|
||||
return value
|
||||
|
||||
self.assertEqual((probe("x"), probe("x")), ("x", "x"))
|
||||
self.assertEqual(calls, ["x"])
|
||||
probe.cache_clear()
|
||||
probe("x")
|
||||
self.assertEqual(calls, ["x", "x"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
@@ -3,13 +3,13 @@
|
||||
|
||||
核心问题：PAC 模式下，目标域名解析为国外 IP 时，是否强制走隧道代理？
|
||||
|
||||
判定链（proxy_lite._decide_direct）：
|
||||
判定链（main._decide_direct）：
|
||||
0. 代理白名单命中 -> 强制直连（最优先，PAC/固定模式均生效）
|
||||
1. pac_domains.json 域名表命中 -> 强制隧道（优先于 IP 判定）
|
||||
2. 内网 IP 且勾选 bypass_lan -> 直连
|
||||
3. 智能分流：中国 IP -> 直连；其余（国外 IP）-> 走隧道
|
||||
|
||||
直接 import proxy_lite 调用真实决策函数，数据用真实的 pac_ip_cn.json /
|
||||
直接 import main 调用真实决策函数，数据用真实的 pac_ip_cn.json /
|
||||
pac_domains.json；PyQt5 用桩替代（决策逻辑不依赖 Qt）。
|
||||
"""
|
||||
|
||||
@@ -25,8 +25,8 @@ sys.path.insert(0, str(LITE_DIR))
|
||||
|
||||
import json
|
||||
import socket
|
||||
import proxy_lite as pp
|
||||
from proxy_lite import _decide_direct, _is_china_ip, _is_force_proxy, _is_private_ip, _is_whitelisted
|
||||
import main as pp
|
||||
from main import _decide_direct, _is_china_ip, _is_force_proxy, _is_private_ip, _is_whitelisted
|
||||
|
||||
# PAC 数据已不再随项目保留：从 CDN 拉取到临时用户目录（真实数据端到端）
|
||||
import tempfile
|
||||
@@ -241,10 +241,11 @@ if real_nodes:
|
||||
print(f" [{c}] {n.get('domain'):<24} ipv6_only={str(io):<6} "
|
||||
f"可达={str(reach):<6} 菜单文案: {n.get('domain')} ({c})"
|
||||
+ (f" {mark}" if io else ""))
|
||||
# ipv6_only 节点在本机无 IPv6 时必须判定为不可达，否则必须可达
|
||||
# ipv6_only 节点在本机无 IPv6 时必须判定为不可达；有 IPv6 时仍取
|
||||
# 实际 TCP 探测结果（本机有 IPv6 不代表每个远端节点都必然在线）。
|
||||
if io:
|
||||
expect = has_v6 or reach # 本机有 IPv6 -> 应可达；无 IPv6 -> 不应可达
|
||||
ok = reach == expect
|
||||
expect = "实际探测" if has_v6 else False
|
||||
ok = isinstance(reach, bool) if has_v6 else reach is False
|
||||
ipv6_failed += 0 if ok else 1
|
||||
print(f" IPv6 专用节点可达性: {reach} (期望 {expect}) "
|
||||
f"{'PASS' if ok else 'FAIL'}")
|
||||
|
||||
Reference in New Issue
Block a user