ykxmz/_tests/test_indicator_visible.py

# -*- coding: utf-8 -*-
"""验证「屏幕共享时被控端一定显示可见提示」这条安全属性。

界面重做之后：主控制窗口已经删掉了，被控端默认静默运行。
所以这个测试改成：

    用 --add-device 启动（静默 + 临时开放配对）
    → 配对
    → screen_start
    → 用 EnumWindows 枚举被控端进程的窗口，**必须**出现那条红色提示
    → 检查它在右下角
    → screen_stop
    → 提示必须消失

注意：静默模式**不显示**主窗口，但屏幕共享提示**保留** —— 这是故意的：
有人在看你的屏幕时，屏幕前面的人应该知道。
"""
from __future__ import annotations

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

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

AGENT = ROOT / "windows" / "run_agent.py"
DATA = ROOT / "windows" / "data"
PORT = 8893

PASSED = 0
FAILED = 0
NOTICE_TITLE = "屏幕正在被共享"


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(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)
            rect = ctypes.wintypes.RECT() if hasattr(ctypes, "wintypes") else None
            left = top = width = height = 0
            try:
                box = ctypes.create_string_buffer(16)
                user32.GetWindowRect(hwnd, box)
                values = ctypes.cast(box, ctypes.POINTER(ctypes.c_long * 4)).contents
                left, top, right, bottom = values
                width, height = right - left, bottom - top
            except Exception:                             # noqa: BLE001
                pass
            found.append({
                "title": buffer.value,
                "visible": bool(user32.IsWindowVisible(hwnd)),
                "rect": (left, top, width, height),
            })
        return True

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


def notice_windows(pid: int) -> list:
    return [w for w in windows_of(pid)
            if NOTICE_TITLE in (w["title"] or "") and w["visible"]]


def wait_notice(pid: int, want: bool, timeout: float = 8.0) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        found = notice_windows(pid)
        if bool(found) == want:
            return True
        time.sleep(0.3)
    return False


class Talk:
    def __init__(self, port, timeout=10.0):
        self.sock = socket.create_connection(("127.0.0.1", port), timeout=timeout)
        self.sock.settimeout(timeout)
        self.buf = b""

    def call(self, obj, timeout=12.0):
        self.sock.sendall((json.dumps(obj, ensure_ascii=False) + "\n").encode("utf-8"))
        deadline = time.time() + timeout
        while True:
            while b"\n" not in self.buf:
                if time.time() > deadline:
                    raise TimeoutError("超时")
                chunk = self.sock.recv(65536)
                if not chunk:
                    raise ConnectionError("断开")
                self.buf += chunk
            line, _, self.buf = self.buf.partition(b"\n")
            message = json.loads(line.decode("utf-8"))
            if "event" in message:          # 主动推送的事件不是回复
                continue
            return message

    def close(self):
        try:
            self.sock.close()
        except OSError:
            pass


def seed(data_dir: Path) -> None:
    """准备好数据目录。

    ⚠ 这里**故意把 show_screen_notice 设成 True**。

    为什么：这个测试验的是"提示条显示出来的时候位置和尺寸对不对"，
    而 2.1.7 起**默认是不显示的**（用户要求"像 UU 远程一样啥窗口都没有"）。
    开着开关跑，才能继续验到提示条本身还在正常工作 ——
    否则这个能力哪天被改坏了也没人知道。

    "默认不显示"这件事由下面的第六节单独验（不重启进程，直接读源码/默认值）。
    """
    data_dir.mkdir(parents=True, exist_ok=True)
    (data_dir / "settings.json").write_text(
        json.dumps({"setup_done": True, "show_screen_notice": True},
                   ensure_ascii=False), encoding="utf-8")


