# -*- coding: utf-8 -*-
"""
P1-04 全局热键平台矩阵单测（适配器替身，不依赖真实 X11 显示 / 不新增依赖）

覆盖：
  H1 session_kind 矩阵（win32 / x11 / wayland / offscreen-unknown，mock 环境变量 + sys.platform）
  H2 hotkey_plan 路由（win32→RegisterHotKey 线程；x11→XGrabKey 线程；wayland/offscreen→None+明确日志）
  H3 X11HotkeyThread 成功路径（fake libX11：XGrabKey 参数正确、命中 keycode 发射 triggered、
     stop 后 XUngrabKey/XCloseDisplay 释放）
  H4 X11 注册失败（XGrabKey=0 键被占用 / 无显示 / 不支持的组合 → _registered=False，线程安静退出）
  H5 Windows GlobalHotkeyThread 行为保持（非 Windows run() 静默就绪退出；Windows 常量完整）

运行: PYTHONIOENCODING=utf-8 python tests/test_global_hotkey_platforms.py
"""
import contextlib
import io
import os
import socket
import sys
import threading
import time

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")

from PyQt6.QtWidgets import QApplication  # noqa: E402

from ui.views.system_tools import desktop_session as ds  # noqa: E402
from ui.views.system_tools import x11_hotkey as xh  # noqa: E402
from ui.views.system_tools import global_hotkey as gh  # noqa: E402

RESULTS = []
APP = QApplication.instance() or QApplication(sys.argv)


def check(name, cond, extra=""):
    RESULTS.append((name, bool(cond)))
    print(f"{'PASS' if cond else 'FAIL'}  {name}  {extra if not cond else ''}", flush=True)


def with_env(patch: dict, fn):
    saved = {k: os.environ.get(k) for k in patch}
    for k, v in patch.items():
        if v is None:
            os.environ.pop(k, None)
        else:
            os.environ[k] = v
    try:
        return fn()
    finally:
        for k, v in saved.items():
            if v is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = v


# ======================================================================
# H1. session_kind 矩阵
# ======================================================================
_orig_platform = ds.sys.platform

def _set_platform(p):
    ds.sys = type("FakeSys", (), {"platform": p})()


try:
    _set_platform("win32")
    check("H1.1 win32", ds.session_kind() == "win32")
    _set_platform("linux")
    check("H1.2 X11（DISPLAY 有、无 WAYLAND_DISPLAY）",
          with_env({"DISPLAY": ":0", "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": None,
                    "QT_QPA_PLATFORM": "xcb"}, lambda: ds.session_kind()) == "x11")
    check("H1.3 Wayland（WAYLAND_DISPLAY 有）",
          with_env({"WAYLAND_DISPLAY": "wayland-0", "DISPLAY": ":0", "XDG_SESSION_TYPE": None,
                    "QT_QPA_PLATFORM": "wayland"}, lambda: ds.session_kind()) == "wayland")
    check("H1.4 QT_QPA_PLATFORM=wayland（无 WAYLAND_DISPLAY）",
          with_env({"WAYLAND_DISPLAY": None, "DISPLAY": None, "XDG_SESSION_TYPE": None,
                    "QT_QPA_PLATFORM": "wayland"}, lambda: ds.session_kind()) == "wayland")
    check("H1.5 offscreen → unknown（即使有 DISPLAY）",
          with_env({"DISPLAY": ":0", "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": None,
                    "QT_QPA_PLATFORM": "offscreen"}, lambda: ds.session_kind()) == "unknown")
    check("H1.6 无显示无 WAYLAND → unknown",
          with_env({"DISPLAY": None, "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": None,
                    "QT_QPA_PLATFORM": None}, lambda: ds.session_kind()) == "unknown")
    check("H1.7 XDG_SESSION_TYPE=wayland（无 WAYLAND_DISPLAY/QT 平台）",
          with_env({"DISPLAY": None, "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": "wayland",
                    "QT_QPA_PLATFORM": None}, lambda: ds.session_kind()) == "wayland")
    check("H1.8 XDG_SESSION_TYPE=x11（无 DISPLAY）",
          with_env({"DISPLAY": None, "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": "x11",
                    "QT_QPA_PLATFORM": None}, lambda: ds.session_kind()) == "x11")
    check("H1.9 矛盾时 WAYLAND_DISPLAY 优先于 XDG_SESSION_TYPE=x11",
          with_env({"DISPLAY": ":0", "WAYLAND_DISPLAY": "wayland-0", "XDG_SESSION_TYPE": "x11",
                    "QT_QPA_PLATFORM": None}, lambda: ds.session_kind()) == "wayland")
