#!/usr/bin/env python3 """Clipboard bursts and application chooser on a private Xvfb/AT-SPI session.""" import json import os from pathlib import Path import queue import subprocess import sys import threading sys.dont_write_bytecode = True from native_x11 import wait_for def owner(): import gi gi.require_version("Gtk", "3.0") gi.require_version("Gdk", "3.0") gi.require_version("GdkPixbuf", "2.0") from gi.repository import Gdk, GdkPixbuf, GLib, Gtk GLib.set_prgname("example-clipboard-app") GLib.set_application_name("Example Clipboard Application") window = Gtk.Window(title="Example Clipboard Owner") window.add(Gtk.Entry()) window.show_all() clipboard = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD) print("ready", flush=True) def command(source, condition): line = source.readline() if not line: Gtk.main_quit() return False request = json.loads(line) if "text" in request: clipboard.set_text(request["text"], -1) elif request.get("image"): image = GdkPixbuf.Pixbuf.new(GdkPixbuf.Colorspace.RGB, False, 8, 2, 2) image.fill(0) clipboard.set_image(image) elif "burst" in request: texts = iter(request["burst"]) def next_text(): text = next(texts, None) if text is None: print("ready", flush=True) return False clipboard.set_text(text, -1) return True GLib.timeout_add(20, next_text) return True print("ready", flush=True) return True GLib.io_add_watch(sys.stdin, GLib.IO_IN | GLib.IO_HUP, command) Gtk.main() def inside(): import pyatspi binary = os.environ["CTHULHU_BINARY"] children = [] def launch(arguments): process = subprocess.Popen(arguments, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True) children.append(process) return process def send(process, value): process.stdin.write(json.dumps(value) + "\n") process.stdin.flush() def walk(node): yield node for child in node: if child is not None: yield from walk(child) def find_window(name): for application in pyatspi.Registry.getDesktop(0): for child in application: if child is not None and child.name == name: return child return None def key(*keys): subprocess.run(["xdotool", "key", "--clearmodifiers", *keys], check=True) def focus_window(title): xid = subprocess.check_output(["xdotool", "search", "--onlyvisible", "--name", "^" + title + "$"], text=True).splitlines()[0] subprocess.run(["xdotool", "windowfocus", xid], check=True) try: fixture = launch([sys.executable, __file__, "--owner"]) assert fixture.stdout.readline().strip() == "ready" wait_for(lambda: find_window("Example Clipboard Owner"), "clipboard fixture missing") watcher = launch([binary, "--clipboard-watch"]) replies = queue.Queue() def receive(): for line in watcher.stdout: replies.put(json.loads(line)) threading.Thread(target=receive, daemon=True).start() assert replies.get(timeout=5)["Ready"] def configure(enabled, epoch): send(watcher, {"Enabled": enabled, "Epoch": epoch}) response = replies.get(timeout=5) assert response["Ready"] and response["Epoch"] == epoch, response configure(False, 1) send(fixture, {"text": "Disabled text"}) assert fixture.stdout.readline().strip() == "ready" try: assert False, ("disabled clipboard was read", replies.get(timeout=0.2)) except queue.Empty: pass configure(True, 2) texts = ["First clipboard", "Second clipboard", "Third clipboard", "Fourth clipboard"] send(fixture, {"burst": texts}) assert fixture.stdout.readline().strip() == "ready" captured = [replies.get(timeout=5) for _ in texts] assert [value["Text"] for value in captured] == texts, captured assert all(value["Epoch"] == 2 for value in captured), captured send(fixture, {"image": True}) assert fixture.stdout.readline().strip() == "ready" assert replies.get(timeout=5)["Text"] == "", "image data was read" uriOwner = launch(["xclip", "-quiet", "-selection", "clipboard", "-target", "text/uri-list", "-in"]) uriOwner.stdin.write("file:///tmp/example.txt\n") uriOwner.stdin.close() assert replies.get(timeout=5)["Text"] == "", "file URI was read" configure(False, 3) send(fixture, {"text": "Still disabled"}) assert fixture.stdout.readline().strip() == "ready" try: assert False, ("disabled clipboard was read", replies.get(timeout=0.2)) except queue.Empty: pass watcher.stdin.close() assert watcher.wait(timeout=5) == 0 print("PASS: clipboard disabled, four changes 20ms apart in order, image/URI ignored, disable, shutdown", flush=True) wine = launch([sys.executable, "-u", str(Path(__file__).with_name("window_title_fixture.py"))]) json.loads(wine.stdout.readline()) def chooser(): process = launch([binary, "--preferences-ui"]) send(process, {"ChooseApplication": True}) window = wait_for(lambda: find_window("Choose Application Preferences"), "application chooser missing") tableNode = next(node for node in walk(window) if node.name == "Applications" and node.getRole() == pyatspi.ROLE_TABLE) table = tableNode.queryTable() names = [table.getAccessibleAt(row, 0).name for row in range(table.nRows)] assert "game.exe" in names, names assert "example-clipboard-app" in names, names assert names.count("game.exe") == 1, names assert not any(name.lower().startswith("cthulhu") for name in names), names focus_window("Choose Application Preferences") assert tableNode.queryComponent().grabFocus() return process, tableNode, table, names process, tableNode, table, names = chooser() key("Tab") wait_for(lambda: not any(node.getState().contains(pyatspi.STATE_FOCUSED) for node in walk(tableNode)), "Tab trapped in application list") key("shift+Tab") wait_for(lambda: any(node.getState().contains(pyatspi.STATE_FOCUSED) for node in walk(tableNode)), "Shift+Tab did not return to application list") key("Escape") assert process.wait(timeout=5) == 0 assert process.stdout.read() == "", "Escape selected an application" process, tableNode, table, names = chooser() row = names.index("game.exe") assert table.getAccessibleAt(row, 0).queryComponent().grabFocus() key("Return") selected = json.loads(process.stdout.readline()) assert selected["Application"] == "game.exe", selected send(process, {"Application": "game.exe", "CanSave": True, "Accepted": True, "Values": {"KeyboardLayout": 1, "CthulhuModifierKeys": ["KP_Insert"], "TimeFormat": "24-hour", "DateFormat": "numbers-mdy", "SpeechHistoryLimit": 50, "Punctuation": 1, "Capitalization": "none"}}) wait_for(lambda: find_window("Cthulhu Preferences for game.exe"), "chosen application preferences missing") focus_window("Cthulhu Preferences for game.exe") key("Escape") assert process.wait(timeout=5) == 0 assert process.stdout.read() == "", "Escape submitted preferences" print("PASS: AT-SPI/X11/Wine application identities, deduplication, Tab/Shift+Tab, Escape cancels, Enter opens selected preferences", flush=True) finally: for process in children: if process.poll() is None: process.terminate() process.wait(timeout=5) def main(): if "--owner" in sys.argv: owner() return if "--inside" in sys.argv: inside() return readFd, writeFd = os.pipe() server = subprocess.Popen(["Xvfb", "-displayfd", str(writeFd), "-screen", "0", "1024x768x24", "-nolisten", "tcp"], pass_fds=(writeFd,), stderr=subprocess.DEVNULL) os.close(writeFd) try: with os.fdopen(readFd) as reader: display = reader.readline().strip() assert display, "Xvfb failed" env = dict(os.environ, DISPLAY=":" + display, GDK_BACKEND="x11", GTK_MODULES="gail:atk-bridge", GIO_USE_VFS="local") for name in ("DBUS_SESSION_BUS_ADDRESS", "AT_SPI_BUS_ADDRESS", "WAYLAND_DISPLAY"): env.pop(name, None) subprocess.run(["dbus-run-session", "--", sys.executable, __file__, "--inside"], env=env, check=True, timeout=75) finally: server.terminate() server.wait(timeout=5) if __name__ == "__main__": main()