# -*- coding: utf-8 -*-
"""PAC（智能分流）模式路由决策测试。

核心问题：PAC 模式下，目标域名解析为国外 IP 时，是否强制走隧道代理？

判定链（proxy_lite._decide_direct）：
  0. 代理白名单命中 -> 强制直连（最优先，PAC/固定模式均生效）
  1. pac_domains.json 域名表命中 -> 强制隧道（优先于 IP 判定）
  2. 内网 IP 且勾选 bypass_lan -> 直连
  3. 智能分流：中国 IP -> 直连；其余（国外 IP）-> 走隧道

直接 import proxy_lite 调用真实决策函数，数据用真实的 pac_ip_cn.json /
pac_domains.json；PyQt5 用桩替代（决策逻辑不依赖 Qt）。
"""

import sys
import threading
from pathlib import Path

TEST_DIR = Path(__file__).resolve().parent
LITE_DIR = TEST_DIR.parent

sys.path.insert(0, str(TEST_DIR))  # 假 PyQt5 桩
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

# PAC 数据已不再随项目保留：从 CDN 拉取到临时用户目录（真实数据端到端）
import tempfile
pp._USER_DIR = Path(tempfile.mkdtemp()) / "RRPP"
pp._user_init_remote(base_url="https://cdn.rrrosmontis.icu/RRPP/PAC/")

pp._load_cn_ranges()
pp._load_force_proxy_suffixes()
if not pp._CN_RANGES:
    print("警告: 未能获取 pac_ip_cn.json（CDN 不可达？），中国 IP 判定将失效")
print(f"中国IP段: {len(pp._CN_RANGES or [])} 段, 强制域名表: {len(pp._FORCE_PROXY_SUFFIXES or [])} 条\n")


def resolved_ips(host):
    try:
        return sorted({a[4][0] for a in socket.getaddrinfo(host, 443, socket.AF_INET)})
    except Exception:
        return []


def mark(ip):
    if _is_private_ip(ip):
        return "内网"
    if _is_china_ip(ip):
        return "中国IP"
    return "国外IP"


PATH = {True: "直连", False: "隧道"}
MODES = [
    ("PAC+bypass_lan", dict(bypass_lan=True, smart=True)),   # 默认勾选状态
    ("PAC无bypass_lan", dict(bypass_lan=False, smart=True)),
    ("固定代理模式",    dict(bypass_lan=False, smart=False)),  # 全部走隧道
]

# (名称, host, 期望 PAC+bypass_lan 决策: True=直连 / False=隧道)
cases = [
    ("境外域名 google.com",       "google.com",          False),
    ("境外域名 github.com",       "github.com",          False),
    ("境外域名 youtube.com",      "youtube.com",         False),
    ("域名表强制 claude.ai",      "claude.ai",           False),
    ("域名表强制 *.google 后缀",  "antigravity.google",  False),
    ("国内域名 www.baidu.com",    "www.baidu.com",       True),
    ("国内域名 www.qq.com",       "www.qq.com",          True),
    ("内网 localhost",            "localhost",           True),
    ("内网 IP 192.168.1.1",       "192.168.1.1",         True),
    ("境外 IP 8.8.8.8",           "8.8.8.8",             False),
    ("国内 IP 114.114.114.114",   "114.114.114.114",     True),
]

failed = 0
for name, host, expect in cases:
    ips = resolved_ips(host)
    force = _is_force_proxy(host)
    marks = f"[{','.join(mark(ip) for ip in ips)}]" if ips else "[解析失败]"
    row = f"{name:<22} {host:<18} {str(ips) if ips else '无IP':<45} {marks}"
    results = []
    for mode_name, kw in MODES:
        got = _decide_direct(host, **kw)
        results.append(f"{mode_name}={PATH[got]}")
    pac = _decide_direct(host, **MODES[0][1])
    ok = pac == expect
    failed += 0 if ok else 1
    print(f"{row}\n    {'域名表命中' if force else '域名表未命中':<10} | {'  '.join(results)} | 期望 {PATH[expect]} -> {'PASS' if ok else 'FAIL'}")

print(f"\n{'='*70}")
print(f"结论: {'全部 PASS' if not failed else f'{failed} 个 FAIL'}")
print("=" * 70)
print("PAC 模式(智能分流)下: 域名解析结果全部为国外 IP -> 走隧道代理 ✓"
      if not failed else "存在 FAIL，请检查上方明细")

# ── 附加测试：不在域名表的被墙境外站 ─────────────────────
# 这类站点不能靠域名表兜底，必须仅凭「解析结果为国外 IP」强制走隧道，
# 否则会尝试直连 -> 被墙。
print("\n" + "=" * 70)
print("附加: 不在 pac_domains.json 的被墙境外站（只能靠 IP 判定走隧道）")
print("=" * 70)

