#!/usr/bin/env python3 """Isolated native AT-SPI tree capture and invalidation proof.""" import os import ast import json from pathlib import Path import subprocess import sys import tempfile import time from native_test_session import run_private_session 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 run_inside(): if os.environ.get("CTHULHU_NATIVE_EDIT_LABELS") == "1": run_edit_label_probe() return if os.environ.get("CTHULHU_FOREGROUND_PID") == "1": if os.environ.get("CTHULHU_FOREGROUND_PROXY") == "1": run_proxy_probe() else: run_probe("list") return if os.environ.get("CTHULHU_MOUSE_REVIEW") == "1": run_probe("list") return if os.environ.get("CTHULHU_SPEECH_METADATA") == "1": run_probe("list") run_probe("list", echo_focus=True) return # Keep the original ListBox defunct proof; GtkBox also forwards additions. for container in ("list", "box"): run_probe(container) run_probe("box", flat=True) run_probe("box", flat=True, terminal=True) run_probe("box", flat=True, terminal=True, terminal_mode="large") run_probe("box", flat=True, terminal=True, terminal_mode="prompt") run_table_probe() run_table_probe(multi=True) run_table_probe(list_role=True) run_table_probe(multi=True, list_role=True) run_table_probe(list_role=True, speech=True) if os.environ.get("CTHULHU_TEST_THUNAR") == "1": run_thunar_probe() def run_edit_label_probe(): fixture = subprocess.Popen([sys.executable, str(Path(__file__).with_name("gtk_edit_labels.py"))]) try: import pyatspi def find_window(): return any(window.name == "Edit label probe" for app in pyatspi.Registry.getDesktop(0) for window in app) wait_for(find_window, "edit label probe missing") environment = dict(os.environ, GOCACHE="/tmp/cthulhu-go-cache", GOPROXY="off") subprocess.run( ["go", "test", "./internal/atspi", "-run", "^TestNativeEditFieldLabels$", "-count=1", "-timeout=25s", "-v"], cwd=Path(__file__).resolve().parents[1], env=environment, check=True, ) finally: fixture.terminate() fixture.wait(timeout=3) def run_thunar_probe(): with tempfile.TemporaryDirectory(prefix="cthulhu-thunar-") as temporary: home = Path(temporary) folder = home / "Files" folder.mkdir() for name in ("Alpha", "Beta"): (folder / name).write_text("Example content\n", encoding="utf-8") environment = dict(os.environ, HOME=str(home), XDG_CONFIG_HOME=str(home / ".config"), XDG_CACHE_HOME=str(home / ".cache"), XDG_DATA_HOME=str(home / ".local/share")) fixture = subprocess.Popen(["thunar", "--window", str(folder)], env=environment) try: def find_window(): result = subprocess.run( ["xdotool", "search", "--onlyvisible", "--class", "Thunar"], capture_output=True, text=True, ) return result.stdout.strip().splitlines()[0] if result.stdout.strip() else None window = wait_for(find_window, "Thunar window missing") subprocess.run(["xdotool", "windowfocus", window], check=True) subprocess.run(["xdotool", "key", "ctrl+2"], check=True) environment.update(CTHULHU_NATIVE_THUNAR="1", GOCACHE="/tmp/cthulhu-cache") subprocess.run( ["go", "test", "./internal/atspi", "-run", "^TestNativeThunarNestedFileCellLabel$", "-count=1", "-timeout=20s"], cwd=Path(__file__).resolve().parents[1], env=environment, check=True, timeout=25, ) finally: if fixture.poll() is None: fixture.terminate() fixture.wait(timeout=3) def run_table_probe(multi: bool = False, screen_lock: bool = False, list_role: bool = False, speech: bool = False) -> None: fixture_env = dict(os.environ) if multi: fixture_env["CTHULHU_GTK_TABLE_MULTI"] = "1" if list_role: fixture_env["CTHULHU_GTK_TABLE_LIST"] = "1" fixture = subprocess.Popen([sys.executable, str(Path(__file__).with_name("gtk_table.py"))], env=fixture_env) try: def find_window(): result = subprocess.run( ["xdotool", "search", "--onlyvisible", "--name", "^Table presentation probe$"], capture_output=True, text=True, ) return result.stdout.strip().splitlines()[0] if result.stdout.strip() else None window = wait_for(find_window, "GTK table probe window missing") subprocess.run(["xdotool", "windowfocus", window], check=True) environment = dict(os.environ, CTHULHU_NATIVE_GUI_TABLE="1", CTHULHU_GUI_TABLE_WINDOW=window) if list_role: environment["CTHULHU_NATIVE_GUI_LIST_TABLE"] = "1" if multi: environment["CTHULHU_NATIVE_GUI_TABLE_MULTI"] = "1" environment.setdefault("GOCACHE", "/tmp/cthulhu-cache") test_name = "^TestNativeGUITablePresentation$" if screen_lock: test_name = "^TestNativeGUITableScreenLockRecovery$" elif multi: test_name = "^TestNativeGUITableMultiselectionFocus$" elif list_role: test_name = "^TestNativeGUIListTableSpeech$" if speech else "^TestNativeGUIListTableFocus$" subprocess.run( ["go", "test", "./internal/app" if speech else "./internal/atspi", "-v", "-run", test_name, "-count=1", "-timeout=20s"], cwd=Path(__file__).resolve().parents[1], env=environment, check=True, timeout=25, ) finally: if fixture.poll() is None: fixture.terminate() fixture.wait(timeout=3) def run_proxy_probe(): # Real D-Bus proxy and separate mount namespaces, with the shared instance # metadata Flatpak gives its application and proxy. No installed Flatpak # application or running desktop reader is used. address = ast.literal_eval(subprocess.check_output([ "gdbus", "call", "--session", "--dest", "org.a11y.Bus", "--object-path", "/org/a11y/bus", "--method", "org.a11y.Bus.GetAddress", ], text=True))[0] with tempfile.TemporaryDirectory(prefix="cthulhu-flatpak-") as directory: root = Path(directory) info = root / "info" info.write_text("[Application]\nname=org.example.AccessibilityProbe\n" "[Instance]\ninstance-id=12345\n") # A fresh tmpfs root lets us add the metadata without writing to /. prefix = ["bwrap", "--die-with-parent"] for path in ("/usr", "/etc", "/var", "/run", "/tmp", "/home", "/proc", "/bin", "/sbin", "/lib", "/lib64"): if Path(path).is_symlink(): prefix += ["--symlink", os.readlink(path), path] elif Path(path).exists(): prefix += ["--ro-bind", path, path] prefix += ["--dev", "/dev", "--bind", directory, directory, "--ro-bind", str(info), "/.flatpak-info"] read_fd, write_fd = os.pipe() socket = str(root / "bus") proxy = subprocess.Popen(prefix + ["xdg-dbus-proxy", address, socket, "--fd=" + str(write_fd)], pass_fds=(write_fd,)) os.close(write_fd) try: with os.fdopen(read_fd, "rb") as ready: if ready.read(1) != b"x": raise RuntimeError("Accessibility proxy failed to start") fixture_prefix = prefix if os.environ.get("CTHULHU_FOREGROUND_PROXY_OTHER") == "1": other = root / "other-info" other.write_text(info.read_text().replace("12345", "67890")) fixture_prefix = prefix + ["--ro-bind", str(other), "/.flatpak-info"] run_probe("list", fixture_prefix=fixture_prefix, fixture_overrides={"AT_SPI_BUS_ADDRESS": "unix:path=" + socket, "GSETTINGS_BACKEND": "memory"}) finally: proxy.terminate() proxy.wait(timeout=3) def run_probe(container, flat=False, terminal=False, terminal_mode="", echo_focus=False, fixture_prefix=(), fixture_overrides=None): environment = dict(os.environ, CTHULHU_TREE_CONTAINER=container) environment["CTHULHU_VTE_MODE"] = terminal_mode prefix = list(fixture_prefix) pass_fds = () if prefix: read_fd, write_fd = os.pipe() prefix += ["--info-fd", str(write_fd)] pass_fds = (write_fd,) fixture = subprocess.Popen(prefix + [ sys.executable, str(Path(__file__).with_name("vte_terminal.py" if terminal else "gtk_tree.py")), ], env=dict(environment, **(fixture_overrides or {})), pass_fds=pass_fds) try: fixture_pid = fixture.pid if prefix: os.close(write_fd) with os.fdopen(read_fd) as information: fixture_pid = json.load(information)["child-pid"] def find_window(): result = subprocess.run( ["xdotool", "search", "--onlyvisible", "--name", "^Cthulhu Tree Probe$"], capture_output=True, text=True, ) return result.stdout.strip().splitlines()[0] if result.stdout.strip() else None window = wait_for(find_window, "GTK tree probe window missing") subprocess.run(["xdotool", "windowfocus", window], check=True) if os.environ.get("CTHULHU_FOREGROUND_PID") == "1": environment["CTHULHU_TREE_PID"] = str(fixture_pid) subprocess.run(["xprop", "-id", window, "-f", "_NET_WM_PID", "32c", "-set", "_NET_WM_PID", "252"], check=True) environment.update(CTHULHU_NATIVE_ATSPI="1", CTHULHU_TREE_WINDOW=window) if os.environ.get("CTHULHU_MOUSE_REVIEW") == "1": environment["CTHULHU_NATIVE_MOUSE_REVIEW"] = "1" if flat: environment["CTHULHU_NATIVE_FLAT"] = "1" if terminal: environment["CTHULHU_NATIVE_TERMINAL_FLAT"] = "1" environment.setdefault("GOCACHE", "/tmp/cthulhu-cache") subprocess.run( [ "go", "test", "./internal/review" if flat else "./internal/atspi", "-run", "^TestNativeForegroundProcessIdentity$" if os.environ.get("CTHULHU_FOREGROUND_PID") == "1" else "^TestNativeMouseReviewPointerAndHitTest$" if os.environ.get("CTHULHU_MOUSE_REVIEW") == "1" else "^TestNativeEchoEditFieldFocus$" if echo_focus else "^TestNativeSpeechMetadata$" if os.environ.get("CTHULHU_SPEECH_METADATA") == "1" else "^TestNativeTerminalFlatReview$" if terminal else "^TestNativeFlatReview$" if flat else "^TestNativeTreeCaptureDefunctAndShutdownInvalidation$", "-count=1", "-timeout=20s", ], cwd=Path(__file__).resolve().parents[1], env=environment, check=True, timeout=25, ) finally: if fixture.poll() is None: fixture.terminate() fixture.wait(timeout=3) def main() -> None: if len(sys.argv) == 2 and sys.argv[1] == "--inside": run_inside() return if len(sys.argv) == 2 and sys.argv[1] == "--inside-screen-lock": run_table_probe(screen_lock=True) return if len(sys.argv) == 2 and sys.argv[1] == "--inside-table": run_table_probe() run_table_probe(multi=True) run_table_probe(list_role=True) run_table_probe(multi=True, list_role=True) run_table_probe(list_role=True, speech=True) return with tempfile.TemporaryDirectory(prefix="cthulhu-atspi-"): read_fd, write_fd = os.pipe() server = subprocess.Popen( ["Xvfb", "-displayfd", str(write_fd), "-screen", "0", "1920x1080x24", "-nolisten", "tcp"], pass_fds=(write_fd,), stderr=subprocess.DEVNULL, ) os.close(write_fd) try: with os.fdopen(read_fd) as reader: display = reader.readline().strip() if not display: raise RuntimeError("Xvfb failed to start") environment = dict(os.environ, DISPLAY=":" + display, XDG_SESSION_TYPE="x11") environment.pop("DBUS_SESSION_BUS_ADDRESS", None) environment.pop("AT_SPI_BUS_ADDRESS", None) environment.pop("WAYLAND_DISPLAY", None) run_private_session( ["dbus-run-session", "--", sys.executable, str(Path(__file__).resolve()), "--inside-screen-lock" if len(sys.argv) == 2 and sys.argv[1] == "--screen-lock" else "--inside-table" if len(sys.argv) == 2 and sys.argv[1] == "--table-only" else "--inside"], env=environment, check=True, timeout=60, ) finally: server.terminate() server.wait(timeout=3) if __name__ == "__main__": main()