309 lines
13 KiB
Python
309 lines
13 KiB
Python
#!/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()
|