extra_candidates = [
    "dailymotion.com", "vimeo.com", "notion.so", "linear.app",
    "quora.com", "feedly.com", "dropbox.com", "tumblr.com",
    "flickr.com", "pastebin.com", "fandom.com", "archive.org",
    "godaddy.com", "imgur.com", "kickstarter.com", "shopify.com",
]

extra_failed = 0
skipped = 0
for host in extra_candidates:
    if _is_force_proxy(host):
        print(f"{host:<20} 已在域名表中，跳过")
        skipped += 1
        continue
    ips = resolved_ips(host)
    if not ips:
        print(f"{host:<20} 解析失败，跳过")
        skipped += 1
        continue
    foreign = all(not _is_private_ip(ip) and not _is_china_ip(ip) for ip in ips)
    if not foreign:
        print(f"{host:<20} 解析到非国外 IP {ips}，跳过（不是有效的境外用例）")
        skipped += 1
        continue
    got = _decide_direct(host, bypass_lan=True, smart=True)  # PAC 模式
    if got is not False:
        # 决策函数内部会重新解析域名，CDN 轮询 DNS 可能与其上方展示的解析结果
        # 不一致：清缓存重试一次，仍不一致则按 DNS 不稳定跳过（非代码问题）
        _is_china_ip.cache_clear()
        _is_private_ip.cache_clear()
        ips = resolved_ips(host)
        got = _decide_direct(host, bypass_lan=True, smart=True)
        if got is not False:
            print(f"{host:<20} 解析/判定随 DNS 轮询不一致 {ips}，跳过（非代码问题）")
            skipped += 1
            continue
        print(f"{host:<20} 重试后走隧道（DNS 瞬时差异） PASS")
        continue
    extra_failed += 0
    print(f"{host:<20} {str(ips):<50} [{'，'.join(mark(ip) for ip in ips)}] -> "
          f"{PATH[got]} (期望隧道) PASS")

print(f"\n附加测试: {len(extra_candidates)} 个候选, 有效用例 {len(extra_candidates) - skipped} 个, "
      f"{'全部 PASS' if not extra_failed else f'{extra_failed} 个 FAIL'}")
if extra_failed:
    failed += extra_failed
print("=" * 70)

# ── 代理白名单测试（用户可配置：命中即直连，最优先，含 IPv4/IPv6/CIDR） ─────
print("\n" + "=" * 70)
print("代理白名单: 命中规则强制直连（优先于域名表/固定代理，支持 IPv4/IPv6/CIDR）")
print("=" * 70)

wl_cases = [
    # (规则, 测试host, 期望命中)
    ("example.com",         "example.com",          True),
    ("example.com",         "www.example.com",      True),
    ("example.com",         "sub.www.example.com",  True),
    ("example.com",         "example.org",          False),
    ("claude.ai",           "claude.ai",            True),   # 域名表强制隧道，白名单优先
    ("localhost",           "localhost",            True),
    ("8.8.8.8",             "8.8.8.8",              True),   # IPv4 字面量
    ("8.8.8.8",             "8.8.4.4",              False),
    ("10.0.0.0/8",          "10.1.2.3",             True),   # IPv4 CIDR
    ("192.0.2.0/24",        "192.0.2.55",           True),   # IPv4 CIDR
    ("2001:db8::/32",       "2001:db8::1",          True),   # IPv6 网段
    ("fe80::1",             "fe80::1",              True),   # IPv6 单地址
]

# 注入白名单规则（绕过 settings.json 直接改内存缓存；版本 +1 使匹配缓存失效），
# 另加一条"禁用"的规则验证不参与匹配
pp._WHITELIST = [{"value": "203.0.113.0/24", "enabled": False}] + \
                [{"value": r, "enabled": True} for r, _, _ in wl_cases]
wl_ver = pp._WHITELIST_VERSION + 1
pp._WHITELIST_VERSION = wl_ver

wl_failed = 0
for rule, host, expect in wl_cases:
    got = _is_whitelisted(host, wl_ver)
    ok = got == expect
    wl_failed += 0 if ok else 1
    print(f"规则 {rule:<16} host {host:<22} 命中={got} 期望={expect} -> {'PASS' if ok else 'FAIL'}")
got = _is_whitelisted("203.0.113.9", wl_ver)  # 禁用的规则不应命中
print(f"禁用规则 203.0.113.0/24 host 203.0.113.9 -> 命中={got} 期望=False -> {'PASS' if not got else 'FAIL'}")
wl_failed += 0 if not got else 1