finally:
    ds.sys = sys  # 还原

# ======================================================================
# H2. hotkey_plan 路由
# ======================================================================
try:
    _set_platform("win32")
    factory, msg = ds.hotkey_plan("win32")
    check("H2.1 win32 → GlobalHotkeyThread 工厂 + 说明",
          factory is gh.GlobalHotkeyThread and "Windows" in msg, repr((factory, msg)))
finally:
    ds.sys = sys

factory, msg = ds.hotkey_plan("x11")
check("H2.2 x11 → X11HotkeyThread 工厂 + 说明",
      factory is xh.X11HotkeyThread and "X11" in msg, repr((factory, msg)))

factory, msg = ds.hotkey_plan("wayland")
check("H2.3 wayland → None + 明确不可用说明（不绕过 compositor）",
      factory is None and "Wayland" in msg and "Alt+S" in msg, repr((factory, msg)))

factory, msg = ds.hotkey_plan("unknown")
check("H2.4 unknown → None + 明确不可用说明",
      factory is None and "不可用" in msg, repr((factory, msg)))

# ======================================================================
# H3–H4. X11HotkeyThread（fake libX11）
# ======================================================================

class FakeX11:
    def __init__(self, grab_rc=1, open_rc=0xAB, keycode=39, unsupported_vk=False):
        self.grab_rc = grab_rc
        self.open_rc = open_rc
        self.keycode = keycode
        self.unsupported_vk = unsupported_vk
        self.grab_calls = []
        self.ungrab_calls = []
        self.close_calls = 0
        self.select_calls = 0
        self.opened = False
        self.pending_left = 1
        # socketpair：跨平台可被 select() 监听（Windows 上 os.pipe 的 fd 不行）
        self._a, self._b = socket.socketpair()
        self._b.send(b"\x01")   # 让 a 有可读数据 → select 首次就绪
        self._event = xh.XEvent()
        self._event.type = xh.KeyPress
        self._event.keycode = self.keycode

    # -- libX11 API（鸭子类型替身） --
    def XOpenDisplay(self, name):
        self.opened = True
        return self.open_rc
    def XCloseDisplay(self, d):
        self.close_calls += 1
    def XConnectionNumber(self, d):
        return self._a.fileno()
    def XDefaultRootWindow(self, d):
        return 123
    def XKeysymToKeycode(self, d, keysym):
        if self.unsupported_vk:
            return 0
        return self.keycode
    def XSelectInput(self, d, w, mask):
        self.select_calls += 1
    def XGrabKey(self, d, kc, mod, w, owner):
        self.grab_calls.append((kc, mod, w, owner))
        return self.grab_rc
    def XUngrabKey(self, d, kc, mod, w):
        self.ungrab_calls.append((kc, mod, w))
    def XPending(self, d):
        if self.pending_left > 0:
            self.pending_left -= 1
            return 1
        return 0
    def XNextEvent(self, d, evp):
        # 生产代码传入 ctypes.byref(ev)（CArgObject）；真实 CDLL 自行解引用，
        # 替身可调用对象则通过 ._obj 拿回原始 struct
        import ctypes as _ct
        target = getattr(evp, "_obj", evp)
        _ct.memmove(_ct.byref(target), _ct.byref(self._event), _ct.sizeof(xh.XEvent))
        self._event.type = 0   # 之后无事件

    def close(self):
        for s in (self._a, self._b):
            try:
                s.close()
            except Exception:
                pass


