#!/usr/bin/env python3 # Copyright (c) 2026 Storm Dragon # Authors: Storm Dragon, Angus Kola # SPDX-License-Identifier: GPL-3.0-or-later """GTK/Qt keyboard fixture and isolated X11 foreground acceptance proof.""" import json import os from pathlib import Path import subprocess import sys import tempfile sys.dont_write_bytecode = True from native_x11 import wait_for def fixture(toolkit): title = "Cthulhu " + toolkit + " Controls" if toolkit == "GTK": import gi gi.require_version("Gtk", "3.0") gi.require_version("Gdk", "3.0") from gi.repository import Gdk, Gtk window = Gtk.Window(title=title) window.set_default_size(500, 650) window.connect("destroy", Gtk.main_quit) box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) window.add(box) def add(widget, name): widget.get_accessible().set_name(name) box.pack_start(widget, False, False, 0) return widget button = add(Gtk.Button(label="Open dialog"), "Open dialog") def dialog(*args): child = Gtk.Dialog(title="Ownership dialog", transient_for=window, modal=True) entry = Gtk.Entry() entry.get_accessible().set_name("Dialog entry") child.get_content_area().add(entry) child.add_button("Close", Gtk.ResponseType.CLOSE) child.show_all() child.run() child.destroy() button.connect("clicked", dialog) add(Gtk.CheckButton(label="Updates"), "Updates") radio = add(Gtk.RadioButton.new_with_label(None, "Automatic"), "Automatic") add(Gtk.RadioButton.new_with_label_from_widget(radio, "Manual"), "Manual") add(Gtk.Entry(), "Example text") combo = add(Gtk.ComboBoxText(), "Choice") for text in ("Alpha", "Beta", "Gamma"): combo.append_text(text) combo.set_active(0) add(Gtk.SpinButton.new_with_range(0, 100, 1), "Amount") add(Gtk.Scale.new_with_range(Gtk.Orientation.HORIZONTAL, 0, 100, 1), "Level") for tree, name in ((False, "Results"), (True, "Folders")): store = Gtk.TreeStore(str) if tree else Gtk.ListStore(str) if tree: parent = store.append(None, ["Parent folder"]) store.append(parent, ["Child folder"]) else: for text in ("First result", "Second result"): store.append([text]) view = add(Gtk.TreeView(model=store), name) view.append_column(Gtk.TreeViewColumn(name, Gtk.CellRendererText(), text=0)) view.set_size_request(-1, 80) disabled = add(Gtk.Button(label="Unavailable action"), "Unavailable action") disabled.set_sensitive(False) menu = Gtk.MenuBar() root = Gtk.MenuItem.new_with_mnemonic("_Actions") submenu = Gtk.Menu() submenu.append(Gtk.MenuItem(label="Example action")) unavailable = Gtk.MenuItem(label="Unavailable menu action") unavailable.set_sensitive(False) submenu.append(unavailable) root.set_submenu(submenu) menu.append(root) box.pack_start(menu, False, False, 0) siblings = [] def on_key(_window, event): if event.keyval == Gdk.KEY_F7: popup = Gtk.Menu() popup.append(Gtk.MenuItem(label="First link")) popup.append(Gtk.MenuItem(label="Second link")) popup.show_all() popup.popup_at_widget(button, Gdk.Gravity.SOUTH_WEST, Gdk.Gravity.NORTH_WEST, None) return True if event.keyval != Gdk.KEY_F6: return False sibling = Gtk.Window(title="Ownership sibling") sibling.set_default_size(350, 200) entry = Gtk.Entry() entry.get_accessible().set_name("Sibling entry") sibling.add(entry) sibling.show_all() sibling.present() siblings.append(sibling) return True window.connect("key-press-event", on_key) window.show_all() button.grab_focus() Gtk.main() else: from PySide6 import QtCore, QtWidgets app = QtWidgets.QApplication([]) window = QtWidgets.QWidget() window.setWindowTitle(title) layout = QtWidgets.QVBoxLayout(window) def add(widget, name): widget.setAccessibleName(name) layout.addWidget(widget) return widget button = add(QtWidgets.QPushButton("Open dialog"), "Open dialog") def dialog(): child = QtWidgets.QDialog(window) child.setWindowTitle("Ownership dialog") content = QtWidgets.QVBoxLayout(child) entry = QtWidgets.QLineEdit() entry.setAccessibleName("Dialog entry") content.addWidget(entry) close = QtWidgets.QPushButton("Close") close.clicked.connect(child.accept) content.addWidget(close) child.exec() button.clicked.connect(dialog) add(QtWidgets.QCheckBox("Updates"), "Updates") radio = add(QtWidgets.QRadioButton("Automatic"), "Automatic") radio.setChecked(True) add(QtWidgets.QRadioButton("Manual"), "Manual") add(QtWidgets.QLineEdit(), "Example text") combo = add(QtWidgets.QComboBox(), "Choice") combo.addItems(["Alpha", "Beta", "Gamma"]) add(QtWidgets.QSpinBox(), "Amount") add(QtWidgets.QSlider(QtCore.Qt.Horizontal), "Level") view = add(QtWidgets.QListWidget(), "Results") view.addItems(["First result", "Second result"]) tree = add(QtWidgets.QTreeWidget(), "Folders") tree.setHeaderLabels(["Folders"]) parent = QtWidgets.QTreeWidgetItem(tree, ["Parent folder"]) QtWidgets.QTreeWidgetItem(parent, ["Child folder"]) table = add(QtWidgets.QTableWidget(2, 2), "Table") for row in range(2): for col in range(2): table.setItem(row, col, QtWidgets.QTableWidgetItem(f"Cell {row+1} {col+1}")) disabled = add(QtWidgets.QPushButton("Unavailable action"), "Unavailable action") disabled.setEnabled(False) menu = QtWidgets.QMenuBar() root = menu.addMenu("&Actions") root.addAction("Example action") root.addAction("Unavailable menu action").setEnabled(False) layout.addWidget(menu) window.resize(500, 650) window.show() button.setFocus() app.exec() def inside(directory): root = Path(directory) processes = [] log = root / "cthulhu.log" def run(*args): return subprocess.check_output(args, text=True).strip() def records(): if not log.exists(): return [] return [json.loads(line) for line in log.read_text().splitlines() if line.endswith("}")] def wait_record(message, start=0, **fields): return wait_for(lambda: next((r for r in records()[start:] if r.get("message") == message and all(r.get(k) == v for k,v in fields.items())), None), message + " " + str(fields)) def key(*keys): run("xdotool", "key", "--clearmodifiers", *keys) def activate(xid): run("xdotool", "windowactivate", "--sync", xid) def find(title): def search(): p = subprocess.run(["xdotool", "search", "--onlyvisible", "--name", "^"+title+"$"], capture_output=True, text=True) return p.stdout.strip().splitlines()[0] if p.stdout.strip() else None return wait_for(search, "missing " + title) try: wm_config = root / "i3.conf" wm_config.write_text('font pango:monospace 10\nfocus_follows_mouse no\n') processes.append(subprocess.Popen(["i3", "-c", str(wm_config)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)) wait_for(lambda: subprocess.run(["i3-msg", "-t", "get_workspaces"], capture_output=True).returncode == 0, "i3 startup") settings = root / "settings.toml" settings.write_text('[general]\nenableSpeech = false\nenableSound = false\nkeyboardLayout = 1\n') binary = os.environ["CTHULHU_BINARY"] reader = subprocess.Popen([binary, "--config", str(settings), "--debug-file", str(log)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) processes.append(reader) wait_record("listening for focus events") windows = [] for toolkit in ("GTK", "Qt"): initial = len(records()) processes.append(subprocess.Popen([sys.executable, __file__, "--fixture", toolkit])) xid = find("Cthulhu " + toolkit + " Controls") windows.append(xid) activate(xid) wait_for(lambda: any(r.get("message") == "queued accessible presentation" and r.get("role") in ("button", "push button") for r in records()[initial:]), "initial button presentation") start = len(records()) key("Tab", "space") wait_record("queued accessible presentation", start, role="check box") wait_record("queued accessible presentation", start, type="object:state-changed:checked") # Exercise normal traversal without choosing controls through AT-SPI. key(*(["Tab"]*12)) wait_record("queued accessible presentation", start, role="combo box") wait_record("queued accessible presentation", start, role="spin button") wait_record("queued accessible presentation", start, role="radio button") wait_record("queued accessible presentation", start, role="text") wait_record("queued accessible presentation", start, role="slider") wait_record("queued accessible presentation", start, role="table cell") key(*(["shift+Tab"]*12)) start = len(records()) key("alt+a", "Down") wait_for(lambda: any(r.get("message") == "queued accessible presentation" and r.get("role") in ("menu", "menu item") for r in records()[start:]), "menu presentation") key("Escape", "Escape") if toolkit == "GTK": start = len(records()) key("F7", "Down") wait_record("queued accessible presentation", start, role="menu item") key("Escape") print("PASS:", toolkit, "forward/reverse controls, checked state, and menus") for xid in windows + windows[::-1]: start = len(records()) activate(xid) key("Tab") wait_record("queued accessible presentation", start) # A distinct window in the same process must get its own owner. activate(windows[0]) start = len(records()) key("F6") sibling = find("Ownership sibling") wait_record("queued accessible presentation", start, role="text") owned = wait_record("activated accessible window", start) expected_window = owned["window_path"] for xid in (windows[0], sibling, windows[0]): start = len(records()) activate(xid) key("KP_Enter") wait_record("queued where am I presentation", start) # Restart with overlapping same-process windows to force recovery to # distinguish the active sibling from an earlier geometry match. for xid in (windows[0], sibling): run("i3-msg", f"[id={xid}] floating enable, border pixel 0, resize set 600 700, move position 100 100") def geometry(xid): values = dict(line.split("=", 1) for line in run("xdotool", "getwindowgeometry", "--shell", xid).splitlines()) return tuple(values[k] for k in ("X", "Y", "WIDTH", "HEIGHT")) wait_for(lambda: geometry(windows[0]) == geometry(sibling), "identical sibling geometry") activate(windows[0]) activate(sibling) reader.terminate() assert reader.wait(timeout=5) == 0, "reader restart shutdown failed" start = len(records()) reader = subprocess.Popen([binary, "--config", str(settings), "--debug-file", str(log)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) processes.append(reader) recovered = wait_record("activated accessible window", start) assert recovered["window_path"] == expected_window, ("recovered inactive same-geometry sibling", recovered, owned) key("Tab") wait_record("queued accessible presentation", start, role="text") print("PASS: recovery selects active same-process, same-geometry sibling") run("i3-msg", f"[id={sibling}] kill") # Actual i3 empty workspace, retaining live background applications. start = len(records()) run("i3-msg", "workspace 2") wait_record("X11 foreground ownership changed", start, x_window=0) key("KP_Enter") wait_record("queued where am I presentation", start, role="system") start = len(records()) run("i3-msg", "workspace 1") wait_record("queued accessible presentation", start) # Activate dialog from a predictable keyboard position in a fresh window. processes.append(subprocess.Popen([sys.executable, __file__, "--fixture", "GTK"])) new_window = wait_for(lambda: next((w for w in run("xdotool", "search", "--name", "^Cthulhu GTK Controls$").splitlines() if w not in windows), None), "second same-toolkit window") activate(new_window) start = len(records()) key("space") dialog = find("Ownership dialog") wait_record("queued accessible presentation", start, role="text") start = len(records()) key("Escape") wait_record("queued accessible presentation", start) key("KP_Down", "KP_Right", "KP_Left", "Tab") start = len(records()) run("i3-msg", f"[id={new_window}] kill") wait_record("X11 foreground ownership changed", start) # Brave publishes multiple top-levels on one application bus. Exercise # the reported browser workflow with both windows at the same bounds. import shutil browser = shutil.which("brave") or shutil.which("brave-browser") or shutil.which("chromium") assert browser, "Chromium browser required for ownership regression" browser_args = [browser, "--user-data-dir=" + str(root / "browser-profile"), "--no-first-run", "--no-default-browser-check", "--disable-background-networking", "--force-renderer-accessibility", "--ozone-platform=x11", "--new-window"] start = len(records()) processes.append(subprocess.Popen(browser_args + [Path(__file__).with_name("browser.html").resolve().as_uri()], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)) first_browser = find("Cthulhu Browser Probe.*") activate(first_browser) wait_record("activated accessible window", start) start = len(records()) subprocess.run(browser_args + [Path(__file__).with_name("browser-next.html").resolve().as_uri()], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) second_browser = find("Cthulhu Second Page.*") activate(second_browser) expected_browser = wait_record("activated accessible window", start)["window_path"] for xid in (first_browser, second_browser): run("i3-msg", f"[id={xid}] floating enable, border pixel 0, resize set 600 700, move position 100 100") wait_for(lambda: geometry(first_browser) == geometry(second_browser), "identical browser geometry") activate(second_browser) reader.terminate() assert reader.wait(timeout=5) == 0, "browser recovery restart shutdown failed" start = len(records()) reader = subprocess.Popen([binary, "--config", str(settings), "--debug-file", str(log)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) processes.append(reader) recovered = wait_record("activated accessible window", start) assert recovered["window_path"] == expected_browser, ("recovered inactive browser", recovered, expected_browser) start = len(records()) key("ctrl+l") wait_record("queued accessible presentation", start, role="entry") print("PASS: Brave same-geometry window recovery and address focus without Alt+Tab") key("Insert+q") assert reader.wait(timeout=5) == 0, "reader shutdown failed" print("PASS: window switching, empty workspace/recovery, dialog/destruction, review-key churn, shutdown") except Exception: import pyatspi for application in pyatspi.Registry.getDesktop(0): def describe(node, depth=0): if depth > 4: return try: print("AX", depth, node.getRoleName(), repr(node.name), "pid", node.get_process_id(), "bounds", node.queryComponent().getExtents(pyatspi.DESKTOP_COORDS), file=sys.stderr) except Exception as error: print("AX error", error, file=sys.stderr) for child in node: if child: describe(child, depth+1) describe(application) print(log.read_text() if log.exists() else "No runtime log", file=sys.stderr) raise finally: for process in reversed(processes): if process.poll() is None: process.terminate() try: process.wait(timeout=3) except subprocess.TimeoutExpired: process.kill() process.wait() def main(): if len(sys.argv) == 3 and sys.argv[1] == "--fixture": fixture(sys.argv[2]); return if len(sys.argv) == 3 and sys.argv[1] == "--inside": inside(sys.argv[2]); return with tempfile.TemporaryDirectory(prefix="cthulhu-ownership-") as directory: read_fd, write_fd = os.pipe() server = subprocess.Popen(["Xvfb", "-displayfd", str(write_fd), "-screen", "0", "1280x1024x24", "-nolisten", "tcp"], pass_fds=(write_fd,), stderr=subprocess.DEVNULL) os.close(write_fd) try: with os.fdopen(read_fd) as pipe: display = pipe.readline().strip() assert display, "Xvfb startup failed" environment = dict(os.environ, DISPLAY=":"+display, XDG_SESSION_TYPE="x11", QT_ACCESSIBILITY="1", QT_LINUX_ACCESSIBILITY_ALWAYS_ON="1") for name in ("DBUS_SESSION_BUS_ADDRESS", "AT_SPI_BUS_ADDRESS", "WAYLAND_DISPLAY", "I3SOCK"): environment.pop(name, None) subprocess.run(["dbus-run-session", "--", sys.executable, __file__, "--inside", directory], env=environment, check=True, timeout=100) finally: server.terminate(); server.wait(timeout=3) if __name__ == "__main__": main()