233 lines
11 KiB
Python
233 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Real shell/VTE/reader regression with private X11, AT-SPI and recorded SSIP."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
sys.dont_write_bytecode = True
|
|
from native_echo_preview import PreviewSpeechServer
|
|
from native_x11 import wait_for
|
|
from native_test_session import run_private_session
|
|
|
|
TITLE = "Terminal scrolling probe"
|
|
PROMPT = "Username@Host> "
|
|
|
|
|
|
def run_fixture():
|
|
import gi
|
|
gi.require_version("Gtk", "3.0")
|
|
gi.require_version("Vte", "2.91")
|
|
from gi.repository import GLib, Gtk, Vte
|
|
|
|
window = Gtk.Window(title=TITLE)
|
|
window.set_default_size(600, 220)
|
|
terminal = Vte.Terminal()
|
|
terminal.set_scrollback_lines(0)
|
|
window.add(terminal)
|
|
window.connect("destroy", Gtk.main_quit)
|
|
window.show_all()
|
|
terminal.grab_focus()
|
|
childEnv = [f"HOME={os.environ['HOME']}", "PATH=/usr/bin:/bin", "TERM=xterm",
|
|
"LC_ALL=C.UTF-8", "HISTFILE=/dev/null", "PS1=" + PROMPT, "PS2=Continue> "]
|
|
terminal.spawn_async(Vte.PtyFlags.DEFAULT, os.environ["HOME"],
|
|
["/bin/bash", "--noprofile", "--norc"], childEnv,
|
|
GLib.SpawnFlags.DEFAULT, None, None, -1, None, None, None)
|
|
Gtk.main()
|
|
|
|
|
|
def run_inside():
|
|
import pyatspi
|
|
directory = Path(tempfile.mkdtemp(prefix="cthulhu-terminal-runtime-"))
|
|
print(f"Evidence: {directory}", flush=True)
|
|
server = PreviewSpeechServer(directory / "speech.sock")
|
|
config = directory / "settings.toml"
|
|
config.write_text('''schema-version = 2
|
|
active-profile = "default"
|
|
[profiles.default.speech]
|
|
output-module = "main"
|
|
[profiles.default.braille]
|
|
enabled = false
|
|
[profiles.default.sound]
|
|
enabled = false
|
|
[profiles.default.key-echo]
|
|
enabled = false
|
|
''')
|
|
wmConfig = directory / "i3.config"
|
|
wmConfig.write_text(f"font pango:monospace 10\nipc-socket {directory / 'i3.sock'}\n"
|
|
f'for_window [title="^{TITLE}$"] floating enable\n')
|
|
logPath = directory / "reader.log"
|
|
processes = []
|
|
report = {}
|
|
|
|
def key(chord):
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", chord], check=True)
|
|
|
|
def type_text(text):
|
|
offset = len(logs())
|
|
subprocess.run(["xdotool", "type", "--clearmodifiers", "--delay", "40", "--", text], check=True)
|
|
# Wait for the reader to receive VTE's final character event before
|
|
# sending another edit; otherwise VTE can merge both redraws.
|
|
wait_for(lambda: any(row.get("message") == "received accessibility event"
|
|
and row.get("type") == "object:text-caret-moved"
|
|
and row.get("native_caret_offset") == terminal.queryText().caretOffset
|
|
for row in logs()[offset:]), "typed text did not reach the reader")
|
|
|
|
def logs():
|
|
result = []
|
|
for line in logPath.read_text().splitlines():
|
|
try:
|
|
result.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return result
|
|
|
|
def settle():
|
|
wait_for(lambda: server.snapshot() and server.snapshot()[-1]["finished"],
|
|
"last speech request did not complete")
|
|
|
|
try:
|
|
processes.append(subprocess.Popen(["i3", "-c", str(wmConfig)],
|
|
stdout=subprocess.DEVNULL, stderr=(directory / "wm.log").open("w")))
|
|
wait_for(lambda: "0x" in subprocess.check_output(
|
|
["xprop", "-root", "_NET_SUPPORTING_WM_CHECK"], text=True), "private WM missing")
|
|
processes.append(subprocess.Popen([
|
|
os.environ["CTHULHU_BINARY"], "--config", str(config), "--no-tentacles",
|
|
"--speech-socket", str(directory / "speech.sock"), "--debug-file", str(logPath)],
|
|
stdout=subprocess.DEVNULL, stderr=(directory / "reader-stderr.log").open("w")))
|
|
wait_for(lambda: logPath.exists() and "listening for focus events" in logPath.read_text(),
|
|
"private reader did not start")
|
|
processes.append(subprocess.Popen([sys.executable, __file__, "--fixture"],
|
|
stdout=subprocess.DEVNULL, stderr=(directory / "fixture.log").open("w")))
|
|
|
|
def find_terminal():
|
|
for application in pyatspi.Registry.getDesktop(0):
|
|
for window in application:
|
|
if window and window.name == TITLE:
|
|
return next((child for child in window if child and child.getRoleName() == "terminal"), None)
|
|
|
|
terminal = wait_for(find_terminal, "terminal fixture missing")
|
|
text = terminal.queryText()
|
|
wait_for(lambda: PROMPT in text.getText(0, -1), "shell prompt missing")
|
|
windowId = subprocess.check_output(
|
|
["xdotool", "search", "--onlyvisible", "--name", "^" + TITLE + "$"], text=True).splitlines()[0]
|
|
subprocess.run(["xdotool", "windowactivate", "--sync", windowId], check=True)
|
|
settle()
|
|
# The unchecked default types into the real shell, without touching
|
|
# clipboard contents or submitting a command.
|
|
subprocess.run(["xclip", "-selection", "clipboard", "-in"],
|
|
input="emoji clipboard fixture", text=True, check=True)
|
|
key("Insert+ctrl+e")
|
|
wait_for(lambda: any(row["text"].startswith("Emoji Menu. All tab")
|
|
for row in server.snapshot()), "emoji menu did not open")
|
|
for chord in ("Tab", "a", "s", "t", "r", "o", "n", "a", "u", "t", "Return"):
|
|
key(chord)
|
|
wait_for(lambda: text.getText(0, -1).rstrip().endswith(PROMPT + "🧑🚀"),
|
|
"complete emoji sequence did not reach the shell")
|
|
clipboard = subprocess.check_output(["xclip", "-selection", "clipboard", "-out"], text=True)
|
|
if clipboard != "emoji clipboard fixture":
|
|
raise AssertionError("emoji typing changed clipboard contents")
|
|
key("ctrl+u")
|
|
wait_for(lambda: text.getText(0, -1).rstrip().endswith(PROMPT.rstrip()),
|
|
"emoji command line did not clear")
|
|
# Checked means copy only, including when the terminal has input focus.
|
|
offset = len(server.snapshot())
|
|
key("Insert+ctrl+e")
|
|
wait_for(lambda: any(row["text"].startswith("Emoji Menu. All tab")
|
|
for row in server.snapshot()[offset:]), "second emoji menu did not open")
|
|
for chord in ("Tab", "g", "o", "a", "t", "Tab", "space", "shift+Tab", "Return"):
|
|
key(chord)
|
|
wait_for(lambda: any(row["text"] == "Copied Goat to clipboard"
|
|
for row in server.snapshot()[offset:]), "emoji was not copied")
|
|
if subprocess.check_output(["xclip", "-selection", "clipboard", "-out"], text=True) != "🐐":
|
|
raise AssertionError("emoji clipboard sequence changed")
|
|
if not text.getText(0, -1).rstrip().endswith(PROMPT.rstrip()):
|
|
raise AssertionError("copy mode also typed into the terminal")
|
|
report["emoji"] = {"typed_sequence": "🧑🚀", "copy_only": "🐐", "clipboard_preserved": True}
|
|
print("PASS: emoji typing, full sequence, clipboard preservation and copy-only mode", flush=True)
|
|
settle()
|
|
logOffset = len(logs())
|
|
speechOffset = len(server.snapshot())
|
|
for row in range(16):
|
|
output = f"Output{row:02d}"
|
|
type_text("echo " + output)
|
|
key("Return")
|
|
wait_for(lambda: any(output in item["text"] for item in server.snapshot()[speechOffset:]),
|
|
f"command output {output} was not spoken")
|
|
wait_for(lambda: text.getText(0, -1).rstrip().endswith(PROMPT.rstrip()), "prompt did not return")
|
|
settle()
|
|
scrollLogs = logs()[logOffset:]
|
|
replay = [row["utterance"] for row in scrollLogs
|
|
if row.get("message") == "delivering accessible presentation"
|
|
and row.get("cause") == "object:text-changed:delete"
|
|
and "Username@Host>" in row.get("utterance", "")]
|
|
recorded = [row["text"] for row in server.snapshot()[speechOffset:]]
|
|
ignored = sum(row.get("message") == "skipped terminal scrollback deletion" for row in scrollLogs)
|
|
report.update(replayed_prompts=replay, ignored_scroll_deletions=ignored)
|
|
if "--expect-replay" in sys.argv:
|
|
assert replay and any(value in recorded for value in replay), (replay, recorded)
|
|
print(f"CONFIRMED: X11 sent {len(replay)} old prompt deletions to SSIP", flush=True)
|
|
else:
|
|
assert not replay, replay
|
|
assert ignored > 0, "fixture did not exercise scroll deletions"
|
|
print(f"PASS: {ignored} scroll deletions suppressed; all 16 command outputs reached SSIP", flush=True)
|
|
|
|
def check_edit(chord, expected):
|
|
offset = len(server.snapshot())
|
|
key(chord)
|
|
wait_for(lambda: any(row["text"] == expected for row in server.snapshot()[offset:]),
|
|
f"{chord} did not speak {expected!r}")
|
|
settle()
|
|
report[chord] = [row["text"] for row in server.snapshot()[offset:]]
|
|
|
|
type_text("abcd")
|
|
check_edit("BackSpace", "d")
|
|
check_edit("ctrl+w", "abc")
|
|
type_text("abcd")
|
|
check_edit("Left", "d")
|
|
check_edit("Delete", "d")
|
|
wait_for(lambda: text.getText(0, -1).rstrip().endswith(PROMPT + "abc"), "Delete did not remove d")
|
|
offset = len(server.snapshot())
|
|
type_text("Z")
|
|
key("Return")
|
|
wait_for(lambda: any("abcZ" in row["text"] for row in server.snapshot()[offset:]),
|
|
"editing command did not reach shell")
|
|
settle()
|
|
report["editing_command"] = [row["text"] for row in server.snapshot()[offset:]]
|
|
# History and flat review must still announce terminal content.
|
|
check_edit("Up", "abcZ")
|
|
key("ctrl+u")
|
|
check_edit("KP_Up", "Username@Host>")
|
|
print("PASS: Backspace, Ctrl+W, Left, Delete, history and flat review", flush=True)
|
|
finally:
|
|
(directory / "speech.json").write_text(json.dumps(server.snapshot(), indent=2))
|
|
(directory / "report.json").write_text(json.dumps(report, indent=2))
|
|
for process in reversed(processes):
|
|
if process.poll() is None:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|
|
server.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if "--fixture" in sys.argv:
|
|
run_fixture()
|
|
elif "--inside" in sys.argv:
|
|
run_inside()
|
|
else:
|
|
with tempfile.TemporaryDirectory(prefix="cthulhu-terminal-home-") as temporary:
|
|
environment = dict(os.environ, HOME=temporary, XDG_CONFIG_HOME=temporary,
|
|
XDG_DATA_HOME=temporary, XDG_CACHE_HOME=temporary,
|
|
XDG_SESSION_TYPE="x11", GDK_BACKEND="x11",
|
|
GTK_MODULES="gail:atk-bridge", GIO_USE_VFS="local")
|
|
for name in ("DISPLAY", "WAYLAND_DISPLAY", "DBUS_SESSION_BUS_ADDRESS", "AT_SPI_BUS_ADDRESS"):
|
|
environment.pop(name, None)
|
|
run_private_session(["xvfb-run", "-a", "dbus-run-session", "--", sys.executable,
|
|
__file__, "--inside", *sys.argv[1:]], env=environment, check=True, timeout=100)
|