feat(factory): 内容工厂与链接处理
- 第三栏内容工厂：按拖入类型展示动作，动作按钮本身可接收 drop - 链接分类：http(s) 按扩展名判定图片/音视频/文档/压缩包/网页 - http 链接动作：打开、下载、另存为、复制、Markdown、复制文件名、待办、问小埃 - 非标准 scheme：尝试打开、复制；thunder 解析 base64 真实地址，magnet 解析 dn - 文字动作：ICU 分词、正则提取链接、朗读（spd-say）、复制、待办 - 文件/图片动作：打开、提取文字（占位）、生成描述、保存图片、复制路径、待办 - Downloader：异步下载到 XDG 下载目录 / 另存为，面板 toast 提示 - TodoModel：持久化到 todo.json，供「添加到待办」使用 - 无头自测 gxde-ai-see-selftest：链接分类/分词/提取链接/动作匹配 - 合成器 portal 自测脚本 tests/portal（虚拟指针 + GTK 源/目标）
This commit is contained in:
@@ -0,0 +1,242 @@
|
||||
#!/usr/bin/env python3
|
||||
# SPDX-FileCopyrightText: 2026 GXDE OS Team
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""End-to-end test for the GXDE portal (native drop into a window).
|
||||
|
||||
Run inside a private session bus:
|
||||
|
||||
dbus-run-session -- python3 tests/portal/run.py [compositor]
|
||||
|
||||
It starts a nested GXDE compositor, a GTK drag source and a GTK drop target,
|
||||
drives the pointer through wlr-virtual-pointer, asks the compositor to open the
|
||||
portal overview and asserts that the drop lands in the target window.
|
||||
"""
|
||||
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
RUNTIME = f"/run/user/{os.getuid()}"
|
||||
PORTAL = ["org.gxde.Portal", "/org/gxde/Portal", "org.gxde.Portal"]
|
||||
|
||||
|
||||
def log(message):
|
||||
print(f"[test] {message}", flush=True)
|
||||
|
||||
|
||||
def wait_for(predicate, timeout=20.0, interval=0.1):
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
value = predicate()
|
||||
if value:
|
||||
return value
|
||||
time.sleep(interval)
|
||||
return None
|
||||
|
||||
|
||||
def portal_call(method, *args):
|
||||
address = os.environ.get("DBUS_SESSION_BUS_ADDRESS", "")
|
||||
cmd = ["busctl", f"--address={address}", "--json=short", "call", *PORTAL, method, *args]
|
||||
out = subprocess.check_output(cmd, text=True)
|
||||
return json.loads(out)
|
||||
|
||||
|
||||
def list_windows():
|
||||
data = portal_call("ListWindows")["data"]
|
||||
rows = data[0] if data else []
|
||||
return [
|
||||
{
|
||||
"uuid": row[0],
|
||||
"title": row[1],
|
||||
"app": row[2],
|
||||
"x": row[3],
|
||||
"y": row[4],
|
||||
"w": row[5],
|
||||
"h": row[6],
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
def layout():
|
||||
data = portal_call("GetLayout")["data"]
|
||||
rows = data[0] if data else []
|
||||
return [
|
||||
{
|
||||
"uuid": row[0],
|
||||
"title": row[1],
|
||||
"x": row[3],
|
||||
"y": row[4],
|
||||
"w": row[5],
|
||||
"h": row[6],
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
def find_by_title(windows, title):
|
||||
for window in windows:
|
||||
if window["title"] == title:
|
||||
return window
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
compositor = (
|
||||
sys.argv[1]
|
||||
if len(sys.argv) > 1
|
||||
else os.path.expanduser("~/playground/flake-wm/build-portal/flakewm")
|
||||
)
|
||||
parent_display = os.environ.get("WAYLAND_DISPLAY", "wayland-0")
|
||||
|
||||
before = set(glob.glob(f"{RUNTIME}/wayland-*"))
|
||||
env = os.environ.copy()
|
||||
env["WAYLAND_DISPLAY"] = parent_display
|
||||
|
||||
compositor_log = open("/tmp/pt-comp.log", "wb")
|
||||
compositor_proc = subprocess.Popen(
|
||||
[compositor], env=env, stdout=compositor_log, stderr=subprocess.STDOUT
|
||||
)
|
||||
|
||||
procs = [compositor_proc]
|
||||
pointer = None
|
||||
try:
|
||||
def nested_display():
|
||||
for path in set(glob.glob(f"{RUNTIME}/wayland-*")) - before:
|
||||
if not path.endswith(".lock"):
|
||||
return os.path.basename(path)
|
||||
return None
|
||||
|
||||
display = wait_for(nested_display, 25.0)
|
||||
assert display, "nested compositor did not create a Wayland socket"
|
||||
log(f"nested display: {display}")
|
||||
env["WAYLAND_DISPLAY"] = display
|
||||
|
||||
target_log = open("/tmp/pt-target.log", "wb")
|
||||
source_log = open("/tmp/pt-source.log", "wb")
|
||||
target = subprocess.Popen(
|
||||
[sys.executable, f"{HERE}/drop_target.py"],
|
||||
env=env, stdout=target_log, stderr=subprocess.STDOUT,
|
||||
)
|
||||
source = subprocess.Popen(
|
||||
[sys.executable, f"{HERE}/drag_source.py"],
|
||||
env=env, stdout=source_log, stderr=subprocess.STDOUT,
|
||||
)
|
||||
procs += [target, source]
|
||||
|
||||
def portal_ready():
|
||||
try:
|
||||
portal_call("ListWindows")
|
||||
return True
|
||||
except subprocess.CalledProcessError:
|
||||
return False
|
||||
|
||||
assert wait_for(portal_ready, 20.0, 0.3), "org.gxde.Portal never appeared"
|
||||
log("portal D-Bus ready")
|
||||
|
||||
windows = wait_for(
|
||||
lambda: list_windows()
|
||||
if find_by_title(list_windows(), "PortalDragSource")
|
||||
and find_by_title(list_windows(), "PortalDropTarget")
|
||||
else None,
|
||||
20.0,
|
||||
0.2,
|
||||
)
|
||||
assert windows, "windows did not appear"
|
||||
log(f"windows: {json.dumps(windows)}")
|
||||
|
||||
src = find_by_title(windows, "PortalDragSource")
|
||||
dst = find_by_title(windows, "PortalDropTarget")
|
||||
|
||||
pointer = subprocess.Popen(
|
||||
[f"{HERE}/pointer"], env=env, stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE, text=True, bufsize=1,
|
||||
)
|
||||
procs.append(pointer)
|
||||
|
||||
def pcmd(line):
|
||||
pointer.stdin.write(line + "\n")
|
||||
pointer.stdin.flush()
|
||||
response = pointer.stdout.readline().strip()
|
||||
if not (response.startswith("ok") or response.startswith("size")):
|
||||
raise RuntimeError(f"pointer failed: {response!r}")
|
||||
return response
|
||||
|
||||
log(f"pointer {pcmd('size')}")
|
||||
|
||||
sx = src["x"] + src["w"] / 2
|
||||
sy = src["y"] + src["h"] / 2
|
||||
pcmd(f"move {sx:.0f} {sy:.0f}")
|
||||
time.sleep(0.2)
|
||||
pcmd("press")
|
||||
time.sleep(0.2)
|
||||
pcmd(f"move {sx + 60:.0f} {sy:.0f}")
|
||||
time.sleep(0.8)
|
||||
|
||||
if "DRAG-BEGIN" not in open("/tmp/pt-source.log").read():
|
||||
raise RuntimeError("GTK drag did not start; see /tmp/pt-source.log")
|
||||
log("drag started")
|
||||
|
||||
started = portal_call("BeginPortal")["data"][0]
|
||||
assert started, "BeginPortal returned false"
|
||||
log("portal opened")
|
||||
|
||||
time.sleep(0.7) # let the overview animation settle
|
||||
tiles = wait_for(lambda: layout() or None, 5.0, 0.2)
|
||||
assert tiles, "portal layout is empty"
|
||||
log(f"layout: {json.dumps(tiles)}")
|
||||
|
||||
tile = find_by_title(tiles, "PortalDropTarget") or find_by_title(tiles, "PortalDragSource")
|
||||
assert tile, "target window not in the portal layout"
|
||||
|
||||
# If we only see the source tile, the drop still validates retargeting.
|
||||
tx = tile["x"] + tile["w"] / 2
|
||||
ty = tile["y"] + tile["h"] / 2
|
||||
pcmd(f"move {tx:.0f} {ty:.0f}")
|
||||
time.sleep(0.6)
|
||||
|
||||
subprocess.run(["grim", "/tmp/pt-before-drop.png"], env=env, check=False)
|
||||
pcmd("release")
|
||||
time.sleep(1.0)
|
||||
subprocess.run(["grim", "/tmp/pt-after-drop.png"], env=env, check=False)
|
||||
|
||||
target_text = open("/tmp/pt-target.log").read()
|
||||
log(f"target log: {target_text.strip()!r}")
|
||||
assert "DROP" in target_text, "the drop did not reach the target window"
|
||||
assert "portal-payload-42" in target_text, "wrong payload"
|
||||
|
||||
log("PASS: native drop landed in the target window")
|
||||
return 0
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log(f"FAIL: {exc}")
|
||||
for path in ("/tmp/pt-comp.log", "/tmp/pt-source.log", "/tmp/pt-target.log"):
|
||||
try:
|
||||
content = open(path).read().strip()
|
||||
except OSError:
|
||||
continue
|
||||
if content:
|
||||
log(f"--- {path} ---\n{content[-2000:]}")
|
||||
return 1
|
||||
finally:
|
||||
if pointer is not None:
|
||||
try:
|
||||
pointer.stdin.write("quit\n")
|
||||
pointer.stdin.flush()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
for proc in reversed(procs):
|
||||
if proc.poll() is None:
|
||||
proc.terminate()
|
||||
for proc in reversed(procs):
|
||||
try:
|
||||
proc.wait(timeout=5)
|
||||
except Exception: # noqa: BLE001
|
||||
proc.kill()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user