208 lines
10 KiB
Python
208 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Firefox/Thunderbird reading with a private reader, X11, profile and SSIP."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
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
|
|
|
|
|
|
TITLE = "Cthulhu Mozilla Reading Probe"
|
|
CONTENT = ["Reading start", "First readable paragraph.", "Reading section",
|
|
"Second readable paragraph.", "Local destination link", "Final readable paragraph."]
|
|
HTML = f'''<!doctype html><html lang="en"><meta charset="utf-8"><title>{TITLE}</title>
|
|
<body><h1>{CONTENT[0]}</h1><p>{CONTENT[1]}</p><h2>{CONTENT[2]}</h2>
|
|
<p>{CONTENT[3]}</p><a href="#destination">{CONTENT[4]}</a>
|
|
<p id="destination">{CONTENT[5]}</p></body></html>'''
|
|
|
|
|
|
def write_profile(profile, application):
|
|
profile.mkdir()
|
|
prefs = {"browser.startup.homepage_override.mstone": "ignore",
|
|
"datareporting.policy.firstRunURL": "",
|
|
"browser.shell.checkDefaultBrowser": False,
|
|
"browser.startup.page": 0,
|
|
"browser.newtabpage.enabled": False,
|
|
"datareporting.policy.dataSubmissionEnabled": False}
|
|
if application == "thunderbird":
|
|
(profile / "Mail" / "Local Folders").mkdir(parents=True)
|
|
prefs.update({"mail.shell.checkDefaultClient": False,
|
|
"mailnews.start_page.enabled": False,
|
|
"mail.provider.enabled": False,
|
|
"mail.rights.override": True,
|
|
"mailnews.database.global.indexer.enabled": False,
|
|
"mail.accountmanager.accounts": "account1",
|
|
"mail.accountmanager.defaultaccount": "account1",
|
|
"mail.accountmanager.localfoldersserver": "server1",
|
|
"mail.account.account1.server": "server1",
|
|
"mail.account.account1.identities": "id1",
|
|
"mail.server.server1.type": "none",
|
|
"mail.server.server1.hostname": "Local Folders",
|
|
"mail.server.server1.userName": "nobody",
|
|
"mail.server.server1.name": "Local Folders",
|
|
"mail.server.server1.directory": str(profile / "Mail" / "Local Folders"),
|
|
"mail.server.server1.directory-rel": "[ProfD]Mail/Local Folders",
|
|
"mail.identity.id1.fullName": "Username",
|
|
"mail.identity.id1.useremail": "username@example.invalid",
|
|
"mail.identity.id1.valid": True,
|
|
"mail.compose.autosave": False})
|
|
(profile / "user.js").write_text("".join(
|
|
f"user_pref({json.dumps(key)}, {json.dumps(value)});\n" for key, value in prefs.items()))
|
|
|
|
|
|
def walk(node):
|
|
if node is not None:
|
|
yield node
|
|
for child in node:
|
|
yield from walk(child)
|
|
|
|
|
|
def run_inside(application):
|
|
import pyatspi
|
|
directory = Path(os.environ["CTHULHU_MOZILLA_EVIDENCE"])
|
|
executable = shutil.which(application)
|
|
if not executable:
|
|
raise RuntimeError(f"{application} executable missing")
|
|
profile = directory / "profile"
|
|
write_profile(profile, application)
|
|
fixture = directory / "reading.html"
|
|
fixture.write_text(HTML)
|
|
args = [executable, "--no-remote", "--profile", str(profile), "--new-window", fixture.as_uri()]
|
|
if application == "thunderbird":
|
|
fixture = directory / "message.eml"
|
|
fixture.write_text("From: Username <username@example.invalid>\n"
|
|
"To: Recipient <recipient@example.invalid>\n"
|
|
f"Subject: {TITLE}\nMIME-Version: 1.0\n"
|
|
"Content-Type: text/html; charset=UTF-8\n\n" + HTML)
|
|
args = [executable, "--no-remote", "--profile", str(profile), "-file", str(fixture)]
|
|
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")
|
|
server = PreviewSpeechServer(directory / "speech.sock")
|
|
processes, report = [], []
|
|
logPath = directory / "reader.log"
|
|
|
|
def key(chord):
|
|
subprocess.run(["xdotool", "key", "--clearmodifiers", chord], check=True)
|
|
|
|
def find_document():
|
|
for app in pyatspi.Registry.getDesktop(0):
|
|
if app.name.lower() == application:
|
|
for node in walk(app):
|
|
if node.getRoleName() == "document web" and node.name == TITLE:
|
|
return node
|
|
|
|
def check(chord, expected):
|
|
offset = len(server.snapshot())
|
|
key(chord)
|
|
wait_for(lambda: any(row["text"] == expected
|
|
for row in server.snapshot()[offset:]),
|
|
f"{application}: {chord} did not speak {expected!r}", timeout=8)
|
|
# Normal presentation queues several chunks without SSIP end events.
|
|
# The recorder completes the final chunk; earlier chunks can remain open.
|
|
wait_for(lambda: server.snapshot()[-1]["finished"], "presentation did not settle")
|
|
rows = server.snapshot()[offset:]
|
|
actual = [row["text"] for row in rows if row["text"] in CONTENT]
|
|
assert actual == [expected], (chord, expected, rows)
|
|
assert not any("Could not" in row["text"] for row in rows), (chord, rows)
|
|
report.append({"key": chord, "expected": expected, "speech": rows})
|
|
|
|
try:
|
|
processes.append(subprocess.Popen(["i3", "-c", str(wmConfig)],
|
|
stdout=subprocess.DEVNULL, stderr=(directory / "wm.log").open("w")))
|
|
wait_for(lambda: "0x" in subprocess.check_output(
|
|
["xprop", "-root", "_NET_SUPPORTING_WM_CHECK"], text=True), "private WM missing")
|
|
processes.append(subprocess.Popen([
|
|
os.environ["CTHULHU_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 missing")
|
|
processes.append(subprocess.Popen(args, env=dict(os.environ, MOZ_ACCESSIBILITY_ENABLED="1"),
|
|
stdout=subprocess.DEVNULL, stderr=(directory / "application.log").open("w")))
|
|
document = wait_for(find_document, "fixture document missing", timeout=25)
|
|
window = subprocess.check_output(["xdotool", "search", "--onlyvisible", "--name", TITLE],
|
|
text=True).splitlines()[0]
|
|
subprocess.run(["xdotool", "windowactivate", "--sync", window], check=True)
|
|
assert document.queryComponent().grabFocus(), "document refused focus"
|
|
wait_for(lambda: document.getState().contains(pyatspi.STATE_FOCUSED), "document not focused")
|
|
wait_for(lambda: server.snapshot() and server.snapshot()[-1]["finished"],
|
|
"initial focus presentation did not settle")
|
|
check("ctrl+Home", CONTENT[0])
|
|
for text in CONTENT[1:]:
|
|
check("Down", text)
|
|
for text in reversed(CONTENT[:-1]):
|
|
check("Up", text)
|
|
check("h", CONTENT[2])
|
|
check("Down", CONTENT[3])
|
|
check("shift+h", CONTENT[2])
|
|
check("shift+h", CONTENT[0])
|
|
check("k", CONTENT[4])
|
|
check("ctrl+Home", CONTENT[0])
|
|
offset = len(server.snapshot())
|
|
key("KP_Add")
|
|
wait_for(lambda: any(row["text"] == CONTENT[-1] and row["finished"]
|
|
for row in server.snapshot()[offset:]), "Read All did not finish", timeout=12)
|
|
rows = server.snapshot()[offset:]
|
|
assert [row["text"] for row in rows if row["text"] in CONTENT] == CONTENT, rows
|
|
assert not any("Could not" in row["text"] for row in rows), rows
|
|
report.append({"key": "KP_Add", "speech": rows})
|
|
print(f"PASS: {application} forward/reverse lines, heading-to-arrow, link and Read All", 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():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("application", choices=("firefox", "thunderbird", "both"), nargs="?", default="both")
|
|
parser.add_argument("--inside", action="store_true", help=argparse.SUPPRESS)
|
|
args = parser.parse_args()
|
|
if not os.environ.get("CTHULHU_BINARY"):
|
|
parser.error("Set CTHULHU_BINARY to a freshly built reader")
|
|
if args.inside:
|
|
run_inside(args.application)
|
|
return
|
|
applications = ("firefox", "thunderbird") if args.application == "both" else (args.application,)
|
|
for application in applications:
|
|
directory = Path(tempfile.mkdtemp(prefix=f"cthulhu-{application}-reading-"))
|
|
print(f"{application} evidence: {directory}", flush=True)
|
|
environment = dict(os.environ, XDG_SESSION_TYPE="x11", CTHULHU_MOZILLA_EVIDENCE=str(directory))
|
|
for name in ("DBUS_SESSION_BUS_ADDRESS", "AT_SPI_BUS_ADDRESS", "WAYLAND_DISPLAY"):
|
|
environment.pop(name, None)
|
|
with (directory / "session.log").open("w") as log:
|
|
run_private_session(["xvfb-run", "-a", "-s", "-screen 0 1280x900x24", "dbus-run-session", "--",
|
|
sys.executable, str(Path(__file__).resolve()), application, "--inside"],
|
|
env=environment, stderr=log, check=True, timeout=120)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|