def main() -> int:
    backup = None
    if DATA.exists():
        backup = DATA.parent / f"data_bk_ind_{int(time.time())}"
        shutil.move(str(DATA), str(backup))

    proc = None
    try:
        seed(DATA)
        print("启动被控端（静默 + 临时开放配对）…")
        proc = subprocess.Popen(
            [sys.executable, str(AGENT), "--port", str(PORT),
             "--new-code", "--add-device", "10"],
            cwd=str(ROOT / "windows"),
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

        # 先等端口真的在监听 —— 只看 settings.json 会误判：
        # 那个文件在服务端启动之前就写好了，
        # 而且如果实例锁被别的进程占着，被控端会直接退出。
        settings = DATA / "settings.json"
        listening = False
        deadline = time.time() + 30
        while time.time() < deadline:
            if proc.poll() is not None:
                print(f"  被控端启动后退出了（退出码 {proc.returncode}）")
                log = DATA / "silent.log"
                if log.is_file():
                    print("  silent.log:")
                    for line in log.read_text(encoding="utf-8",
                                              errors="replace").splitlines()[-6:]:
                        print(f"    {line}")
                return 1
            try:
                probe = socket.create_connection(("127.0.0.1", PORT), timeout=1)
                probe.close()
                listening = True
                break
            except OSError:
                time.sleep(0.4)
        check("被控端在监听", listening)
        if not listening:
            return 1

        code = ""
        try:
            code = json.loads(settings.read_text(
                encoding="utf-8")).get("pairing_code", "")
        except Exception:                                 # noqa: BLE001
            code = ""
        check("生成配对码", bool(code), code)
        if not code:
            return 1

        # 静默模式：不该有任何窗口
        time.sleep(1.5)
        print()
        print("一、静默模式本身不该有窗口")
        visible = [w for w in windows_of(proc.pid) if w["visible"]]
        check("启动后没有可见窗口", not visible, str(visible[:3]))

        print()
        print("二、配对")
        talk = Talk(PORT)
        reply = talk.call({"cmd": "pair", "deviceId": "ind-test",
                           "deviceName": "提示条测试", "code": code})
        check("配对成功", reply.get("ok") is True, str(reply)[:120])

        print()
        print("三、开始共享 → 必须出现提示")
        check("共享前没有提示条", not notice_windows(proc.pid))
        reply = talk.call({"cmd": "screen_start", "maxWidth": 320,
                           "quality": 40, "fps": 2}, timeout=15)
        check("开始屏幕共享", reply.get("ok") is True, str(reply)[:120])
        appeared = wait_notice(proc.pid, True)
        check("★ 共享中出现了红色提示条", appeared,
              str([w["title"] for w in windows_of(proc.pid) if w["visible"]]))
        if appeared:
            found = notice_windows(proc.pid)
            left, top, width, height = found[0]["rect"]
            user32 = ctypes.windll.user32
            screen_w = user32.GetSystemMetrics(0)
            screen_h = user32.GetSystemMetrics(1)
            print(f"       提示条位置: ({left},{top}) {width}x{height}，"
                  f"屏幕 {screen_w}x{screen_h}")
            check("提示条在右下角",
                  left > screen_w * 0.5 and top > screen_h * 0.5,
                  f"({left},{top})")
            check("提示条可见尺寸合理", width > 40 and height > 10,
                  f"{width}x{height}")

        print()
        print("四、停止共享 → 提示必须消失")
        reply = talk.call({"cmd": "screen_stop"}, timeout=15)
        check("停止屏幕共享", reply.get("ok") is True, str(reply)[:120])
        check("★ 停止后提示条消失", wait_notice(proc.pid, False))

        print()
        print("五、再次共享 → 提示再次出现")
        talk.call({"cmd": "screen_start", "maxWidth": 320,
                   "quality": 40, "fps": 2}, timeout=15)
        check("★ 提示条再次出现", wait_notice(proc.pid, True))
        talk.call({"cmd": "screen_stop"}, timeout=15)
        check("再次停止后消失", wait_notice(proc.pid, False))

        # ------------------------------------------------------------------
        print()
        print("六、默认必须是**不显示**（用户 2.1.7 的要求）")
        #
        # 用户原话："现在不是要会弹一个屏幕已被共享吗？以后不需要弹，
        #           啥窗户都没有，就可以像 UU 远程一样"
        #
        # 这一节**不重启进程**（重启一次要十几秒），
        # 直接查默认值 —— 被控端读取设置时，没写这一项就用默认值，
        # 所以默认值是 False 就意味着"新装的机器不会弹提示条"。
        sys.path.insert(0, str(ROOT / "windows"))
        from lanremote.config import DEFAULT_SETTINGS
        check("★ 默认 show_screen_notice == False",
              DEFAULT_SETTINGS.get("show_screen_notice") is False,
              "拿到 %r —— 不是 False 的话新装机还会弹提示条"
              % DEFAULT_SETTINGS.get("show_screen_notice"))

        # 而且设置关着的时候，代码路径上**连窗口都不该建**
        headless_src = (ROOT / "windows" / "lanremote"
                        / "headless.py").read_text(encoding="utf-8")
        m = re.search(r"def set_screen_sharing\(self.*?(?=\n    def )",
                      headless_src, re.S)
        body = m.group(0) if m else ""
        check("★ 关着时直接 return，不建窗口",
              'store.get("show_screen_notice")' in body
              and "return" in body,
              "开关没接进 set_screen_sharing")

        talk.close()

    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))
        except Exception as exc:                          # noqa: BLE001
            print(f"  !! 还原失败: {exc}")

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


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