728 lines
42 KiB
Python
728 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
"""Isolated X11 command consumption and pass-through proof; never uses desktop DISPLAY."""
|
|
import json
|
|
import os
|
|
import ctypes
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
|
|
def wait_for(check, message, timeout=15):
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
value = check()
|
|
if value:
|
|
return value
|
|
time.sleep(0.02)
|
|
raise RuntimeError(message)
|
|
|
|
|
|
def count_records(path, *needles):
|
|
return sum(all(needle in line for needle in needles) for line in path.read_text().splitlines())
|
|
|
|
|
|
def press_physical_key(symbol):
|
|
# xdotool's keysym synthesis can toggle NumLock between press and release.
|
|
# XTest the physical keycode without synthesizing any modifier changes.
|
|
x11 = ctypes.CDLL('libX11.so.6')
|
|
xtst = ctypes.CDLL('libXtst.so.6')
|
|
x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
|
|
x11.XOpenDisplay.restype = ctypes.c_void_p
|
|
x11.XKeysymToKeycode.argtypes = [ctypes.c_void_p, ctypes.c_ulong]
|
|
x11.XKeysymToKeycode.restype = ctypes.c_ubyte
|
|
x11.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
|
x11.XCloseDisplay.argtypes = [ctypes.c_void_p]
|
|
xtst.XTestFakeKeyEvent.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_int, ctypes.c_ulong]
|
|
display = x11.XOpenDisplay(None)
|
|
if not display:
|
|
raise RuntimeError('private X11 display unavailable')
|
|
try:
|
|
code = x11.XKeysymToKeycode(display, symbol)
|
|
assert code, 'physical key missing from private keymap'
|
|
assert xtst.XTestFakeKeyEvent(display, code, 1, 0)
|
|
assert xtst.XTestFakeKeyEvent(display, code, 0, 0)
|
|
x11.XSync(display, 0)
|
|
finally:
|
|
x11.XCloseDisplay(display)
|
|
|
|
|
|
def set_active_window(window_id):
|
|
x11 = ctypes.CDLL('libX11.so.6')
|
|
x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
|
|
x11.XOpenDisplay.restype = ctypes.c_void_p
|
|
x11.XDefaultRootWindow.argtypes = [ctypes.c_void_p]
|
|
x11.XDefaultRootWindow.restype = ctypes.c_ulong
|
|
x11.XInternAtom.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int]
|
|
x11.XInternAtom.restype = ctypes.c_ulong
|
|
x11.XChangeProperty.argtypes = [
|
|
ctypes.c_void_p,
|
|
ctypes.c_ulong,
|
|
ctypes.c_ulong,
|
|
ctypes.c_ulong,
|
|
ctypes.c_int,
|
|
ctypes.c_int,
|
|
ctypes.c_void_p,
|
|
ctypes.c_int,
|
|
]
|
|
x11.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
|
x11.XCloseDisplay.argtypes = [ctypes.c_void_p]
|
|
display = x11.XOpenDisplay(None)
|
|
if not display:
|
|
raise RuntimeError('private X11 display unavailable')
|
|
try:
|
|
root = x11.XDefaultRootWindow(display)
|
|
property_atom = x11.XInternAtom(display, b'_NET_ACTIVE_WINDOW', False)
|
|
window_atom = x11.XInternAtom(display, b'WINDOW', False)
|
|
active_window = ctypes.c_ulong(window_id)
|
|
x11.XChangeProperty(
|
|
display,
|
|
root,
|
|
property_atom,
|
|
window_atom,
|
|
32,
|
|
0,
|
|
ctypes.byref(active_window),
|
|
1,
|
|
)
|
|
x11.XSync(display, 0)
|
|
finally:
|
|
x11.XCloseDisplay(display)
|
|
|
|
|
|
def set_window_class(window_id, instance_name, class_name):
|
|
x11 = ctypes.CDLL('libX11.so.6')
|
|
x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
|
|
x11.XOpenDisplay.restype = ctypes.c_void_p
|
|
x11.XDefaultRootWindow.argtypes = [ctypes.c_void_p]
|
|
x11.XDefaultRootWindow.restype = ctypes.c_ulong
|
|
x11.XInternAtom.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int]
|
|
x11.XInternAtom.restype = ctypes.c_ulong
|
|
x11.XChangeProperty.argtypes = [
|
|
ctypes.c_void_p,
|
|
ctypes.c_ulong,
|
|
ctypes.c_ulong,
|
|
ctypes.c_ulong,
|
|
ctypes.c_int,
|
|
ctypes.c_int,
|
|
ctypes.c_void_p,
|
|
ctypes.c_int,
|
|
]
|
|
x11.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
|
x11.XCloseDisplay.argtypes = [ctypes.c_void_p]
|
|
display = x11.XOpenDisplay(None)
|
|
if not display:
|
|
raise RuntimeError('private X11 display unavailable')
|
|
try:
|
|
window = ctypes.c_ulong(window_id)
|
|
property_atom = x11.XInternAtom(display, b'WM_CLASS', False)
|
|
string_atom = x11.XInternAtom(display, b'STRING', False)
|
|
class_data = ctypes.create_string_buffer(f'{instance_name}\0{class_name}\0'.encode())
|
|
x11.XChangeProperty(
|
|
display,
|
|
window,
|
|
property_atom,
|
|
string_atom,
|
|
8,
|
|
0,
|
|
class_data,
|
|
len(class_data) - 1,
|
|
)
|
|
x11.XSync(display, 0)
|
|
finally:
|
|
x11.XCloseDisplay(display)
|
|
|
|
|
|
def check_speech_mode_shortcut(log, events, modifier):
|
|
for modifier_first in (True, False):
|
|
before_s = events.read_text().count(', s)')
|
|
before_command = count_records(log, '"message":"keyboard command received"',
|
|
'"command":"cycle-speech-mode"')
|
|
subprocess.run(['xdotool', 'keydown', modifier, 'keydown', 's', 'keydown', 's'], check=True)
|
|
release_keys = [modifier, 's'] if modifier_first else ['s', modifier]
|
|
subprocess.run(['xdotool', 'keyup', release_keys[0], 'keyup', release_keys[1]], check=True)
|
|
wait_for(lambda: count_records(log, '"message":"keyboard command received"',
|
|
'"command":"cycle-speech-mode"') == before_command + 1,
|
|
'speech mode shortcut did not dispatch once')
|
|
time.sleep(0.05)
|
|
assert events.read_text().count(', s)') == before_s, 'speech mode shortcut leaked S'
|
|
|
|
|
|
def run_inside(directory):
|
|
root = Path(directory)
|
|
config = root / "settings.toml"
|
|
config.write_text('[general]\nenableSpeech = false\nenableSound = false\nkeyboardLayout = 1\n')
|
|
speech_modes_only = os.environ.get('CTHULHU_NATIVE_SPEECH_MODES_ONLY') == '1'
|
|
if speech_modes_only:
|
|
config.write_text('[general]\nenableSpeech = false\nenableSound = false\nkeyboardLayout = 1\n'
|
|
'cthulhuModifierKeys = ["Insert", "KP_Insert", "Caps_Lock"]\n')
|
|
log = root / "cthulhu.log"
|
|
events = root / "keys.log"
|
|
binary = Path(os.environ.get('CTHULHU_BINARY', str(Path(__file__).resolve().parents[1] / "cthulhu")))
|
|
processes = []
|
|
try:
|
|
with events.open("w") as output:
|
|
probe = subprocess.Popen(['stdbuf', '-oL', 'xev', '-name', 'CthulhuInputProbe', '-event', 'keyboard'], stdout=output, stderr=subprocess.DEVNULL)
|
|
processes.append(probe)
|
|
def find_window():
|
|
result = subprocess.run(['xdotool', 'search', '--name', '^CthulhuInputProbe$'], capture_output=True, text=True)
|
|
return result.stdout.strip().splitlines()[0] if result.stdout.strip() else None
|
|
window = wait_for(find_window, 'probe window missing')
|
|
subprocess.run(['xdotool', 'windowfocus', window], check=True)
|
|
nativeTest = os.environ.get('CTHULHU_XRECORD_TEST')
|
|
if nativeTest:
|
|
subprocess.run([nativeTest, '-test.run', '^TestNative(ModifierPressRelease|CommandGrabSuspensionAndRestoration|BrowseKeyTransitions|DispatchBeforeRelease|CustomBindingsUpdateAndRelease|TentacleBindingsUpdateAndRelease|ModifierConfigurationUpdate|LockStatesAndShiftCapsLock|XTermHandoffKeepsKeypadReleased|SuspendedQueuedKey|LearnModeCapturesKeysAndRestoresCommands|SecureInputDropsKeysAndRestoresGrabs)$', '-test.timeout=40s'], env=dict(os.environ, CTHULHU_NATIVE_X11='1', CTHULHU_NATIVE_BROWSER='1'), check=True, timeout=45, cwd=Path(__file__).resolve().parents[1] / 'internal/platform/xorg')
|
|
if nativeTest and os.environ.get('CTHULHU_NATIVE_LISTENER_ONLY') == '1':
|
|
print('PASS: native command bindings, press/release ordering, browse transitions, suspension, custom updates, and grab rollback')
|
|
return
|
|
app = subprocess.Popen([str(binary), '--config', str(config), '--debug-file', str(log)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
processes.append(app)
|
|
def ready():
|
|
if app.poll() is not None:
|
|
raise RuntimeError(log.read_text() if log.exists() else 'Cthulhu exited before logging')
|
|
return log.exists() and 'listening for focus events' in log.read_text()
|
|
wait_for(ready, 'Cthulhu startup timed out')
|
|
if speech_modes_only:
|
|
for layout in (1, 2):
|
|
for modifier in ('Insert', 'KP_Insert', 'Caps_Lock'):
|
|
check_speech_mode_shortcut(log, events, modifier)
|
|
before_s = events.read_text().count(', s)')
|
|
subprocess.run(['xdotool', 'key', 's'], check=True)
|
|
wait_for(lambda: events.read_text().count(', s)') == before_s + 2,
|
|
'ordinary S stayed grabbed after speech mode shortcut')
|
|
before_save = count_records(log, '"message":"speech setting command received"',
|
|
'"command":"save-speech-settings"')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'keydown', 'Control_L',
|
|
'keydown', 'Shift_L', 'key', 's', 'keyup', 'Shift_L',
|
|
'keyup', 'Control_L', 'keyup', 'Insert'], check=True)
|
|
wait_for(lambda: count_records(log, '"message":"speech setting command received"',
|
|
'"command":"save-speech-settings"') == before_save + 1,
|
|
'speech mode shortcut broke speech settings save')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'key', 'q', 'keyup', 'Insert'], check=True)
|
|
assert app.wait(timeout=5) == 0, log.read_text()
|
|
print(f'PASS: layout {layout} speech-mode shortcut, three modifiers, repeat, release orders, '
|
|
'ordinary S, speech settings save, and quit')
|
|
if layout == 1:
|
|
config.write_text('[general]\nenableSpeech = false\nenableSound = false\nkeyboardLayout = 2\n'
|
|
'cthulhuModifierKeys = ["Insert", "KP_Insert", "Caps_Lock"]\n')
|
|
log = root / 'speech-mode-laptop.log'
|
|
app = subprocess.Popen([str(binary), '--config', str(config), '--debug-file', str(log)],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
processes.append(app)
|
|
wait_for(ready, 'laptop startup timed out')
|
|
return
|
|
selection_probe = subprocess.Popen([
|
|
sys.executable,
|
|
str(Path(__file__).with_name('gtk_selection.py')),
|
|
], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
processes.append(selection_probe)
|
|
def find_selection_window():
|
|
result = subprocess.run(
|
|
['xdotool', 'search', '--name', '^Cthulhu Selection Probe$'],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return result.stdout.strip().splitlines()[0] if result.stdout.strip() else None
|
|
selection_window = wait_for(find_selection_window, 'selection probe window missing')
|
|
subprocess.run(['xdotool', 'windowfocus', selection_window], check=True)
|
|
try:
|
|
wait_for(
|
|
lambda: any(
|
|
'"message":"queued accessible presentation"' in line
|
|
and '"type":"object:selection-changed"' in line
|
|
for line in log.read_text().splitlines()
|
|
),
|
|
'focused GTK selection was not presented',
|
|
)
|
|
except RuntimeError:
|
|
print(log.read_text(), file=sys.stderr)
|
|
raise
|
|
assert selection_probe.wait(timeout=5) == 0, 'GTK selection probe failed'
|
|
print('PASS: focused GTK selection resolved to its selected child')
|
|
|
|
xterm = subprocess.Popen(
|
|
[
|
|
'xterm',
|
|
'-T',
|
|
'Cthulhu XTerm Handoff',
|
|
'-e',
|
|
sys.executable,
|
|
'-c',
|
|
(
|
|
'import pathlib, sys, tty\n'
|
|
'tty.setcbreak(sys.stdin)\n'
|
|
"while sys.stdin.read(1) != 'q':\n"
|
|
' pass\n'
|
|
f"pathlib.Path({str(root / 'xterm-key.log')!r}).write_text('q')\n"
|
|
),
|
|
],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
processes.append(xterm)
|
|
def find_xterm():
|
|
result = subprocess.run(
|
|
['xdotool', 'search', '--onlyvisible', '--name', '^Cthulhu XTerm Handoff$'],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return result.stdout.strip().splitlines()[0] if result.stdout else None
|
|
xterm_window = wait_for(find_xterm, 'XTerm handoff window missing')
|
|
def focus_xterm():
|
|
result = subprocess.run(
|
|
['xdotool', 'windowfocus', '--sync', xterm_window],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return result.returncode == 0
|
|
wait_for(focus_xterm, 'XTerm handoff window could not be focused')
|
|
set_active_window(int(xterm_window))
|
|
wait_for(lambda: xterm.poll() is None, 'XTerm key-recording child exited before focus handoff')
|
|
wait_for(
|
|
lambda: 'suspended for XTerm focus handoff' in log.read_text(),
|
|
'XTerm focus did not suspend Cthulhu',
|
|
)
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'key', 'q', 'keyup', 'Insert'], check=True)
|
|
xterm_key = root / 'xterm-key.log'
|
|
wait_for(lambda: xterm_key.exists(), 'Insert+Q did not reach the focused XTerm')
|
|
assert xterm_key.read_text() == 'q', 'XTerm received an unexpected key'
|
|
assert app.poll() is None, 'Cthulhu consumed Insert+Q while XTerm owned focus'
|
|
print('PASS: XTerm focus suspended Cthulhu and passed through Insert+Q')
|
|
|
|
subprocess.run(['xdotool', 'windowfocus', '--sync', window], check=True)
|
|
set_active_window(int(window))
|
|
before_unknown_q = events.read_text().count('keysym 0x71, q)')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'key', 'q', 'keyup', 'Insert'], check=True)
|
|
wait_for(
|
|
lambda: events.read_text().count('keysym 0x71, q)') == before_unknown_q + 2,
|
|
'unknown active window restored Cthulhu command grabs',
|
|
)
|
|
assert app.poll() is None, 'unknown active window stopped Cthulhu'
|
|
print('PASS: unknown active window remained suspended for Fenrir handoff')
|
|
|
|
set_window_class(
|
|
int(window),
|
|
'CthulhuInputProbe',
|
|
'CthulhuInputProbe',
|
|
)
|
|
subprocess.run(['xdotool', 'windowfocus', '--sync', window], check=True)
|
|
set_active_window(int(window))
|
|
wait_for(
|
|
lambda: 'resumed after XTerm focus handoff' in log.read_text(),
|
|
'leaving XTerm did not restore Cthulhu',
|
|
)
|
|
assert app.poll() is None, 'Cthulhu exited while restoring command grabs'
|
|
subprocess.run(['xdotool', 'key', 'q', 'h', 'shift+h', 'b', 'x', 'l', 'Tab', 'Control_L', 'Shift_L'], check=True)
|
|
wait_for(lambda: 'keysym 0x71, q)' in events.read_text(), 'ordinary q was consumed')
|
|
assert app.poll() is None, 'ordinary typing quit Cthulhu'
|
|
before = events.read_text().count('keysym 0x71, q)')
|
|
assert before >= 2 and before % 2 == 0, ('ordinary q press/release imbalance', before)
|
|
|
|
before_x = events.read_text().count('keysym 0x78, x)')
|
|
before_insert = events.read_text().count('keysym 0xff63, Insert)')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'keydown', 'x', 'keyup', 'Insert', 'keyup', 'x'], check=True)
|
|
wait_for(lambda: events.read_text().count('keysym 0x78, x)') == before_x + 2, 'unbound Insert+x did not reach application')
|
|
wait_for(lambda: events.read_text().count('keysym 0xff63, Insert)') == before_insert + 2, 'unbound Insert press/release was not balanced')
|
|
assert app.poll() is None, 'unbound Insert chord stopped Cthulhu'
|
|
|
|
before_return = events.read_text().count('keysym 0xff0d, Return)')
|
|
before_insert = events.read_text().count('keysym 0xff63, Insert)')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'key', 'Return', 'keyup', 'Insert'], check=True)
|
|
wait_for(lambda: events.read_text().count('keysym 0xff0d, Return)') == before_return + 2, 'desktop Insert+Return did not reach application')
|
|
wait_for(lambda: events.read_text().count('keysym 0xff63, Insert)') == before_insert + 2, 'desktop Insert+Return was not balanced')
|
|
assert log.read_text().count('where am I command received') == 0, 'desktop Insert+Return incorrectly routed Where Am I'
|
|
|
|
before_keypad_enter = events.read_text().count('keysym 0xff8d, KP_Enter)')
|
|
subprocess.run(['xdotool', 'key', 'KP_Enter'], check=True)
|
|
wait_for(lambda: log.read_text().count('where am I command received') == 1, 'desktop KP Enter did not route Where Am I')
|
|
time.sleep(0.05)
|
|
assert events.read_text().count('keysym 0xff8d, KP_Enter)') == before_keypad_enter, 'desktop Where Am I KP Enter leaked to application'
|
|
|
|
subprocess.run(['xdotool', 'key', 'Num_Lock'], check=True)
|
|
subprocess.run(['xdotool', 'key', 'KP_Enter'], check=True)
|
|
wait_for(lambda: log.read_text().count('where am I command received') == 2, 'NumLock prevented desktop KP Enter Where Am I')
|
|
time.sleep(0.05)
|
|
assert events.read_text().count('keysym 0xff8d, KP_Enter)') == before_keypad_enter, 'NumLock desktop KP Enter leaked to application'
|
|
subprocess.run(['xdotool', 'key', 'Num_Lock'], check=True)
|
|
assert app.poll() is None, 'where am I command stopped Cthulhu'
|
|
|
|
before_space = events.read_text().count('keysym 0x20, space)')
|
|
before_insert = events.read_text().count('keysym 0xff63, Insert)')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'keydown', 'space', 'keydown', 'space', 'keyup', 'Insert', 'keyup', 'space'], check=True)
|
|
wait_for(lambda: log.read_text().count('preferences command received') == 1, 'preferences command was not routed exactly once')
|
|
time.sleep(0.05)
|
|
assert events.read_text().count('keysym 0x20, space)') == before_space, 'preferences Space leaked to application'
|
|
assert events.read_text().count('keysym 0xff63, Insert)') == before_insert, 'preferences Insert leaked to application'
|
|
assert app.poll() is None, 'preferences command stopped Cthulhu'
|
|
def preferences_window():
|
|
result = subprocess.run(['xdotool', 'search', '--onlyvisible', '--name', '^Cthulhu Preferences$'], capture_output=True, text=True)
|
|
return result.stdout.strip().splitlines()[0] if result.stdout.strip() else None
|
|
preferences = wait_for(preferences_window, 'preferences window did not open')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'key', 'space', 'keyup', 'Insert'], check=True)
|
|
wait_for(lambda: log.read_text().count('preferences command received') == 2, 'second preferences command missing')
|
|
assert preferences_window() == preferences, 'repeated command replaced the preferences window'
|
|
set_active_window(int(preferences))
|
|
subprocess.run(['xdotool', 'windowfocus', preferences], check=True)
|
|
focus_log_offset = len(log.read_text())
|
|
focus_before = log.read_text().count('"type":"object:state-changed:focused"')
|
|
def preferences_control(controlName, controlRole):
|
|
import pyatspi
|
|
def walk(node):
|
|
yield node
|
|
for child in node:
|
|
if child is not None:
|
|
yield from walk(child)
|
|
for application in pyatspi.Registry.getDesktop(0):
|
|
for window in application:
|
|
if window is None or window.name != 'Cthulhu Preferences':
|
|
continue
|
|
for node in walk(window):
|
|
if node.name == controlName and node.getRole() == controlRole:
|
|
return node
|
|
return None
|
|
import pyatspi
|
|
progressBackground = wait_for(lambda: preferences_control('Report background progress bars', pyatspi.ROLE_CHECK_BOX), 'preferences progress background option missing')
|
|
assert progressBackground.queryComponent().grabFocus(), 'cannot focus preferences progress background option'
|
|
wait_for(lambda: '"role":"check box"' in log.read_text()[focus_log_offset:],
|
|
'preferences progress background focus event missing')
|
|
subprocess.run(['xdotool', 'key', '--delay', '100', *(["Tab"] * 12)], check=True)
|
|
try:
|
|
wait_for(
|
|
lambda: log.read_text().count('"type":"object:state-changed:focused"') >= focus_before + 8,
|
|
'preferences Tab traversal stopped producing focus events',
|
|
)
|
|
except RuntimeError:
|
|
focused = [
|
|
line for line in log.read_text().splitlines()
|
|
if '"type":"object:state-changed:focused"' in line
|
|
]
|
|
print("\n".join(focused[-16:]), file=sys.stderr)
|
|
raise
|
|
focus_log = log.read_text()[focus_log_offset:]
|
|
assert '"role":"combo box"' in focus_log, (
|
|
'preferences combo focused as an internal button',
|
|
focus_log,
|
|
)
|
|
combo_snapshot_durations = []
|
|
for line in focus_log.splitlines():
|
|
if '"role":"combo box"' not in line or '"type":"object:state-changed:focused"' not in line:
|
|
continue
|
|
record = json.loads(line)
|
|
if 'snapshot_duration' in record:
|
|
combo_snapshot_durations.append(record['snapshot_duration'])
|
|
assert combo_snapshot_durations, 'preferences combo focus snapshots missing durations'
|
|
slowest_combo_snapshot = max(combo_snapshot_durations)
|
|
assert slowest_combo_snapshot < 100_000_000, (
|
|
'preferences combo focus snapshot was slow',
|
|
slowest_combo_snapshot,
|
|
)
|
|
progress_snapshot_durations = []
|
|
for line in focus_log.splitlines():
|
|
if '"role":"check box"' not in line or '"type":"object:state-changed:focused"' not in line:
|
|
continue
|
|
record = json.loads(line)
|
|
if 'snapshot_duration' in record:
|
|
progress_snapshot_durations.append(record['snapshot_duration'])
|
|
assert progress_snapshot_durations, 'preferences progress checkbox focus snapshots missing durations'
|
|
slowest_progress_snapshot = max(progress_snapshot_durations)
|
|
assert slowest_progress_snapshot < 20_000_000, (
|
|
'preferences progress checkbox focus snapshot was slow',
|
|
slowest_progress_snapshot,
|
|
)
|
|
# Start at the retained keyboard layout group and verify the
|
|
# current General page's actual keyboard path.
|
|
desktop = wait_for(lambda: preferences_control('Desktop', pyatspi.ROLE_RADIO_BUTTON),
|
|
'preferences Desktop layout missing')
|
|
assert desktop.queryComponent().grabFocus(), 'cannot focus Desktop layout'
|
|
# GTK combo focus belongs to its internal button, which is not
|
|
# exposed in the combo's AT-SPI children. Check the reader's
|
|
# semantic focus events rather than the outer widget's state.
|
|
for controlName, expectedRole in (
|
|
('Time format', 'combo box'),
|
|
('Date format', 'combo box'),
|
|
('Progress bar output', 'combo box'),
|
|
('Report background progress bars', 'check box'),
|
|
):
|
|
tabOffset = len(log.read_text())
|
|
subprocess.run(['xdotool', 'key', 'Tab'], check=True)
|
|
wait_for(lambda: any(
|
|
record.get('type') == 'object:state-changed:focused' and
|
|
record.get('role') == expectedRole
|
|
for record in (json.loads(line) for line in log.read_text()[tabOffset:].splitlines())
|
|
), 'Tab did not reach ' + controlName)
|
|
set_active_window(int(preferences))
|
|
subprocess.run(['xdotool', 'windowfocus', preferences, 'key', 'Escape'], check=True)
|
|
wait_for(lambda: preferences_window() is None, 'Escape did not close preferences')
|
|
set_active_window(int(window))
|
|
subprocess.run(['xdotool', 'windowfocus', window], check=True)
|
|
print('PASS: Insert+X and desktop Insert+Return passed through; KP Enter routed Where Am I with both NumLock states; repeated Insert+Space routed preferences once')
|
|
|
|
object_commands = [
|
|
('Up', 'parent'),
|
|
('Down', 'first-child'),
|
|
('Left', 'previous-sibling'),
|
|
('Right', 'next-sibling'),
|
|
]
|
|
for arrow, direction in object_commands:
|
|
before_arrow = events.read_text().count(f', {arrow})')
|
|
before_insert = events.read_text().count('keysym 0xff63, Insert)')
|
|
before_command = count_records(
|
|
log,
|
|
'"message":"object review command received"',
|
|
f'"direction":"{direction}"',
|
|
)
|
|
subprocess.run([
|
|
'xdotool', 'keydown', 'Insert', 'keydown', 'Control_L',
|
|
'key', arrow, 'keyup', 'Control_L', 'keyup', 'Insert',
|
|
], check=True)
|
|
wait_for(
|
|
lambda direction=direction, before_command=before_command: count_records(
|
|
log,
|
|
'"message":"object review command received"',
|
|
f'"direction":"{direction}"',
|
|
) == before_command + 1,
|
|
f'Insert+Control+{arrow} did not route {direction}',
|
|
)
|
|
time.sleep(0.05)
|
|
assert events.read_text().count(f', {arrow})') == before_arrow, f'{arrow} object navigation leaked to application'
|
|
assert events.read_text().count('keysym 0xff63, Insert)') == before_insert, f'Insert+Control+{arrow} modifier leaked to application'
|
|
|
|
before_up = events.read_text().count(', Up)')
|
|
before_insert = events.read_text().count('keysym 0xff63, Insert)')
|
|
before_parent = count_records(
|
|
log,
|
|
'"message":"object review command received"',
|
|
'"direction":"parent"',
|
|
)
|
|
subprocess.run([
|
|
'xdotool', 'keydown', 'Control_L', 'keydown', 'Insert',
|
|
'key', 'Up', 'keyup', 'Insert', 'keyup', 'Control_L',
|
|
], check=True)
|
|
wait_for(
|
|
lambda: count_records(
|
|
log,
|
|
'"message":"object review command received"',
|
|
'"direction":"parent"',
|
|
) == before_parent + 1,
|
|
'Control-first Insert+Control+Up did not route parent',
|
|
)
|
|
time.sleep(0.05)
|
|
assert events.read_text().count(', Up)') == before_up, 'Control-first object navigation leaked Up to application'
|
|
assert events.read_text().count('keysym 0xff63, Insert)') == before_insert, 'Control-first object navigation leaked Insert to application'
|
|
assert app.poll() is None, 'object navigation command stopped Cthulhu'
|
|
print('PASS: Cthulhu+Control+Arrow consumed with balanced Insert-first and Control-first chords')
|
|
|
|
for arrow, command in [('Left', 'select-previous-speech-setting'),
|
|
('Right', 'select-next-speech-setting'),
|
|
('Up', 'increase-speech-setting'),
|
|
('Down', 'decrease-speech-setting')]:
|
|
before_arrow = events.read_text().count(f', {arrow})')
|
|
before_command = count_records(log, '"message":"speech setting command received"',
|
|
f'"command":"{command}"')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'keydown', arrow,
|
|
'keyup', 'Insert', 'keyup', arrow], check=True)
|
|
wait_for(lambda: count_records(log, '"message":"speech setting command received"',
|
|
f'"command":"{command}"') == before_command + 1,
|
|
'speech shortcut did not dispatch: ' + command)
|
|
time.sleep(0.05)
|
|
assert events.read_text().count(f', {arrow})') == before_arrow, 'speech shortcut leaked arrow'
|
|
print('PASS: Cthulhu+Arrow speech commands consumed with modifier released first')
|
|
|
|
check_speech_mode_shortcut(log, events, 'Insert')
|
|
print('PASS: Cthulhu+S consumed once with repeated keydown and both release orders')
|
|
|
|
review_keys = [
|
|
('KP_Home', 'u', 'review-previous-line'),
|
|
('KP_Up', 'i', 'review-current-line'),
|
|
('KP_Page_Up', 'o', 'review-next-line'),
|
|
('KP_Left', 'j', 'review-previous-word'),
|
|
('KP_Begin', 'k', 'review-current-word'),
|
|
('KP_Right', 'l', 'review-next-word'),
|
|
('KP_End', 'm', 'review-previous-character'),
|
|
('KP_Down', 'comma', 'review-current-character'),
|
|
('KP_Page_Down', 'period', 'review-next-character'),
|
|
]
|
|
|
|
def flush_probe_events():
|
|
before_f12 = events.read_text().count(', F12)')
|
|
subprocess.run(['xdotool', 'key', 'F12'], check=True)
|
|
wait_for(lambda: events.read_text().count(', F12)') >= before_f12 + 2,
|
|
'ordinary key did not reach probe after review')
|
|
|
|
for desktop_key, _, command in review_keys:
|
|
before_key = events.read_text().count(', ' + desktop_key + ')')
|
|
before_command = count_records(log, '"message":"flat review command received"',
|
|
'"command":"' + command + '"')
|
|
subprocess.run(['xdotool', 'key', desktop_key], check=True)
|
|
wait_for(lambda: count_records(log, '"message":"flat review command received"',
|
|
'"command":"' + command + '"') == before_command + 1,
|
|
'desktop review command missing: ' + command)
|
|
flush_probe_events()
|
|
assert events.read_text().count(', ' + desktop_key + ')') == before_key, \
|
|
'desktop review key leaked: ' + desktop_key
|
|
for desktop_key, command in [('KP_Home', 'review-window-start'), ('KP_Page_Up', 'review-window-end')]:
|
|
before_key = events.read_text().count(', ' + desktop_key + ')')
|
|
for modifier_first in (False, True):
|
|
before_command = count_records(log, '"message":"flat review command received"',
|
|
'"command":"' + command + '"')
|
|
releases = ['Insert', desktop_key] if modifier_first else [desktop_key, 'Insert']
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'keydown', desktop_key,
|
|
'keyup', releases[0], 'keyup', releases[1]], check=True)
|
|
wait_for(lambda: count_records(log, '"message":"flat review command received"',
|
|
'"command":"' + command + '"') == before_command + 1,
|
|
'desktop window review command missing: ' + command)
|
|
flush_probe_events()
|
|
assert events.read_text().count(', ' + desktop_key + ')') == before_key, \
|
|
'desktop window review key leaked: ' + desktop_key
|
|
before_review = count_records(log, '"message":"flat review command received"')
|
|
subprocess.run(['xdotool', 'key', 'Num_Lock'], check=True)
|
|
before_numeric = events.read_text().count(', KP_7)')
|
|
numeric_offset = len(events.read_text())
|
|
press_physical_key(0xff95) # physical keypad 7/Home, with NumLock on
|
|
try:
|
|
wait_for(lambda: events.read_text().count(', KP_7)') == before_numeric + 2,
|
|
'NumLock numeric keypad was consumed by review')
|
|
except RuntimeError:
|
|
print(events.read_text()[numeric_offset:], file=sys.stderr)
|
|
print('review records before/after:', before_review,
|
|
count_records(log, '"message":"flat review command received"'), file=sys.stderr)
|
|
raise
|
|
assert count_records(log, '"message":"flat review command received"') == before_review
|
|
subprocess.run(['xdotool', 'key', 'Num_Lock'], check=True)
|
|
print('PASS: all nine desktop flat-review keys plus window start/end consumed; NumLock numeric typing preserved')
|
|
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'key', 'space', 'keyup', 'Insert'], check=True)
|
|
preferences = wait_for(preferences_window, 'preferences did not reopen')
|
|
set_active_window(int(preferences))
|
|
subprocess.run(['xdotool', 'windowfocus', preferences], check=True)
|
|
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'key', 'q'], check=True)
|
|
time.sleep(0.1)
|
|
assert app.poll() is None, 'quit occurred before modifier release'
|
|
subprocess.run(['xdotool', 'keyup', 'Insert'], check=True)
|
|
assert app.wait(timeout=10) == 0, log.read_text()
|
|
wait_for(lambda: preferences_window() is None, 'quit left preferences running')
|
|
assert 'quit command received' in log.read_text(), log.read_text()
|
|
time.sleep(0.1)
|
|
assert events.read_text().count('keysym 0x71, q)') == before, 'command q leaked to application'
|
|
subprocess.run(['xdotool', 'windowfocus', window], check=True)
|
|
subprocess.run(['xdotool', 'key', 'q', 'Insert'], check=True)
|
|
wait_for(lambda: events.read_text().count('keysym 0x71, q)') == before + 2, 'q grab remained after quit')
|
|
assert 'keysym 0xff63, Insert)' in events.read_text(), 'Insert grab remained after quit'
|
|
print('PASS: ordinary typing; Insert+Q consumed; release-safe quit; grabs removed')
|
|
for modifier, lock, modifierFirst in [('Insert','Num_Lock',True), ('Insert','Caps_Lock',False), ('KP_Insert',None,True)]:
|
|
if lock:
|
|
subprocess.run(['xdotool','key',lock],check=True)
|
|
app = subprocess.Popen([str(binary),'--config',str(config),'--debug-file',str(root / 'next.log')],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
|
|
processes.append(app)
|
|
nextLog = root / 'next.log'
|
|
wait_for(lambda: nextLog.exists() and 'listening for focus events' in nextLog.read_text(), 'restart did not become ready')
|
|
# Give the startup loop a chance to enter native event dispatch.
|
|
time.sleep(0.05)
|
|
beforeQ = events.read_text().count('keycode 24 ')
|
|
subprocess.run(['xdotool','keydown',modifier,'keydown','q'],check=True)
|
|
if modifierFirst:
|
|
subprocess.run(['xdotool','keyup',modifier,'keyup','q'],check=True)
|
|
else:
|
|
subprocess.run(['xdotool','keyup','q','keyup',modifier],check=True)
|
|
assert app.wait(timeout=5) == 0, nextLog.read_text()
|
|
time.sleep(0.05)
|
|
assert events.read_text().count('keycode 24 ') == beforeQ, ('q leaked with lock keys or alternate release order', modifier, lock, modifierFirst, events.read_text()[-1800:])
|
|
if lock:
|
|
subprocess.run(['xdotool','key',lock],check=True)
|
|
nextLog.unlink()
|
|
print('PASS: NumLock, CapsLock, keypad Insert, alternate release order, repeated startup/shutdown')
|
|
|
|
config.write_text('[general]\nenableSpeech = false\nenableSound = false\nkeyboardLayout = 2\n')
|
|
laptop_log = root / 'laptop.log'
|
|
app = subprocess.Popen([str(binary), '--config', str(config), '--debug-file', str(laptop_log)],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
processes.append(app)
|
|
wait_for(lambda: laptop_log.exists() and 'listening for focus events' in laptop_log.read_text(),
|
|
'laptop layout did not start')
|
|
check_speech_mode_shortcut(laptop_log, events, 'Insert')
|
|
for _, laptop_key, command in review_keys:
|
|
before_key = events.read_text().count(', ' + laptop_key + ')')
|
|
before_insert = events.read_text().count(', Insert)')
|
|
for modifier_first in (False, True):
|
|
expected_command = command
|
|
if modifier_first:
|
|
expected_command = {
|
|
'review-current-word': 'review-spell-word',
|
|
'review-current-character': 'review-phonetic-character',
|
|
}.get(command, command)
|
|
before_command = count_records(laptop_log, '"message":"flat review command received"',
|
|
'"command":"' + expected_command + '"')
|
|
releases = ['Insert', laptop_key] if modifier_first else [laptop_key, 'Insert']
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'keydown', laptop_key,
|
|
'keyup', releases[0], 'keyup', releases[1]], check=True)
|
|
wait_for(lambda: count_records(laptop_log, '"message":"flat review command received"',
|
|
'"command":"' + expected_command + '"') == before_command + 1,
|
|
'laptop review command missing: ' + expected_command)
|
|
flush_probe_events()
|
|
assert events.read_text().count(', ' + laptop_key + ')') == before_key, \
|
|
'laptop review key leaked: ' + laptop_key
|
|
assert events.read_text().count(', Insert)') == before_insert, 'laptop Insert leaked'
|
|
for laptop_key, command in [('u', 'review-window-start'), ('o', 'review-window-end')]:
|
|
for modifier_first in (False, True):
|
|
before_key = events.read_text().count(', ' + laptop_key + ')')
|
|
before_insert = events.read_text().count(', Insert)')
|
|
before_command = count_records(laptop_log, '"message":"flat review command received"',
|
|
'"command":"' + command + '"')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'keydown', 'Control_L',
|
|
'keydown', laptop_key], check=True)
|
|
wait_for(lambda: count_records(laptop_log, '"message":"flat review command received"',
|
|
'"command":"' + command + '"') == before_command + 1,
|
|
'laptop window boundary command did not dispatch on press: ' + command)
|
|
releases = ['Insert', 'Control_L', laptop_key] if modifier_first else [laptop_key, 'Control_L', 'Insert']
|
|
subprocess.run(['xdotool'] + [part for key in releases for part in ('keyup', key)], check=True)
|
|
flush_probe_events()
|
|
assert events.read_text().count(', ' + laptop_key + ')') == before_key, 'laptop boundary key leaked'
|
|
assert events.read_text().count(', Insert)') == before_insert, 'laptop boundary Insert leaked'
|
|
before_keypad = events.read_text().count(', KP_Up)')
|
|
subprocess.run(['xdotool', 'key', 'KP_Up'], check=True)
|
|
wait_for(lambda: events.read_text().count(', KP_Up)') == before_keypad + 2,
|
|
'laptop layout grabbed desktop review keys')
|
|
subprocess.run(['xdotool', 'keydown', 'Insert', 'key', 'q', 'keyup', 'Insert'], check=True)
|
|
assert app.wait(timeout=5) == 0, laptop_log.read_text()
|
|
print('PASS: all nine laptop flat-review keys plus window start/end; both release orders; desktop keypad pass-through; shutdown')
|
|
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] == '--inside':
|
|
run_inside(sys.argv[2])
|
|
return
|
|
with tempfile.TemporaryDirectory(prefix='cthulhu-x11-') as directory:
|
|
readFd, writeFd = os.pipe()
|
|
server = subprocess.Popen(['Xvfb', '-displayfd', str(writeFd), '-screen', '0', '800x600x24', '-nolisten', 'tcp'], pass_fds=(writeFd,), stderr=subprocess.DEVNULL)
|
|
os.close(writeFd)
|
|
try:
|
|
with os.fdopen(readFd) as reader:
|
|
display = reader.readline().strip()
|
|
if not display:
|
|
raise RuntimeError('Xvfb failed to start')
|
|
environment = dict(os.environ, DISPLAY=':' + display)
|
|
environment.pop('DBUS_SESSION_BUS_ADDRESS', None)
|
|
environment.pop('AT_SPI_BUS_ADDRESS', None)
|
|
environment.pop('WAYLAND_DISPLAY', None)
|
|
environment['XDG_SESSION_TYPE'] = 'x11'
|
|
subprocess.run(['dbus-run-session', '--', sys.executable, str(Path(__file__).resolve()), '--inside', directory], env=environment, check=True, timeout=60)
|
|
finally:
|
|
server.terminate()
|
|
server.wait(timeout=3)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|