# 白名单在三种模式下都强制直连（固定代理模式原本全部走隧道）
print("\n白名单在三种模式下的决策（命中白名单应为直连）:")
for mode_name, kw in MODES:
    got = _decide_direct("www.example.com", **kw)
    ok = got is True
    wl_failed += 0 if ok else 1
    print(f"  {mode_name:<16} www.example.com -> {PATH[got]} (期望 直连) {'PASS' if ok else 'FAIL'}")
# 白名单优先于域名表：claude.ai 在 pac_domains.json 中强制隧道，白名单命中应直连
got = _decide_direct("claude.ai", bypass_lan=True, smart=True)
ok = got is True
wl_failed += 0 if ok else 1
print(f"  {'域名表冲突':<14} claude.ai -> {PATH[got]} (期望 直连，白名单优先) {'PASS' if ok else 'FAIL'}")
# 未命中白名单的境外域名仍走隧道（白名单不影响其他路由）
got = _decide_direct("google.com", bypass_lan=True, smart=True)
ok = got is False
wl_failed += 0 if ok else 1
print(f"  {'未命中白名单':<14} google.com -> {PATH[got]} (期望 隧道) {'PASS' if ok else 'FAIL'}")

# 恢复空白名单，避免影响后续用例
pp._WHITELIST = None
pp._WHITELIST_VERSION += 1
if wl_failed:
    failed += wl_failed
print("=" * 70)

# ── IPv6-only 节点标注 ─────────────────────────────────────
# IPv6 专用节点只提供 AAAA 记录，本机没有 IPv6 出口时连不上：
# 客户端据此在节点菜单标注并置灰。判定三级：ipv6_only 字段 →
# remark 里的 "(IPv6)"（服务端 rrpp_client 现网唯一的线索）→ 域名解析兜底。
print("\n" + "=" * 70)
print("IPv6 专用节点：识别（三级判定）与可达性")
print("=" * 70)

ipv6_failed = 0
has_v6 = pp._local_has_ipv6()
print(f"本机是否有全局 IPv6 地址: {has_v6}")

# 用真实 nodes.json（test 目录）；没有则跳过
_nodes_file = TEST_DIR / "nodes.json"
real_nodes = json.loads(_nodes_file.read_text("utf-8")) if _nodes_file.is_file() else []
if real_nodes:
    pp._IPV6_OK.clear()   # 清缓存，做真实探测
    for n in real_nodes:
        c = n.get("country") or "?"
        io = pp._node_ipv6_only(n)
        reach = pp._node_reachable(n)
        mark = "[IPv6]" if reach else "[不可用：本机无 IPv6]"
        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 时必须判定为不可达，否则必须可达
        if io:
            expect = has_v6 or reach  # 本机有 IPv6 -> 应可达；无 IPv6 -> 不应可达
            ok = reach == expect
            ipv6_failed += 0 if ok else 1
            print(f"        IPv6 专用节点可达性: {reach} (期望 {expect}) "
                  f"{'PASS' if ok else 'FAIL'}")
        else:
            ok = reach is True
            ipv6_failed += 0 if ok else 1
            print(f"        普通节点恒可达: {reach} {'PASS' if ok else 'FAIL'}")
    # 缓存生效：再次判定不发新探测（结果一致）
    cached = list(pp._IPV6_OK.items())
    if cached:
        print(f"  可达性缓存: {cached}")

