ykxmz/_tests/test_silent_mode.py

# -*- coding: utf-8 -*-
"""验证静默模式。

三件事必须成立：
  1. 启动后**一个窗口都没有** —— 用 EnumWindows 枚举该进程的所有顶层窗口来证明，
     不是"看起来没有"
  2. 真的在监听，手表能连上
  3. **配对认证仍然生效** —— 静默模式默认不接受新设备，
     否则局域网里任何人都能接进来，那就成了后门
"""
from __future__ import annotations

import ctypes
import json
import shutil
import socket
import subprocess
import sys
import tempfile
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
AGENT = ROOT / "windows" / "run_agent.py"
DATA = ROOT / "windows" / "data"
PORT = 8877

PASSED = 0
FAILED = 0


def check(name, ok, detail=""):
    global PASSED, FAILED
    if ok:
        PASSED += 1
        print(f"  [PASS] {name}")
    else:
        FAILED += 1
        print(f"  [FAIL] {name}  {detail}")


def windows_of_process(pid: int) -> list:
    """枚举属于该进程的所有可见/不可见顶层窗口。"""
    user32 = ctypes.windll.user32
    found: list = []
    EnumProc = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p,
                                  ctypes.c_void_p)

    def callback(hwnd, _lparam):
        owner = ctypes.c_ulong()
        user32.GetWindowThreadProcessId(hwnd, ctypes.byref(owner))
        if owner.value == pid:
            length = user32.GetWindowTextLengthW(hwnd)
            buffer = ctypes.create_unicode_buffer(max(2, length + 1))
            user32.GetWindowTextW(hwnd, buffer, length + 1)
            cls = ctypes.create_unicode_buffer(256)
            user32.GetClassNameW(hwnd, cls, 256)
            found.append({
                "title": buffer.value,
                "class": cls.value,
                "visible": bool(user32.IsWindowVisible(hwnd)),
            })
        return True

    user32.EnumWindows(EnumProc(callback), 0)
    return found


def port_open(port: int) -> bool:
    try:
        sock = socket.create_connection(("127.0.0.1", port), timeout=1)
        sock.close()
        return True
    except OSError:
        return False



def seed_paired_device(data_dir):
    """预置一台已配设备。

    现在的规则是：**一台设备都没配对过时，配对是开着的**
    （这是修用户报的"设完码却连不上"）。所以想测
    "静默模式拒绝新配对"，得先让电脑认为已经有设备配过了。
    """
    import json as _json
    data_dir = Path(data_dir)
    data_dir.mkdir(parents=True, exist_ok=True)
    (data_dir / "devices.json").write_text(
        _json.dumps({"already-paired": {
            "id": "already-paired", "name": "已配的手表",
            "token": "seed-token", "last": 0}}, ensure_ascii=False),
        encoding="utf-8")


def seed_setup_done(data_dir):
    """预先置 setup_done：这些测试测的是别的功能，不该弹出首次设置窗口。"""
    import json as _json
    data_dir.mkdir(parents=True, exist_ok=True)
    settings = data_dir / "settings.json"
    payload = {}
    if settings.is_file():
        try:
            payload = _json.loads(settings.read_text(encoding="utf-8"))
        except Exception:
            payload = {}
    payload["setup_done"] = True
    settings.write_text(_json.dumps(payload, ensure_ascii=False, indent=2),
                        encoding="utf-8")

