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