147 lines
6.2 KiB
Python
147 lines
6.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Private Xvfb/AT-SPI proof for the speech history and monitor helper."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import select
|
|
import subprocess
|
|
import sys
|
|
|
|
sys.dont_write_bytecode = True
|
|
from native_x11 import wait_for
|
|
|
|
|
|
def inside():
|
|
import pyatspi
|
|
|
|
binary = os.environ.get("CTHULHU_BINARY", str(Path(__file__).resolve().parents[1] / "cthulhu"))
|
|
|
|
def walk(node):
|
|
yield node
|
|
for child in node:
|
|
if child is not None:
|
|
yield from walk(child)
|
|
|
|
def window_named(title):
|
|
for application in pyatspi.Registry.getDesktop(0):
|
|
for node in application:
|
|
if node and node.name == title:
|
|
return node
|
|
return None
|
|
|
|
def open_helper(mode, items=None):
|
|
process = subprocess.Popen(
|
|
[binary, "--speech-history-ui"], stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE, text=True,
|
|
)
|
|
send(process, {"mode": mode, "items": items or []})
|
|
return process
|
|
|
|
def send(process, value):
|
|
process.stdin.write(json.dumps(value) + "\n")
|
|
process.stdin.flush()
|
|
|
|
def request(process):
|
|
assert select.select([process.stdout], [], [], 5)[0], "no speech history response"
|
|
return json.loads(process.stdout.readline())
|
|
|
|
def click(node):
|
|
assert node.queryComponent().grabFocus(), ("cannot focus", node.name)
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", "space"], check=True)
|
|
|
|
history = open_helper("history", ["Newest speech", "Older speech", "Another line"])
|
|
try:
|
|
window = wait_for(lambda: window_named("Speech History - Cthulhu"), "history window missing")
|
|
controls = {node.name: node for node in walk(window) if node.name}
|
|
for name in ("Filter:", "Newest speech", "Older speech", "Another line",
|
|
"Copy to clipboard", "Remove from history", "Close"):
|
|
assert name in controls, ("missing history control", name, list(controls))
|
|
table = next(node for node in walk(window) if node.getRole() == pyatspi.ROLE_TABLE)
|
|
assert table.queryComponent().grabFocus(), "cannot focus history table"
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", "Down"], check=True)
|
|
wait_for(lambda: table.querySelection().getSelectedChild(0).name == "Older speech",
|
|
"Down did not select the next history entry")
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", "Up"], check=True)
|
|
wait_for(lambda: table.querySelection().getSelectedChild(0).name == "Newest speech",
|
|
"Up did not restore the first history entry")
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", "o"], check=True)
|
|
wait_for(lambda: table.queryTable().nRows == 1, "typing did not filter by prefix")
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", "BackSpace"], check=True)
|
|
wait_for(lambda: table.queryTable().nRows == 3, "backspace did not clear the filter")
|
|
assert table.queryComponent().grabFocus(), "cannot refocus history table"
|
|
click(controls["Copy to clipboard"])
|
|
assert request(history) == {"action": "copy"}
|
|
copied = subprocess.check_output(["xclip", "-selection", "clipboard", "-o"], text=True)
|
|
assert copied == "Newest speech", copied
|
|
click(controls["Remove from history"])
|
|
removed = request(history)
|
|
assert removed == {"action": "remove", "selection": "Newest speech"}, removed
|
|
send(history, {"items": ["Older speech", "Another line"]})
|
|
wait_for(lambda: table.queryTable().nRows == 2, "removal did not refresh the list")
|
|
click(controls["Remove from history"])
|
|
assert request(history)["selection"] == "Older speech"
|
|
send(history, {"items": ["Another line"]})
|
|
wait_for(lambda: table.queryTable().nRows == 1, "second removal did not refresh the list")
|
|
click(controls["Remove from history"])
|
|
assert request(history)["selection"] == "Another line"
|
|
send(history, {"items": []})
|
|
wait_for(lambda: table.queryTable().nRows == 0, "removing the final item left a row")
|
|
click(controls["Close"])
|
|
assert history.wait(timeout=5) == 0
|
|
finally:
|
|
if history.poll() is None:
|
|
history.kill()
|
|
history.wait(timeout=5)
|
|
|
|
monitor = open_helper("monitor")
|
|
try:
|
|
window = wait_for(lambda: window_named("Speech Monitor - Cthulhu"), "monitor window missing")
|
|
def monitor_text():
|
|
for node in walk(window):
|
|
if node.name == "Speech monitor output":
|
|
return node.queryText().getText(0, -1)
|
|
return ""
|
|
send(monitor, {"append": "Live utterance"})
|
|
wait_for(lambda: "Live utterance" in monitor_text(), "monitor text missing")
|
|
controls = {node.name: node for node in walk(window) if node.name}
|
|
click(controls["Clear"])
|
|
wait_for(lambda: "Live utterance" not in monitor_text(), "monitor text was not cleared")
|
|
click(controls["Close"])
|
|
assert monitor.wait(timeout=5) == 0
|
|
finally:
|
|
if monitor.poll() is None:
|
|
monitor.kill()
|
|
monitor.wait(timeout=5)
|
|
print("PASS: speech history accessibility, copy, remove, and live monitor")
|
|
|
|
|
|
def main():
|
|
if "--inside" in sys.argv:
|
|
inside()
|
|
return
|
|
read_fd, write_fd = os.pipe()
|
|
server = subprocess.Popen(
|
|
["Xvfb", "-displayfd", str(write_fd), "-screen", "0", "1024x768x24", "-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()
|
|
assert display, "Xvfb failed"
|
|
environment = dict(os.environ, DISPLAY=":" + display, GDK_BACKEND="x11",
|
|
GTK_MODULES="gail:atk-bridge", GIO_USE_VFS="local")
|
|
for key in ("DBUS_SESSION_BUS_ADDRESS", "AT_SPI_BUS_ADDRESS", "WAYLAND_DISPLAY"):
|
|
environment.pop(key, None)
|
|
subprocess.run(
|
|
["dbus-run-session", "--", sys.executable, str(Path(__file__).resolve()), "--inside"],
|
|
env=environment, check=True, timeout=60,
|
|
)
|
|
finally:
|
|
server.terminate()
|
|
server.wait(timeout=5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|