def run_hotkey_case(fake, **kw):
    """启动 X11HotkeyThread，等待 triggered 或退出，回收。返回 (got_signal, thread)"""
    orig_open = xh._open_x11
    xh._open_x11 = lambda: (fake, fake.open_rc)
    try:
        t = xh.X11HotkeyThread(**kw)
        got = threading.Event()
        t.triggered.connect(lambda: got.set())
        t.start()
        t.wait_ready(3.0)
        deadline = time.time() + 6.0
        while not got.is_set() and time.time() < deadline:
            APP.processEvents()
            time.sleep(0.02)
        t.stop()
        t.wait(3000)
        return got.is_set(), t
    finally:
        xh._open_x11 = orig_open
        fake.close()


# H3.1 成功路径：grab 参数正确 + 命中发射 + stop 释放
fake = FakeX11()
got, t = run_hotkey_case(fake)
check("H3.1 命中 Alt+S → triggered 发射", got)
check("H3.2 XGrabKey 参数（keycode=39, Mod1Mask=1, root=123, owner_events=1）",
      fake.grab_calls == [(39, xh.Mod1Mask, 123, 1)], repr(fake.grab_calls))
check("H3.3 stop 后 XUngrabKey + XCloseDisplay 释放",
      len(fake.ungrab_calls) == 1 and fake.close_calls == 1,
      repr((fake.ungrab_calls, fake.close_calls)))
check("H3.4 注册成功标志 _registered", t._registered is False)  # cleanup 后复位为 False

# H4.1 键被占用（XGrabKey → 0）
fake = FakeX11(grab_rc=0)
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
    got, t = run_hotkey_case(fake)
check("H4.1 注册失败 → 无信号、_registered=False、明确日志",
      not got and t._registered is False and ("占用" in buf.getvalue() or "失败" in buf.getvalue()),
      buf.getvalue()[-200:])

# H4.2 无显示（XOpenDisplay → None）
fake = FakeX11(open_rc=None)
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
    got, t = run_hotkey_case(fake)
check("H4.2 无 X11 显示 → 安静退出 + 明确日志",
      not got and t._registered is False and "XOpenDisplay" in buf.getvalue(),
      buf.getvalue()[-200:])

# H4.3 不支持的组合（vk 不在窄映射表）
fake = FakeX11()
orig_open = xh._open_x11
t_probe = xh.X11HotkeyThread(vk=0x41)
xh._open_x11 = lambda: (fake, 0xAB)
try:
    t_probe.start()
    t_probe.wait_ready(3.0)
    deadline = time.time() + 5.0
    while t_probe.isRunning() and time.time() < deadline:
        time.sleep(0.02)
    t_probe.stop()
    t_probe.wait(2000)
finally:
    xh._open_x11 = orig_open
    fake.close()
check("H4.3 不支持的组合 → 不打开显示即退出",
      t_probe._registered is False and not fake.opened, repr(t_probe._registered))

# ======================================================================
# H5. Windows 路径保持
# ======================================================================
check("H5.1 GlobalHotkeyThread 常量完整（MOD_ALT/VK_S/WM_HOTKEY）",
      gh.MOD_ALT == 0x0001 and gh.VK_S == 0x53 and gh.WM_HOTKEY == 0x0312)

t_win = gh.GlobalHotkeyThread()
t_win.start()
t_win._ready.wait(3.0)
if gh._is_windows:
    # 本机 Windows：真实 RegisterHotKey + GetMessage 循环 = 行为保持的存活检查（随后立即释放 Alt+S）
    check("H5.2 Windows：RegisterHotKey 线程运行中（行为保持）", t_win.isRunning())
    t_win.stop()
    t_win.wait(3000)
    check("H5.3 Windows：stop() 干净退出（释放热键）", not t_win.isRunning())
else:
    t_win.wait(3000)
    check("H5.2 非 Windows：run() 静默就绪退出（不注册、不崩）",
          not t_win.isRunning() and t_win._ready.is_set(),
          f"running={t_win.isRunning()}")

print(f"\n===== {sum(1 for _, ok in RESULTS if ok)}/{len(RESULTS)} passed =====", flush=True)
sys.exit(0 if all(ok for _, ok in RESULTS) else 1)