222 lines
9.0 KiB
Python
222 lines
9.0 KiB
Python
#!/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()
|