#!/usr/bin/env python3 """Changing-document navigation with a real reader, private X11/bus and SSIP.""" import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import time 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 def run_inside(): binary = os.environ["CTHULHU_BINARY"] directory = Path(tempfile.mkdtemp(prefix="cthulhu-browser-navigation-")) 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") processes = [] report = [] logPath = directory / "reader.log" try: processes.append(subprocess.Popen(["i3", "-c", str(wmConfig)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)) wait_for(lambda: "0x" in subprocess.check_output( ["xprop", "-root", "_NET_SUPPORTING_WM_CHECK"], text=True), "private WM missing") processes.append(subprocess.Popen([ 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 "remote D-Bus service ready" in logPath.read_text(), "private reader did not start") browser = os.environ.get("CTHULHU_BROWSER") or shutil.which("brave") fixture = Path(__file__).with_name("browser-changing.html").resolve() processes.append(subprocess.Popen([ browser, "--user-data-dir=" + str(directory / "profile"), "--no-first-run", "--no-default-browser-check", "--disable-background-networking", "--disable-component-update", "--disable-sync", "--password-store=basic", "--force-renderer-accessibility", "--ozone-platform=x11", "--new-window", fixture.as_uri()], stdout=subprocess.DEVNULL, stderr=(directory / "browser.log").open("w"))) def key(chord): subprocess.run(["xdotool", "key", "--clearmodifiers", chord], check=True) def logs(): result = [] for line in logPath.read_text().splitlines(): try: result.append(json.loads(line)) except json.JSONDecodeError: pass # A writer can still be completing the last record. return result wait_for(lambda: subprocess.run([ "xdotool", "search", "--onlyvisible", "--name", "Cthulhu Changing Document Probe"], capture_output=True).returncode == 0, "private browser did not open") subprocess.run(["xdotool", "mousemove", "0", "0"], check=True) for _ in range(12): offset = len(server.snapshot()) key("Tab") time.sleep(0.2) if any(record["text"] == "Changing document" for record in server.snapshot()[offset:]): break else: raise AssertionError("first document heading did not receive focus") def move(chord, expected, delay=0.2): logOffset, speechOffset = len(logs()), len(server.snapshot()) key(chord) result = wait_for(lambda: next((row for row in logs()[logOffset:] if row.get("message") in {"document navigation completed", "document caret navigation completed"}), None), f"reader did not complete {chord}", timeout=5) assert result.get("moved"), (chord, result) step = {"key": chord, "result": result} report.append(step) wait_for(lambda: any(expected(record["text"]) for record in server.snapshot()[speechOffset:]), f"missing speech for {chord}", timeout=5) time.sleep(delay) speech = [record["text"] for record in server.snapshot()[speechOffset:]] assert not any("Could not" in text for text in speech), (chord, speech) headings = [text for text in speech if text.startswith("Section ") or text == "Statistics"] assert all(expected(text) for text in headings), (chord, "repeated earlier heading", speech) step["speech"] = speech for cycle in range(2): for index in range(1, 11): name = f"Section {index}" move("h", lambda text, name=name: text == name, 1.1 if index == 3 else 0.2) move("h", lambda text: text == "Statistics") for index in range(10, 0, -1): name = f"Section {index}" move("shift+h", lambda text, name=name: text == name) move("shift+h", lambda text: text == "Changing document") # Go directly to the final heading and repeatedly traverse text nodes # which are replaced while the native caret and selection settle. move("shift+h", lambda text: text == "Statistics") for _ in range(4): move("Down", lambda text: text.startswith("Current total ")) move("Right", lambda text: text == "u") move("Home", lambda text: text == "C") move("ctrl+Right", lambda text: text == "Current") move("Down", lambda text: text == "Following stable line") move("Up", lambda text: text.startswith("Current total ")) move("Up", lambda text: text == "Statistics") move("Down", lambda text: text.startswith("Current total ")) move("Down", lambda text: text == "Following stable line") logOffset, speechOffset = len(logs()), len(server.snapshot()) key("Return") wait_for(lambda: any(row.get("message") == "document activation failed" for row in logs()[logOffset:]), "plain-text Enter did not complete") time.sleep(0.2) assert not server.snapshot()[speechOffset:], "plain-text Enter produced reader feedback" move("Down", lambda text: text == "Custom action") for count in (1, 2): logOffset, speechOffset = len(logs()), len(server.snapshot()) key("Return") wait_for(lambda: any(row.get("message") == "document activation completed" for row in logs()[logOffset:]), "custom clickable activation failed") wait_for(lambda: any(record["text"] == f"Action count {count}" for record in server.snapshot()[speechOffset:]), "page did not report exactly one activation") time.sleep(0.2) speech = [record["text"] for record in server.snapshot()[speechOffset:]] assert speech == [f"Action count {count}"], ("activation added feedback", speech) logOffset = len(logs()) subprocess.run(["xdotool", "click", "1"], check=True) wait_for(lambda: any(row.get("message") == "released virtual caret for pointer input" for row in logs()[logOffset:]), "shared pointer signal did not reach the reader") print(f"PASS: {len(report)} navigation commands with changing text", flush=True) finally: for process in reversed(processes): process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) server.close() (directory / "report.json").write_text(json.dumps(report, indent=2)) (directory / "speech.json").write_text(json.dumps(server.snapshot(), indent=2)) def main(): if sys.argv[1:] == ["--inside"]: run_inside() return readFd, writeFd = os.pipe() display = subprocess.Popen(["Xvfb", "-displayfd", str(writeFd), "-screen", "0", "1280x900x24", "-nolisten", "tcp"], pass_fds=(writeFd,), stderr=subprocess.DEVNULL) os.close(writeFd) try: with os.fdopen(readFd) as stream: number = stream.readline().strip() if not number: raise RuntimeError("private X server failed to start") environment = dict(os.environ, DISPLAY=":" + number, XDG_SESSION_TYPE="x11") for name in ("DBUS_SESSION_BUS_ADDRESS", "AT_SPI_BUS_ADDRESS", "WAYLAND_DISPLAY"): environment.pop(name, None) run_private_session(["dbus-run-session", "--", sys.executable, str(Path(__file__).resolve()), "--inside"], env=environment, check=True, timeout=120) finally: display.terminate() display.wait(timeout=5) if __name__ == "__main__": main()