ykxmz/_tests/test_relay_multi.py

# -*- coding: utf-8 -*-
"""测被控端的多中转地址故障切换。

用真实的设备中转（192.168.2.156:8080）当可用地址，
故意在前面塞一个不通的地址，看它会不会自动跳到可用的那个。
"""
import sys
import tempfile
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "windows"))

from lanremote.config import Store                          # noqa: E402
from lanremote.relay import RelayLink                       # noqa: E402

GOOD = "http://192.168.2.156:8080"
BAD = "http://192.0.2.1:9"          # TEST-NET-1，保证不通

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))


def main() -> int:
    logs = []

    def log(message):
        logs.append(str(message))
        print(f"        {message}")

    print("一、先确认设备中转是通的")
    store = Store(Path(tempfile.mkdtemp()))
    store.set_pairing_code("556677")
    store.set("server_name", "多地址测试机")

    probe_link = RelayLink(store, log=log)
    ok = probe_link._probe(GOOD, timeout=8)
    check("设备中转可访问", ok, GOOD)
    if not ok:
        print("        设备中转不通，后面的测试没意义")
        return 1
    check("故意给的坏地址确实不通", not probe_link._probe(BAD, timeout=3), BAD)

    print()
    print("二、只给可用地址")
    logs.clear()
    link = RelayLink(store, log=log)
    check("start 成功", link.start(GOOD))
    check("选中的就是它", link.url == GOOD, link.url)
    check("候选列表有 1 个", len(link.candidates) == 1, link.candidates)
    link.stop()

    print()
    print("三、坏地址在前，可用地址在后（关键测试）")
    logs.clear()
    link2 = RelayLink(store, log=log)
    check("start 成功", link2.start(f"{BAD};{GOOD}"))
    check("自动跳过了坏地址，选中有用的那个", link2.url == GOOD, link2.url)
    check("日志里说了在试", any("不通" in m for m in logs), logs[:3])
    check("候选列表有 2 个", len(link2.candidates) == 2, link2.candidates)
    link2.stop()

    print()
    print("四、分号 / 逗号 / 带空格 / 不带协议头 都要能解析")
    for raw, expect_first in ((f"{BAD}; {GOOD}", BAD),
                              (f"{BAD}, {GOOD}", BAD),
                              (f" {BAD} ; {GOOD} ", BAD),
                              (f"192.168.2.156:8080;{BAD}", GOOD)):
        link3 = RelayLink(store, log=lambda _m: None)
        link3.start(raw)
        check(f"解析 {raw[:34]!r} -> 选中 {link3.url}",
              link3.url == expect_first or link3.url == GOOD,
              link3.url)
        link3.stop()

    print()
    print("五、全部不通时也要能起来（留给主循环重试）")
    logs.clear()
    link4 = RelayLink(store, log=log)
    started = link4.start(f"{BAD};http://192.0.2.2:9")
    check("start 仍然返回 True", started is True)
    check("先返回第一个候选，等主循环重试",
          link4.url in (BAD, "http://192.0.2.2:9"), link4.url)
    check("状态里有候选列表", len(link4.status().get("candidates", [])) == 2,
          link4.status())
    link4.stop()

    print()
    print("六、状态查询")
    link5 = RelayLink(store, log=lambda _m: None)
    link5.start(f"{GOOD}")
    status = link5.status()
    check("status 里有 url", status.get("url") == GOOD, status.get("url"))
    check("status 里有 candidates", isinstance(status.get("candidates"), list),
          status.get("candidates"))
    link5.stop()

    print()
    print("=" * 60)
    print(f"通过 {passed} 项，失败 {failed} 项")
    return 0 if failed == 0 else 1


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