#!/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)