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