ykxmz/_tests/test_lan_bench_guard.py

# -*- coding: utf-8 -*-
"""守住「局域网/中转对比基准」的两个坑。

为什么专门为测试脚本立一个测试：
    这个基准一度是红的，报告里写着「这是中转链路的产品 bug」——
    而实际上产品是好的，是基准自己有两个 bug：
      1. 两条路用了同一个配对码 -> 第 2 台被「一码一机」规则拒掉，
         而第 1 台虽然被杀、中转还以为在线，于是"看着成功、实际没人回包"
      2. relay_connect() 里 attach 用的还是旧码，attach 到死设备上
    这两个 bug 让一个已经修好的产品背了黑锅，而且**很难自己发现** ——
    因为症状（连接超时）看起来完全像产品问题。

    所以既要做源码守卫（防止改回去），也要真跑一遍短版本
    （防止断言与实现脱钩）。

覆盖：
    一、源码守卫：两条路必须用不同的码
    二、源码守卫：attach 必须用 relay_code
    三、源码守卫：start_agent 的判据必须认码（不能被僵尸设备骗过）
    四、真跑一遍短基准：两条路都要通
"""
from __future__ import annotations

import re
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
BENCH = ROOT / "_tests" / "bench_lan_vs_relay.py"

passed = 0
failed = 0


def check(name, ok, detail=""):
    global passed, failed
    if ok:
        passed += 1
        print("  [PASS] " + name)
    else:
        failed += 1
        print("  [FAIL] " + name + (("  —— " + str(detail)) if detail else ""))


def main() -> int:
    check("基准脚本存在", BENCH.is_file(), BENCH)
    if not BENCH.is_file():
        return 1
    src = BENCH.read_text(encoding="utf-8")

    # ---------------------------------------------------------- 一
    print("  ---- 一、两条路用不同的配对码 ----")
    has_lan = re.search(r'^\s*code\s*=\s*"(\d{6})"', src, re.M)
    has_relay = re.search(r'^\s*relay_code\s*=\s*"(\d{6})"', src, re.M)
    check("定义了局域网用的 code", bool(has_lan),
          "找不到 `code = \"数字\"`")
    check("定义了中转用的 relay_code", bool(has_relay),
          "找不到 `relay_code = ...`")
    if has_lan and has_relay:
        check("两个码不一样（否则第 2 台会被一码一机规则拒掉）",
              has_lan.group(1) != has_relay.group(1),
              "两个码都是 %s" % has_lan.group(1))

    # ---------------------------------------------------------- 二
    print("  ---- 二、attach 要用 relay_code ----")
    # 找出 relay_connect 函数体
    m = re.search(r"def relay_connect\(\):(.*?)(?=\n        def |\n        relay =)",
                  src, re.S)
    body = m.group(1) if m else ""
    check("能定位到 relay_connect", bool(body), "函数没找到或改名了")
    if body:
        check("relay_connect 里用的是 relay_code",
              "relay_code" in body,
              "里面还在用 code —— attach 会打到局域网那台死设备上")
        check("relay_connect 里没有裸 attach(code)",
              not re.search(r"attach\(\s*code\s*\)", body),
              "attach(code) 还在，要改成 attach(relay_code)")

    # ---------------------------------------------------------- 三
    print("  ---- 三、start_agent 的判据要认码 ----")
    m2 = re.search(r"def start_agent\(([^)]*)\):(.*?)(?=\n        def )",
                   src, re.S)
    body2 = m2.group(2) if m2 else ""
    check("start_agent 接受码参数", bool(m2) and "agent_code" in m2.group(1),
          m2.group(1) if m2 else "没找到")
    if body2:
        # 不能只判断"有没有在线设备"
        only_online = re.search(
            r'if\s+any\(d\.get\("online"\)', body2)
        check("没有「只看有没有在线设备」的松判据", not only_online,
              "僵尸设备会骗过它 —— 上一台被杀但中转还没清理")
        check("判据里比对了配对码", "agent_code" in body2,
              "应改成「在线且码正好是这一台要用的码」")

    # ---------------------------------------------------------- 四
    print("  ---- 四、真跑一遍短基准（两条路都要通）----")
    try:
        proc = subprocess.run(
            [sys.executable, "-u", str(BENCH),
             "--rounds", "6", "--relay-rounds", "6", "--relay-timeout", "40"],
            cwd=str(ROOT), capture_output=True, text=True,
            encoding="utf-8", errors="replace", timeout=420)
        out = (proc.stdout or "") + (proc.stderr or "")
        m3 = re.search(r"BENCH_RESULT\s+(.*)", out)
        result = m3.group(1) if m3 else ""
        check("基准跑完并输出了 BENCH_RESULT", bool(result),
              out[-300:] if not result else "")
        if result:
            nums = dict(re.findall(r"(\w+)=([\w/.\-]+)", result))
            lan_ok = nums.get("lan_ok", "0/0")
            relay_ok = nums.get("relay_ok", "0/0")
            check("局域网那条路通了", lan_ok.startswith("6/6"), lan_ok)
            check("中转那条路通了", relay_ok.startswith("6/6"), relay_ok)
    except subprocess.TimeoutExpired:
        check("基准在 420 秒内跑完", False, "超时")
    except Exception as exc:                              # noqa: BLE001
        check("基准能跑起来", False, exc)

    print()
    print("=" * 56)
    print("  通过 %d 项，失败 %d 项" % (passed, failed))
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())