def main() -> int:
    backup = None
    if DATA.exists():
        backup = DATA.parent / f"data_backup_silent_{int(time.time())}"
        shutil.move(str(DATA), str(backup))
        print(f"  已备份真实 data")

    seed_setup_done(DATA)
    seed_paired_device(DATA)
    proc = None
    try:
        proc = subprocess.Popen(
            [sys.executable, str(AGENT), "--port", str(PORT), "--silent"],
            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        print(f"  已静默启动（PID {proc.pid}），等待监听…")

        listening = False
        for _ in range(30):
            time.sleep(0.5)
            if port_open(PORT):
                listening = True
                break

        check("静默模式起来了并在监听", listening)

        print()
        print("一、窗口检查（EnumWindows 枚举该进程的顶层窗口）")
        time.sleep(1.5)
        windows = windows_of_process(proc.pid)
        visible = [w for w in windows if w["visible"]]
        # Tk 会创建一个隐藏的辅助窗口，那不算"界面窗口"
        ui_windows = [w for w in visible if w["title"]]
        print(f"       该进程顶层窗口 {len(windows)} 个，"
              f"可见 {len(visible)} 个，可见且有标题 {len(ui_windows)} 个")
        for w in windows[:6]:
            print(f"         class={w['class']!r} title={w['title']!r} "
                  f"visible={w['visible']}")
        check("没有任何可见窗口", not visible,
              f"竟然有 {len(visible)} 个: {visible[:3]}")
        check("没有带标题的界面窗口", not ui_windows, str(ui_windows[:3]))

        print()
        print("二、终端检查")
        check("启动时没有附带控制台窗口",
              not any(w["class"] == "ConsoleWindowClass" for w in windows))

        print()
        print("三、★ 配对认证必须仍然生效")
        settings = DATA / "settings.json"
        code = ""
        if settings.is_file():
            try:
                code = json.loads(settings.read_text(encoding="utf-8")).get(
                    "pairing_code", "")
            except Exception:                             # noqa: BLE001
                pass
        print(f"       配对码 {code or '(无)'}，静默模式默认不接受新设备")

        # 故意用一个**错误的码** —— 安全边界是"码错必须被拒"。
        # （以前这里拿正确的码试也期待被拒，但后来改成
        #   "配对码本身就是密钥，码对就放行"，所以那样测就不对了。）
        wrong = "000000" if code != "000000" else "111111"

        sock = socket.create_connection(("127.0.0.1", PORT), timeout=5)
        sock.settimeout(8)
        request = {"cmd": "pair", "deviceId": "silent-test",
                   "deviceName": "测试", "code": wrong}
        sock.sendall((json.dumps(request, ensure_ascii=False) + "\n").encode("utf-8"))
        raw = b""
        while b"\n" not in raw:
            chunk = sock.recv(4096)
            if not chunk:
                break
            raw += chunk
        reply = json.loads(raw.split(b"\n")[0].decode("utf-8")) if raw else {}
        print(f"       配对回复: {reply}")
        # 行为变了：配对窗口关着时，**码错才拒**，码对放行。
        # 因为配对码本身就是密钥，老逻辑"首台配对后自动关窗"
        # 会让重装手表的人永久连不上（实测踩过）。
        check("静默模式下拿着错误的码配不上",
              reply.get("ok") is False,
              f"竟然配上了: {reply}")
        check("给出的理由是明确的",
              "配对" in str(reply.get("error", "")) or
              "关闭" in str(reply.get("error", "")),
              str(reply.get("error")))

        print()
        print("四、未配对的设备不能执行任何命令")
        sock.sendall((json.dumps({"cmd": "feature_list"},
                                 ensure_ascii=False) + "\n").encode("utf-8"))
        raw = b""
        while b"\n" not in raw:
            chunk = sock.recv(4096)
            if not chunk:
                break
            raw += chunk
        reply = json.loads(raw.split(b"\n")[0].decode("utf-8")) if raw else {}
        check("未配对时功能列表被拒绝", reply.get("ok") is False,
              f"竟然返回了: {str(reply)[:80]}")
        sock.close()

        print()
        print("五、日志文件")
        log = DATA / "silent.log"
        check("静默模式写了日志", log.is_file(), str(log))

    finally:
        if proc is not None:
            try:
                proc.terminate()
                proc.wait(timeout=8)
            except Exception:                             # noqa: BLE001
                try:
                    proc.kill()
                except Exception:                         # noqa: BLE001
                    pass
        time.sleep(1.0)
        try:
            if DATA.exists():
                shutil.rmtree(DATA, ignore_errors=True)
            if backup is not None and backup.exists():
                shutil.move(str(backup), str(DATA))
                print("  已还原真实 data")
        except Exception as exc:                          # noqa: BLE001
            print(f"  !! 还原失败: {exc}")

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


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