sorrow404null
0a62877cde
Windows keeps the legacy path; Linux X11 uses a native XGrabKey hotkey (ctypes libX11) with native capture; Wayland captures through xdg-desktop-portal and degrades gracefully with explicit capability-unavailable logs when the portal/protocol is missing.
294 lines
11 KiB
Python
294 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
P1-04 平台截图能力单测（portal 用替身 subprocess，不依赖真实 portal / compositor）
|
||
|
||
覆盖：
|
||
C1 capture_plan 路由（win32/x11 → overlay；wayland → portal；unknown → unavailable+明确日志）
|
||
C2 detect_portal（非 Linux / 无 D-Bus 会话 / 无 gdbus / 齐备 → 四分支）
|
||
C3 portal_screenshot_sync 成功路径（fake subprocess：Screenshot 返回解析 + FilePicked
|
||
file:// URI 解码（含空格文件名）→ 现有附件流程可用的真实文件路径；校验 gdbus 命令行）
|
||
C4 用户取消/授权被拒（只有 Finished 无 FilePicked → denied）
|
||
C5 portal 缺失（call rc≠0 NotSupported → "portal 不可用" + 原因）
|
||
C6 等待超时（显式预算内无信号 → timeout，不无限挂起）
|
||
C7 PortalScreenshotWorker 信号回主线程（done(ok, path)）
|
||
C8 ScreenCaptureOverlay offscreen 构造 + start() 空画面守卫不崩
|
||
|
||
运行: PYTHONIOENCODING=utf-8 python tests/test_screen_capture_platforms.py
|
||
"""
|
||
import contextlib
|
||
import io
|
||
import json
|
||
import os
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import types
|
||
import urllib.parse
|
||
|
||
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 portal_capture as pc # noqa: E402
|
||
from ui.views.system_tools import screen_capture as sc # 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)
|
||
|
||
|
||
# ======================================================================
|
||
# C1. capture_plan 路由
|
||
# ======================================================================
|
||
mode, msg = ds.capture_plan("win32")
|
||
check("C1.1 win32 → overlay", mode == "overlay" and msg is None, repr((mode, msg)))
|
||
mode, msg = ds.capture_plan("x11")
|
||
check("C1.2 x11 → overlay（原生 grabWindow 路径）", mode == "overlay" and msg is None,
|
||
repr((mode, msg)))
|
||
mode, msg = ds.capture_plan("wayland")
|
||
check("C1.3 wayland → portal（不绕过 compositor）", mode == "portal" and msg is None,
|
||
repr((mode, msg)))
|
||
mode, msg = ds.capture_plan("unknown")
|
||
check("C1.4 unknown → unavailable + 明确日志（主程序其余功能不受影响）",
|
||
mode == "unavailable" and msg and "不可用" in msg, repr((mode, msg)))
|
||
|
||
# ======================================================================
|
||
# C2. detect_portal 四分支
|
||
# ======================================================================
|
||
_orig_sys = pc.sys
|
||
_orig_which = pc.shutil.which
|
||
|
||
try:
|
||
pc.sys = types.SimpleNamespace(platform="win32")
|
||
ok, reason = pc.detect_portal()
|
||
check("C2.1 非 Linux → 不可用", ok is False and reason, repr((ok, reason)))
|
||
|
||
pc.sys = types.SimpleNamespace(platform="linux")
|
||
saved = {k: os.environ.get(k) for k in ("XDG_RUNTIME_DIR", "DBUS_SESSION_BUS_ADDRESS")}
|
||
for k in saved:
|
||
os.environ.pop(k, None)
|
||
ok, reason = pc.detect_portal()
|
||
check("C2.2 无 D-Bus 会话 → 不可用（明确原因）",
|
||
ok is False and "D-Bus" in reason, repr((ok, reason)))
|
||
os.environ["XDG_RUNTIME_DIR"] = "/run/user/1000"
|
||
pc.shutil.which = lambda name: None
|
||
ok, reason = pc.detect_portal()
|
||
check("C2.3 无 gdbus CLI → 不可用（明确原因）",
|
||
ok is False and "gdbus" in reason, repr((ok, reason)))
|
||
pc.shutil.which = lambda name: "/usr/bin/gdbus"
|
||
ok, reason = pc.detect_portal()
|
||
check("C2.4 齐备 → 可用", ok is True and reason is None, repr((ok, reason)))
|
||
finally:
|
||
pc.sys = _orig_sys
|
||
pc.shutil.which = _orig_which
|
||
for k, v in saved.items():
|
||
if v is None:
|
||
os.environ.pop(k, None)
|
||
else:
|
||
os.environ[k] = v
|
||
|
||
|
||
# ======================================================================
|
||
# C3–C7. portal_screenshot_sync（fake subprocess，模拟 Linux 环境）
|
||
# ======================================================================
|
||
_linux_sys = types.SimpleNamespace(platform="linux")
|
||
|
||
|
||
class FakeRun:
|
||
"""替身 subprocess.run：记录 argv，返回预先设定的 call 输出"""
|
||
def __init__(self, rc=0, stdout="", stderr=""):
|
||
self.rc = rc
|
||
self.stdout = stdout
|
||
self.stderr = stderr
|
||
self.cmds = []
|
||
|
||
def __call__(self, cmd, **kw):
|
||
self.cmds.append(list(cmd))
|
||
p = types.SimpleNamespace()
|
||
p.returncode = self.rc
|
||
p.stdout = self.stdout
|
||
p.stderr = self.stderr
|
||
return p
|
||
|
||
|
||
class FakePipe:
|
||
"""替身 monitor 的 stdout：按序输出 JSON 行；readline 可模拟阻塞"""
|
||
def __init__(self, lines, delay_s=0.0):
|
||
self._lines = list(lines)
|
||
self._delay = delay_s
|
||
self.closed = False
|
||
|
||
def readline(self):
|
||
if self._delay:
|
||
time.sleep(self._delay)
|
||
self._delay = 0
|
||
if self._lines:
|
||
return self._lines.pop(0) + "\n"
|
||
# 模拟管道关闭（EOF）
|
||
return ""
|
||
|
||
|
||
class FakePopen:
|
||
def __init__(self, lines, delay_s=0.0):
|
||
self._lines = lines
|
||
self._delay = delay_s
|
||
self.cmd = None
|
||
self.killed = False
|
||
|
||
def __call__(self, cmd, **kw):
|
||
self.cmd = list(cmd)
|
||
p = types.SimpleNamespace()
|
||
p.stdout = FakePipe(self._lines, self._delay)
|
||
p.stderr = io.StringIO()
|
||
self._rc = None
|
||
p.poll = lambda: self._rc
|
||
p.kill = lambda: setattr(p, "killed", True) or setattr(self, "killed", True)
|
||
p.wait = lambda timeout=None: 0
|
||
return p
|
||
|
||
|
||
def _uri(path: str) -> str:
|
||
return "file://" + urllib.parse.quote(path)
|
||
|
||
|
||
def _json(member, iface, body):
|
||
return json.dumps({"interface": iface, "member": member, "body": body},
|
||
ensure_ascii=False)
|
||
|
||
|
||
# C3.1 成功路径（含空格文件名 → URI 解码）
|
||
tmpdir = tempfile.mkdtemp(prefix="haocode_p104_portal_")
|
||
shot = os.path.join(tmpdir, "shot test.png")
|
||
with open(shot, "wb") as f:
|
||
f.write(b"\x89PNG\r\n\x1a\nfake")
|
||
|
||
pc.sys = _linux_sys # C3–C7 模拟 Linux 宿主
|
||
_orig_which2 = pc.shutil.which
|
||
pc.shutil.which = lambda name: "/usr/bin/gdbus" # 本机无 gdbus，替身探测
|
||
_saved_xdg = os.environ.get("XDG_RUNTIME_DIR")
|
||
os.environ["XDG_RUNTIME_DIR"] = "/run/user/1000"
|
||
run_fake = FakeRun(rc=0, stdout="('/org/freedesktop/portal/desktop/request/1000/haocode/7', <>)")
|
||
popen_fake = FakePopen([
|
||
_json("FilePicked", "org.freedesktop.portal.FileChooser", [_uri(shot), {}]),
|
||
_json("Finished", "org.freedesktop.portal.Request", [0]),
|
||
])
|
||
_orig_run, _orig_popen = pc.subprocess.run, pc.subprocess.Popen
|
||
pc.subprocess.run = run_fake
|
||
pc.subprocess.Popen = popen_fake
|
||
try:
|
||
t0 = time.time()
|
||
ok, result = pc.portal_screenshot_sync("haocode-shot", request_timeout_s=5, wait_budget_s=10)
|
||
dt = time.time() - t0
|
||
finally:
|
||
pc.subprocess.run = _orig_run
|
||
pc.subprocess.Popen = _orig_popen
|
||
|
||
check("C3.1 成功：返回真实文件路径（URI 空格解码正确）",
|
||
ok is True and result == shot and os.path.exists(result), repr((ok, result)))
|
||
check("C3.2 Screenshot gdbus 命令行正确（dest/path/method/parent=/）",
|
||
run_fake.cmds and run_fake.cmds[0][:5] == ["gdbus", "call", "--session", "--dest",
|
||
pc.PORTAL_DEST]
|
||
and "--method=org.freedesktop.portal.Screenshot.Screenshot" in run_fake.cmds[0]
|
||
and "/" in run_fake.cmds[0], repr(run_fake.cmds))
|
||
check("C3.3 monitor 监听 request 对象路径",
|
||
popen_fake.cmd and popen_fake.cmd[-2:] == ["--object-path",
|
||
"/org/freedesktop/portal/desktop/request/1000/haocode/7"],
|
||
repr(popen_fake.cmd))
|
||
check("C3.4 全程在显式预算内完成（<10s）", dt < 10, f"{dt:.1f}s")
|
||
|
||
# C4. 用户取消/授权被拒（只有 Finished）
|
||
popen_fake = FakePopen([_json("Finished", "org.freedesktop.portal.Request", [0])])
|
||
pc.subprocess.run = run_fake
|
||
pc.subprocess.Popen = popen_fake
|
||
try:
|
||
ok, result = pc.portal_screenshot_sync("haocode-shot", 5, 10)
|
||
finally:
|
||
pc.subprocess.run = _orig_run
|
||
pc.subprocess.Popen = _orig_popen
|
||
check("C4.1 授权被拒 → 不伪造成功，明确原因",
|
||
ok is False and "取消" in result, repr((ok, result)))
|
||
|
||
# C5. portal 缺失/方法不支持（call rc≠0）
|
||
run_fake2 = FakeRun(rc=1, stderr="Error: org.freedesktop.DBus.Error.NotSupported: "
|
||
"Method not supported by portal")
|
||
pc.subprocess.run = run_fake2
|
||
pc.subprocess.Popen = FakePopen([])
|
||
try:
|
||
ok, result = pc.portal_screenshot_sync("haocode-shot", 5, 10)
|
||
finally:
|
||
pc.subprocess.run = _orig_run
|
||
pc.subprocess.Popen = _orig_popen
|
||
check("C5.1 portal 不支持 → 不可用 + 原因透出",
|
||
ok is False and "portal 不可用" in result and "NotSupported" in result, repr((ok, result)))
|
||
|
||
# C6. 等待超时（monitor 首行前阻塞 > 预算）
|
||
popen_fake = FakePopen([_json("Finished", "org.freedesktop.portal.Request", [0])],
|
||
delay_s=3.0)
|
||
pc.subprocess.run = run_fake
|
||
pc.subprocess.Popen = popen_fake
|
||
try:
|
||
t0 = time.time()
|
||
ok, result = pc.portal_screenshot_sync("haocode-shot", 5, 1.0)
|
||
dt = time.time() - t0
|
||
finally:
|
||
pc.subprocess.run = _orig_run
|
||
pc.subprocess.Popen = _orig_popen
|
||
check("C6.1 超预算 → timeout（不无限挂起）",
|
||
ok is False and "超时" in result and dt < 5, repr((ok, result, f"{dt:.1f}s")))
|
||
|
||
# ======================================================================
|
||
# C7. PortalScreenshotWorker 信号
|
||
# ======================================================================
|
||
popen_fake = FakePopen([
|
||
_json("FilePicked", "org.freedesktop.portal.FileChooser", [_uri(shot), {}]),
|
||
_json("Finished", "org.freedesktop.portal.Request", [0]),
|
||
])
|
||
pc.subprocess.run = run_fake
|
||
pc.subprocess.Popen = popen_fake
|
||
try:
|
||
w = pc.PortalScreenshotWorker(wait_budget_s=10)
|
||
done = {}
|
||
w.done.connect(lambda ok, r: done.update(ok=ok, r=r))
|
||
w.start()
|
||
deadline = time.time() + 15
|
||
while "ok" not in done and time.time() < deadline:
|
||
APP.processEvents()
|
||
time.sleep(0.02)
|
||
w.wait(3000)
|
||
finally:
|
||
pc.subprocess.run = _orig_run
|
||
pc.subprocess.Popen = _orig_popen
|
||
pc.shutil.which = _orig_which2
|
||
if _saved_xdg is None:
|
||
os.environ.pop("XDG_RUNTIME_DIR", None)
|
||
else:
|
||
os.environ["XDG_RUNTIME_DIR"] = _saved_xdg
|
||
pc.sys = _orig_sys
|
||
check("C7.1 worker done 信号回主线程（ok + 路径）",
|
||
done.get("ok") is True and done.get("r") == shot, repr(done))
|
||
|
||
# ======================================================================
|
||
# C8. 覆盖层 offscreen 构造 + start() 空画面守卫
|
||
# ======================================================================
|
||
ov = sc.ScreenCaptureOverlay()
|
||
buf = io.StringIO()
|
||
with contextlib.redirect_stdout(buf):
|
||
ov.start() # offscreen：无屏幕或空画面都不得崩
|
||
shown = ov.isVisible()
|
||
check("C8.1 offscreen start() 不崩；空画面时不显示覆盖层且日志明确",
|
||
(not shown and "空画面" in buf.getvalue()) or (shown and ov._full_pixmap is not None),
|
||
f"shown={shown} log={buf.getvalue()[-160:]!r}")
|
||
if shown:
|
||
ov.close()
|
||
|
||
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)
|