280 lines
11 KiB
Python
280 lines
11 KiB
Python
# -*- 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)
|