# 三级判定：字段 → remark 里的 "(IPv6)" 标记 → 域名解析兜底。
# 服务端 subscription?method=rrpp_client 只吐旧扁平格式，ipv6_only 字段被丢掉，
# 只在 remark 里留 "(IPv6)"（后端 countries_to_rrpp_client），故第 2 级是现网主路径。
synthetic = [
    ({"domain": "example.com"}, False, "无标记（example.com 有 A 记录）"),
    ({"domain": "example.com", "ipv6_only": False}, False, "ipv6_only=false"),
    ({"domain": "example.com", "ipv6_only": True}, True, "ipv6_only=true"),
    ({"domain": "example.com", "remark": "GlobalVM 350G (IPv6)"}, True, "remark 含 (IPv6)"),
    ({"domain": "example.com", "remark": "Azure 15G"}, False, "remark 无 IPv6 标记"),
    ({"domain": "example.com", "remark": "IPv6 专用机"}, True, "remark 含 IPv6（无括号）"),
    ({"domain": "example.com", "remark": "some_ipv6_thing"}, True, "remark 小写 ipv6"),
]
for node, expect, desc in synthetic:
    got = pp._node_ipv6_only(node)
    ok = got == expect
    ipv6_failed += 0 if ok else 1
    print(f"  {desc:<24} _node_ipv6_only={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
pp._IPV6_OK.clear()   # 合成用例可能污染缓存

if ipv6_failed:
    failed += ipv6_failed
print("=" * 70)

# ── CONNECT authority 解析：IPv6 字面量带方括号 ──────────────────
print("\n" + "=" * 70)
print("CONNECT authority 解析（IPv6 字面量剥方括号）")
print("=" * 70)

conn_cases = [
    ("CONNECT [::1]:443 HTTP/1.1",           ("::1", 443)),
    ("CONNECT example.com:443 HTTP/1.1",     ("example.com", 443)),
    ("CONNECT iam.aeolianchi.me:443 HTTP/1.1", ("iam.aeolianchi.me", 443)),
    ("CONNECT [::1] HTTP/1.1",               None),   # 无端口
    ("CONNECT example.com HTTP/1.1",         None),   # 无端口
    ("CONNECT [fe80::1%25eth0]:8443 HTTP/1.1", ("fe80::1%25eth0", 8443)),
    ("CONNECT 127.0.0.1:1080 HTTP/1.1",      ("127.0.0.1", 1080)),
]
conn_failed = 0
for line, expect in conn_cases:
    got = pp._parse_connect_authority(line)
    ok = got == expect
    conn_failed += 0 if ok else 1
    print(f"  {line:<45} -> {got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
if conn_failed:
    failed += conn_failed
print("=" * 70)

# ── 上游代理认证注入：按完整头名精确匹配 ────────────────────────
print("\n" + "=" * 70)
print("认证注入：仅精确的 authorization 头名阻止注入")
print("=" * 70)

def _auth_lines(data: bytes):
    head = data.split(b"\r\n\r\n", 1)[0]
    return [l for l in head.split(b"\r\n")[1:] if l.lower().startswith(b"authorization:")]

auth_cases = [
    # (描述, 请求, 期望)：数字=注入后 authorization 行数；"unchanged"=原样跳过
    ("无任何认证头",      b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nHost: iam.aeolianchi.me\r\n\r\n", 1),
    ("Proxy-Authorization", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nProxy-Authorization: Basic xyz\r\n\r\n", 1),
    ("已有 Authorization", b"GET http://example.com/ HTTP/1.1\r\nAuthorization: Bearer tok\r\n\r\n", "unchanged"),
    ("小写 authorization", b"GET http://example.com/ HTTP/1.1\r\nauthorization: basic abc\r\n\r\n", "unchanged"),
    ("X-Authorization 前缀", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nX-Authorization-Go: 1\r\n\r\n", 1),
    ("无头空块",          b"GET http://example.com/ HTTP/1.1\r\n\r\n", 1),
]
auth_failed = 0
for desc, req, expect in auth_cases:
    out = pp._inject_proxy_auth(req)
    if expect == "unchanged":
        ok = out == req
        print(f"  {desc:<22} 原样跳过 {'PASS' if ok else 'FAIL'}")
    else:
        lines = _auth_lines(out)
        ok = len(lines) == expect and out.split(b"\r\n", 2)[1] == lines[0]
        # 注入位置必须紧跟请求行（上游取首个 Authorization 时读到我们的认证）
        print(f"  {desc:<22} authorization 行数={len(lines)} (期望 {expect}) "
              f"{'PASS' if ok else 'FAIL'}")
    auth_failed += 0 if ok else 1
if auth_failed:
    failed += auth_failed
print("=" * 70)

# ── 本程序 PAC 地址判定（启动清理只认自己的残留） ────────────────
print("\n" + "=" * 70)
print("PAC 地址归属判定（_is_our_pac_url）")
print("=" * 70)

pac_cases = [
    ("http://127.0.0.1:39453/proxy.pac", True),     # 本程序 PAC 服务
    ("http://127.0.0.1:7897/pac?auth=xyz", False),  # 其他代理软件（如 Clash）
    ("http://127.0.0.1:1080/other.pac", False),     # 路径不符
    ("http://192.168.1.1:8080/proxy.pac", False),   # 非本机
    ("", False),
]
pac_failed = 0
for url, expect in pac_cases:
    got = pp._is_our_pac_url(url)
    ok = got == expect
    pac_failed += 0 if ok else 1
    print(f"  {url:<42} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
if pac_failed:
    failed += pac_failed
print("=" * 70)

# ── 强制代理规则格式：后缀 / full / regexp + 关键域名兜底 ───────────
print("\n" + "=" * 70)
print("强制代理规则格式（后缀 / full / regexp / 关键域名兜底）")
print("=" * 70)

rule_failed = 0
rule_dir = Path(tempfile.mkdtemp()) / "RRPP"
rule_dir.mkdir(parents=True)
(rule_dir / "pac_domains.json").write_text(json.dumps({"domains": [
    "suffix.example", "full:only.example", "regexp:^edge-[0-9]+\\.example$",
]}), "utf-8")

# 此段隔离真实 CDN 表，确保三种语义分别受测。
old_user_dir = pp._USER_DIR
old_suffixes = pp._FORCE_PROXY_SUFFIXES
old_exact = pp._FORCE_PROXY_EXACT
old_regexes = pp._FORCE_PROXY_REGEXES
old_loaded = pp._FORCE_PROXY_LOADED
pp._USER_DIR = rule_dir
pp._FORCE_PROXY_SUFFIXES = None
pp._FORCE_PROXY_EXACT = None
pp._FORCE_PROXY_REGEXES = None
pp._FORCE_PROXY_LOADED = False
pp._is_force_proxy.cache_clear()
pp._load_force_proxy_suffixes()

rule_cases = [
    ("普通后缀匹配子域", "cdn.suffix.example", True),
    ("full 精确匹配", "only.example", True),
    ("full 不匹配子域", "cdn.only.example", False),
    ("regexp 匹配", "edge-42.example", True),
    ("regexp 不匹配", "edge-x.example", False),
    ("关键 Google CDN 兜底", "assets.googlecdn.com", True),
    ("无关域名", "unmatched.example", False),
]
for desc, host, expect in rule_cases:
    got = pp._is_force_proxy(host)
    ok = got == expect
    rule_failed += 0 if ok else 1
    print(f"  {desc:<22} {host:<30} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")

pp._USER_DIR = old_user_dir
pp._FORCE_PROXY_SUFFIXES = old_suffixes
pp._FORCE_PROXY_EXACT = old_exact
pp._FORCE_PROXY_REGEXES = old_regexes
pp._FORCE_PROXY_LOADED = old_loaded
pp._is_force_proxy.cache_clear()
if rule_failed:
    failed += rule_failed
print("=" * 70)

# ── 明确直连域名：优先级 / Trie / 固定代理隔离 ───────────────────
print("\n" + "=" * 70)
print("明确直连域名（后缀 / full / regexp / 强制代理优先 / 固定代理隔离）")
print("=" * 70)

direct_failed = 0
direct_dir = Path(tempfile.mkdtemp()) / "RRPP"
direct_dir.mkdir(parents=True)
(direct_dir / "pac_direct_domains.json").write_text(json.dumps({"domains": [
    "direct.example", "full:only-direct.example", "regexp:^edge-direct-[0-9]+\\.example$",
    "claude.ai",
]}), "utf-8")
old_user_dir = pp._USER_DIR
old_direct_trie = pp._DIRECT_SUFFIX_TRIE
old_direct_exact = pp._DIRECT_EXACT
old_direct_regexes = pp._DIRECT_REGEXES
old_direct_loaded = pp._DIRECT_LOADED
old_data_path = pp._data_path
try:
    pp._USER_DIR = direct_dir
    pp._data_path = lambda name: None  # 隔离仓库内置种子，覆盖到临时用户数据。
    pp._DIRECT_SUFFIX_TRIE = None
    pp._DIRECT_EXACT = None
    pp._DIRECT_REGEXES = None
    pp._DIRECT_LOADED = False
    pp._is_direct_domain.cache_clear()
    pp._load_direct_domains()
    direct_cases = [
        ("后缀匹配子域", "cdn.direct.example", True),
        ("full 精确匹配", "only-direct.example", True),
        ("full 不匹配子域", "cdn.only-direct.example", False),
        ("regexp 匹配", "edge-direct-42.example", True),
        ("regexp 不匹配", "edge-direct-x.example", False),
    ]
    for desc, host, expect in direct_cases:
        got = pp._is_direct_domain(host)
        ok = got == expect
        direct_failed += 0 if ok else 1
        print(f"  {desc:<22} {host:<30} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
    smart_direct = pp._decide_direct("cdn.direct.example", True, True)
    fixed_proxy = not pp._decide_direct("cdn.direct.example", True, False)
    force_wins = not pp._decide_direct("claude.ai", True, True)
    direct_target = pp._direct_connect_target("cdn.direct.example", 443, True, True)
    target_keeps_host = direct_target == ("cdn.direct.example", 443)
    print(f"  PAC 模式命中明确直连: {smart_direct} {'PASS' if smart_direct else 'FAIL'}")
    print(f"  固定代理模式不使用明确直连表: {fixed_proxy} {'PASS' if fixed_proxy else 'FAIL'}")
    print(f"  强制代理优先于明确直连: {force_wins} {'PASS' if force_wins else 'FAIL'}")
    print(f"  明确直连保持原始主机名: {target_keeps_host} {'PASS' if target_keeps_host else 'FAIL'}")
    direct_failed += 0 if smart_direct else 1
    direct_failed += 0 if fixed_proxy else 1
    direct_failed += 0 if force_wins else 1
    direct_failed += 0 if target_keeps_host else 1
    pp._pac_write("pac_direct_domains", {"domains": ["updated-direct.example"]})
    pp._invalidate_pac_routing_cache("pac_direct_domains")
    direct_reloaded = (not pp._is_direct_domain("direct.example")
                       and pp._is_direct_domain("updated-direct.example"))
    print(f"  明确直连表热更新立即生效: {direct_reloaded} "
          f"{'PASS' if direct_reloaded else 'FAIL'}")
    direct_failed += 0 if direct_reloaded else 1
finally:
    pp._USER_DIR = old_user_dir
    pp._data_path = old_data_path
    pp._DIRECT_SUFFIX_TRIE = old_direct_trie
    pp._DIRECT_EXACT = old_direct_exact
    pp._DIRECT_REGEXES = old_direct_regexes
    pp._DIRECT_LOADED = old_direct_loaded
    pp._is_direct_domain.cache_clear()
if direct_failed:
    failed += direct_failed
print("=" * 70)

# ── PAC 更新立即生效 + DNS 解析一致性 ──────────────────────────
print("\n" + "=" * 70)
print("PAC 热更新缓存失效 + 国内直连使用同一次 DNS 解析结果")
print("=" * 70)

hot_failed = 0
hot_dir = Path(tempfile.mkdtemp()) / "RRPP"
hot_dir.mkdir(parents=True)
old_user_dir = pp._USER_DIR
old_cn_ranges = pp._CN_RANGES
old_cn_ranges_v6 = pp._CN_RANGES_V6
old_cn_loaded = pp._CN_RANGES_LOADED
old_suffixes = pp._FORCE_PROXY_SUFFIXES
old_exact = pp._FORCE_PROXY_EXACT
old_regexes = pp._FORCE_PROXY_REGEXES
old_force_loaded = pp._FORCE_PROXY_LOADED
old_getaddrinfo = socket.getaddrinfo
try:
    pp._USER_DIR = hot_dir
    pp._CN_RANGES = None
    pp._CN_RANGES_V6 = None
    pp._CN_RANGES_LOADED = False
    pp._FORCE_PROXY_SUFFIXES = None
    pp._FORCE_PROXY_EXACT = None
    pp._FORCE_PROXY_REGEXES = None
    pp._FORCE_PROXY_LOADED = False
    pp._is_china_ip.cache_clear()
    pp._is_force_proxy.cache_clear()

    pp._pac_write("pac_domains", {"domains": ["old-rule.example"]})
    pp._pac_write("pac_ip_cn", {"cidrs": ["114.114.114.0/24"]})
    pp._load_force_proxy_suffixes()
    pp._load_cn_ranges()
    before = pp._is_force_proxy("old-rule.example")
    pp._pac_write("pac_domains", {"domains": ["new-rule.example"]})
    pp._invalidate_pac_routing_cache("pac_domains")
    domains_reloaded = (not pp._is_force_proxy("old-rule.example")
                        and pp._is_force_proxy("new-rule.example"))
    print(f"  域名表热更新立即生效: 初始={before}, 更新后={domains_reloaded} "
          f"{'PASS' if before and domains_reloaded else 'FAIL'}")
    hot_failed += 0 if before and domains_reloaded else 1

    initial_cn = pp._is_china_ip("114.114.114.114")
    pp._pac_write("pac_ip_cn", {"cidrs": ["8.8.8.0/24"]})
    pp._invalidate_pac_routing_cache("pac_ip_cn")
    ip_reloaded = (not pp._is_china_ip("114.114.114.114")
                   and pp._is_china_ip("8.8.8.8"))
    print(f"  IP 段热更新立即生效: 初始={initial_cn}, 更新后={ip_reloaded} "
          f"{'PASS' if initial_cn and ip_reloaded else 'FAIL'}")
    hot_failed += 0 if initial_cn and ip_reloaded else 1

    # 固定本次 DNS 返回值；_direct_connect_target 必须只解析一次并直连该 IP。
    pp._pac_write("pac_ip_cn", {"cidrs": ["114.114.114.0/24"]})
    pp._invalidate_pac_routing_cache("pac_ip_cn")
    dns_calls = []
    def fake_getaddrinfo(host, *args, **kwargs):
        dns_calls.append(host)
        if host == "mixed.example":
            ips = ["114.114.114.114", "8.8.8.8"]
        else:
            ips = ["114.114.114.114"]
        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 0)) for ip in ips]
    socket.getaddrinfo = fake_getaddrinfo
    target = pp._direct_connect_target("cn-cdn.example", 443, False, True)
    stable_dns = target == ("114.114.114.114", 443) and dns_calls == ["cn-cdn.example"]
    mixed = pp._direct_connect_target("mixed.example", 443, False, True)
    mixed_safe = mixed is None
    print(f"  国内 CDN 固定解析 IP 且不二次解析: {stable_dns} {'PASS' if stable_dns else 'FAIL'}")
    print(f"  混合中国/境外 CDN 记录改走隧道: {mixed_safe} {'PASS' if mixed_safe else 'FAIL'}")
    hot_failed += 0 if stable_dns else 1
    hot_failed += 0 if mixed_safe else 1

    # IPv6 中国 CIDR 与 IPv6 私网/回环分流。240e::/16 是测试用的公网 IPv6 段。
    pp._pac_write("pac_ip_cn", {"cidrs": ["240e::/16"]})
    pp._invalidate_pac_routing_cache("pac_ip_cn")
    ipv6_cn = (pp._is_china_ip("240e:1234::1")
               and pp._direct_connect_target("240e:1234::1", 443, False, True)
                   == ("240e:1234::1", 443))
    ipv6_private = all(pp._is_private_ip(ip) for ip in ("::1", "fc00::1", "fe80::1"))
    ipv6_public = not pp._is_private_ip("2001:4860:4860::8888")
    print(f"  IPv6 中国 CIDR 智能直连: {ipv6_cn} {'PASS' if ipv6_cn else 'FAIL'}")
    print(f"  IPv6 回环/ULA/链路本地直连: {ipv6_private and ipv6_public} "
          f"{'PASS' if ipv6_private and ipv6_public else 'FAIL'}")
    hot_failed += 0 if ipv6_cn else 1
    hot_failed += 0 if ipv6_private and ipv6_public else 1
finally:
    socket.getaddrinfo = old_getaddrinfo
    pp._USER_DIR = old_user_dir
    pp._CN_RANGES = old_cn_ranges
    pp._CN_RANGES_V6 = old_cn_ranges_v6
    pp._CN_RANGES_LOADED = old_cn_loaded
    pp._FORCE_PROXY_SUFFIXES = old_suffixes
    pp._FORCE_PROXY_EXACT = old_exact
    pp._FORCE_PROXY_REGEXES = old_regexes
    pp._FORCE_PROXY_LOADED = old_force_loaded
    pp._is_china_ip.cache_clear()
    pp._is_force_proxy.cache_clear()
if hot_failed:
    failed += hot_failed
print("=" * 70)

# ── 智能国内直连失败回退上游隧道 ───────────────────────────────
print("\n" + "=" * 70)
print("智能国内直连失败：仅该路径回退一次上游隧道")
print("=" * 70)

fallback_failed = 0
direct = ("114.114.114.114", 443)
upstream = ("node.example", 1205)
calls = []
remote_obj = object()
def fake_connect(target, timeout):
    calls.append((target, timeout))
    if target == direct:
        raise OSError("mock direct connect failure")
    if target == upstream:
        return remote_obj
    raise AssertionError(f"unexpected target: {target}")

remote, via_tunnel = pp._connect_route(
    direct, True, upstream[0], upstream[1], None, False, fake_connect)
fallback_ok = (remote is remote_obj and via_tunnel is True
               and calls == [(direct, 10), (upstream, 10)])
calls.clear()
try:
    pp._connect_route(direct, False, upstream[0], upstream[1], None, False, fake_connect)
    explicit_not_fallback = False
except OSError:
    explicit_not_fallback = calls == [(direct, 10)]
print(f"  智能中国公网直连失败回退隧道: {fallback_ok} {'PASS' if fallback_ok else 'FAIL'}")
print(f"  显式直连失败不回退: {explicit_not_fallback} "
      f"{'PASS' if explicit_not_fallback else 'FAIL'}")
fallback_failed += 0 if fallback_ok else 1
fallback_failed += 0 if explicit_not_fallback else 1
if fallback_failed:
    failed += fallback_failed
print("=" * 70)

# ── 启动连通性必须验证境外隧道 ─────────────────────────────────
print("\n" + "=" * 70)
print("启动连通性：google.com 隧道失败必须阻止“开启成功”")
print("=" * 70)

class _FakeResponse:
    def __enter__(self):
        return self
    def __exit__(self, *args):
        return False
    def read(self, size=-1):
        return b""

class _FakeOpener:
    def __init__(self, fail_google=False):
        self.fail_google = fail_google
    def open(self, url, timeout):
        if self.fail_google and "www.google.com" in url:
            raise OSError("mock tunnel failure")
        return _FakeResponse()


class _CancellingResponse(_FakeResponse):
    def __init__(self, cancel_event):
        self.cancel_event = cancel_event
    def read(self, size=-1):
        self.cancel_event.set()
        return b""


class _CancellingOpener:
    def __init__(self, cancel_event):
        self.cancel_event = cancel_event
        self.urls = []
    def open(self, url, timeout):
        self.urls.append(url)
        return _CancellingResponse(self.cancel_event)


class _GoogleCancellingOpener(_CancellingOpener):
    def open(self, url, timeout):
        self.urls.append(url)
        if "www.google.com" in url:
            return _CancellingResponse(self.cancel_event)
        return _FakeResponse()

verify_failed = 0
ok_results = pp._verify_proxy_connectivity(opener=_FakeOpener())
ok = ok_results.get("google.com") is True
try:
    pp._verify_proxy_connectivity(opener=_FakeOpener(fail_google=True))
    blocked = False
except RuntimeError:
    blocked = True
cancel_event = threading.Event()
cancelling_opener = _CancellingOpener(cancel_event)
try:
    pp._verify_proxy_connectivity(opener=cancelling_opener, cancel_event=cancel_event)
    cancel_blocked = False
except pp._StartCancelled:
    cancel_blocked = True
google_not_started = all("www.google.com" not in url for url in cancelling_opener.urls)
google_cancel_event = threading.Event()
google_cancelling_opener = _GoogleCancellingOpener(google_cancel_event)
try:
    pp._verify_proxy_connectivity(opener=google_cancelling_opener,
                                  cancel_event=google_cancel_event)
    google_cancelled = False
except pp._StartCancelled:
    google_cancelled = True
google_started = any("www.google.com" in url for url in google_cancelling_opener.urls)
print(f"  境外隧道成功可通过: {ok} {'PASS' if ok else 'FAIL'}")
print(f"  境外隧道失败会阻止开启: {blocked} {'PASS' if blocked else 'FAIL'}")
print(f"  Baidu 探测中取消会中断后续 Google 探测: {cancel_blocked and google_not_started} "
      f"{'PASS' if cancel_blocked and google_not_started else 'FAIL'}")
print(f"  Google 探测中取消会终止本次开启: {google_cancelled and google_started} "
      f"{'PASS' if google_cancelled and google_started else 'FAIL'}")
verify_failed += 0 if ok else 1
verify_failed += 0 if blocked else 1
verify_failed += 0 if cancel_blocked and google_not_started else 1
verify_failed += 0 if google_cancelled and google_started else 1
if verify_failed:
    failed += verify_failed
print("=" * 70)

# ── DNS 分流 TTL 缓存 ────────────────────────────────────────
print("\n" + "=" * 70)
print("TTL 缓存：cache_clear 兼容 + 缓存命中")
print("=" * 70)

ttl_failed = 0
assert hasattr(pp._is_china_ip, "cache_clear"), "cache_clear 缺失"
assert hasattr(pp._is_private_ip, "cache_clear"), "cache_clear 缺失"
assert hasattr(pp._is_whitelisted, "cache_clear"), "cache_clear 缺失"
pp._is_china_ip.cache_clear()
pp._is_private_ip.cache_clear()
pp._is_whitelisted.cache_clear()
a1, a2 = pp._is_private_ip("10.1.2.3"), pp._is_private_ip("10.1.2.3")
ok = a1 is True and a2 is True
ttl_failed += 0 if ok else 1
print(f"  内网 IP 判定缓存命中: {a1}/{a2} {'PASS' if ok else 'FAIL'}")
b1, b2 = pp._is_china_ip("114.114.114.114"), pp._is_china_ip("114.114.114.114")
ok = b1 is True and b2 is True
ttl_failed += 0 if ok else 1
print(f"  中国 IP 判定缓存命中: {b1}/{b2} {'PASS' if ok else 'FAIL'}")
pp._is_china_ip.cache_clear()
pp._is_private_ip.cache_clear()
print(f"  cache_clear 后重查: {pp._is_china_ip('114.114.114.114')} "
      f"{'PASS' if pp._is_china_ip('114.114.114.114') is True else 'FAIL'}")
if ttl_failed:
    failed += ttl_failed
print("=" * 70)

print(f"\n最终结论: {'全部 PASS' if not failed else f'{failed} 个 FAIL'}")
sys.exit(1 if failed else 0)