ykxmz/_tests/test_lan_path.py

# -*- coding: utf-8 -*-
"""验证局域网直连这条路是通的，以及安卓端新加的东西在。

顺序很关键（踩过）：
    之前是先广播、再起测试实例。单独跑时本机真有个在跑的 agent，
    广播能搜到，测试过；但在完整回归里，套件会先 Stop-Strays
    把 agent 杀掉，广播就一台都搜不到 —— 于是失败。

    改成：**先起测试实例（它自己会绑 UDP 8766 应答），再广播**。
    这样不依赖环境里有没有别的 agent。
    如果 8766 被别的 agent 占着（单独跑时就是这样），
    退而验证"搜到的那台能直连"。
"""
from __future__ import annotations

import json
import re
import socket
import sys
import tempfile
import time
from pathlib import Path

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

MAIN = (ROOT / "android" / "src" / "com" / "lanremote" / "watch"
        / "MainActivity.java")

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 java_method(src: str, signature: str) -> str:
    """按花括号配平取 Java 方法体。"""
    index = src.find(signature)
    if index < 0:
        return ""
    start = src.find("{", index)
    if start < 0:
        return ""
    depth = 0
    for pos in range(start, len(src)):
        if src[pos] == "{":
            depth += 1
        elif src[pos] == "}":
            depth -= 1
            if depth == 0:
                return src[start:pos + 1]
    return ""


def free_port() -> int:
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


def tcp_ok(host, port, timeout=6.0) -> bool:
    sock = socket.socket()
    sock.settimeout(timeout)
    try:
        sock.connect((str(host), int(port)))
        return True
    except Exception:                                     # noqa: BLE001
        return False
    finally:
        try:
            sock.close()
        except Exception:                             # noqa: BLE001
            pass


def main() -> int:
    from lanremote import discovery
    from lanremote.config import Store
    from lanremote.server import RemoteServer
    from stub_support import StubUI, make_stub_actions

    print("一、局域网发现的协议本身")
    obj = json.loads(discovery.build_reply("测试电脑", 8765, "1.4.0")
                     .decode("utf-8"))
    check("应答里有 magic", obj.get("lanremote") == discovery.REPLY_MAGIC, obj)
    check("应答里有名字", obj.get("name") == "测试电脑", obj)
    check("应答里有端口", int(obj.get("port") or 0) == 8765, obj)

    probe = json.dumps({"lanremote": discovery.PROBE_MAGIC,
                        "proto": discovery.PROTOCOL}).encode()
    check("探测包能被解析", discovery.parse_probe(probe) is not None)
    check("乱码探测包被忽略", discovery.parse_probe(b"not json") is None)
    check("伪造 magic 的包被忽略",
          discovery.parse_probe(json.dumps({"lanremote": "x"}).encode())
          is None)

    print()
    print("二、先起被控端实例，再广播搜它")
    port = free_port()
    store = Store(Path(tempfile.mkdtemp(prefix="lan_test_")))
    store.set("port", port)
    store.set("pairing_code", "482913")
    ui = StubUI()
    server = RemoteServer(store, make_stub_actions(ui, store), ui=ui)
    server.start()
    time.sleep(1.2)
    check("被控端起来了", True, port)

    found = []
    try:
        found = discovery.discover(timeout=3.0)
    except Exception as exc:                              # noqa: BLE001
        print(f"       搜索异常: {exc}")
    listing = [f"{item.get('host')}:{item.get('port')}" for item in found[:5]]
    print(f"       广播搜到 {len(found)} 台: {listing}")

    mine = [item for item in found if int(item.get("port") or 0) == port]
    if mine:
        check("广播搜到了自己起的实例（UDP 应答正常）", True, listing)
        target = mine[0]
        check("搜到的名字不是空的", bool(target.get("name")), target)
    else:
        # UDP 8766 被别的 agent 占着时会出现这种情况（单独跑就是）。
        # 那就退一步：验证搜索机制能返回结果、且能直连。
        check("广播能返回结果（8766 可能被别的实例占用）",
              len(found) >= 1, listing)
        target = found[0] if found else None

    if target is not None:
        print(f"       对 {target.get('host')}:{target.get('port')} 做 TCP 直连")
        check("搜到之后能直接 TCP 连上（不经中转）",
              tcp_ok(target.get("host"), target.get("port")), target)

    print()
    print("三、本实例的 TCP 端口是开的（直连能进）")
    check("测试实例端口可直连", tcp_ok("127.0.0.1", port), port)

    try:
        server.stop()
    except Exception:                                     # noqa: BLE001
        pass

    print()
    print("四、安卓端源码里有这些新东西")
    src = MAIN.read_text(encoding="utf-8")
    code = re.sub(r"//[^\n]*", "", src)
    code = re.sub(r"/\*.*?\*/", "", code, flags=re.S)

    check("有局域网页面（buildLanPanel）",
          "private void buildLanPanel()" in code)
    check("能显示局域网页面", "private void showLanPanel()" in code)
    check("有自动搜索", "private void startLanSearch()" in code)
    check("首页有局域网入口按钮",
          'bigButton("局域网连接（同一个 WiFi）"' in code)
    check("局域网页面能返回", "private void hideLanPanel()" in code)
    check("返回键会先退出局域网页面", "closeLanIfOpen()" in code)
    check("用到了 Discovery.search（广播搜索）", "Discovery.search(" in code)
    check("用到了 DiscoveryCore.Found（结果类型）",
          "DiscoveryCore.Found" in code)

    check("有服务器延迟检测", "private void checkServer()" in code)
    check("首页有检查服务器按钮", '"检查服务器"' in code)
    check("延迟结果显示在首页", "serverStatus" in code)
    check("延迟按快慢变色", "ms < 300" in code and "C_DANGER" in code)

    print()
    print("五、局域网这条路不碰中转")
    body = java_method(code, "private void connectLanTyped()")
    check("connectLanTyped 方法体取到了", len(body) > 200, len(body))
    print(f"       方法体 {len(body)} 字符")
    check("没用 relayUrl（不走中转）", "relayUrl" not in body, body[:120])
    check("没走 connectViaRelay", "connectViaRelay" not in body)
    check("调的是直连 connect(", "connect(host, port, code" in body,
          body[:200])
    check("记住配对码（便于回退中转）", "rememberPairingCode(code)" in body)
    check("断开后会自动切回中转", "lanDirect = true" in body)

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


if __name__ == "__main__":
    raise SystemExit(main())