Cthulhu is no longer a fork. Now it is a screen reader written in Golang.

tests/native_clipboard_preferences.py

@@ -0,0 +1,221 @@
+#!/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()