383 lines
21 KiB
Python
383 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Local Thunderbird folders, message lists and composition with a private reader."""
|
|
import argparse
|
|
from email.message import EmailMessage
|
|
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_mozilla import walk, write_profile
|
|
from native_x11 import wait_for
|
|
from native_test_session import run_private_session
|
|
|
|
|
|
SUBJECTS = ["First test message", "Second test message"]
|
|
APPLICATIONS = ("thunderbird", "org.mozilla.thunderbird", "xdg-desktop-portal-gtk")
|
|
|
|
|
|
def seed_mailbox(profile, html):
|
|
write_profile(profile, "thunderbird")
|
|
with (profile / "user.js").open("a") as stream:
|
|
stream.write(f'user_pref("mail.identity.id1.compose_html", {str(html).lower()});\n')
|
|
stream.write('user_pref("mailnews.default_sort_order", 1);\n')
|
|
stream.write('user_pref("mailnews.default_view_flags", 0);\n')
|
|
stream.write('user_pref("mail.openMessageBehavior", 2);\n')
|
|
stream.write('user_pref("mail.forward_message_mode", 2);\n')
|
|
for folder, subjects in {"Inbox": SUBJECTS, "Archive": ["Archived test message"]}.items():
|
|
messages = []
|
|
for index, subject in enumerate(subjects):
|
|
message = EmailMessage()
|
|
message["From"] = "Sender <sender@example.invalid>"
|
|
message["To"] = "Username <username@example.invalid>"
|
|
message["Cc"] = "Other Recipient <other@example.invalid>"
|
|
message["Subject"] = subject
|
|
message["Date"] = f"Sat, 17 Oct 2026 12:0{index}:00 +0000"
|
|
message["Message-ID"] = f"<{folder}-{index}@example.invalid>"
|
|
message.set_content(f"Body of {subject}.\n")
|
|
if folder == "Inbox" and index == 0:
|
|
message.add_attachment("Synthetic attachment contents.\n", subtype="plain",
|
|
filename="test-attachment.txt")
|
|
messages.append("From sender@example.invalid Sat Oct 17 12:00:00 2026\n"
|
|
+ message.as_string() + "\n")
|
|
(profile / "Mail" / "Local Folders" / folder).write_text("".join(messages))
|
|
|
|
|
|
def run_inside(html):
|
|
import pyatspi
|
|
from gi.repository import GLib
|
|
directory = Path(os.environ["CTHULHU_THUNDERBIRD_EVIDENCE"])
|
|
executable = shutil.which("thunderbird")
|
|
if not executable:
|
|
raise RuntimeError("Thunderbird executable missing")
|
|
profile = directory / "profile"
|
|
seed_mailbox(profile, html)
|
|
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 = true
|
|
''')
|
|
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 type_text(text):
|
|
subprocess.run(["xdotool", "type", "--clearmodifiers", "--delay", "35", "--", text], check=True)
|
|
|
|
def find_node(role, name=None, focused=False):
|
|
for app in pyatspi.Registry.getDesktop(0):
|
|
try:
|
|
if app.name.lower() not in APPLICATIONS:
|
|
continue
|
|
for node in walk(app):
|
|
if (node.getRoleName() == role and (name is None or name in node.name)
|
|
and node.getState().contains(pyatspi.STATE_SHOWING)
|
|
and (not focused or node.getState().contains(pyatspi.STATE_FOCUSED))):
|
|
return node
|
|
except GLib.Error:
|
|
# A closing portal dialog can retire a child during traversal.
|
|
# The caller's bounded wait retries against the current tree.
|
|
continue
|
|
|
|
def expect_focus(role, name=None):
|
|
return wait_for(lambda: find_node(role, name, focused=True),
|
|
f"native focus missing: {role} {name!r}", timeout=8)
|
|
|
|
def check(action, operation, expected, role=None, name=None, exact=False, selected=False):
|
|
offset = len(server.snapshot())
|
|
operation()
|
|
node = expect_focus(role, name) if role else None
|
|
wait_for(lambda: any((expected == row["text"] if exact else expected in row["text"])
|
|
for row in server.snapshot()[offset:]),
|
|
f"{action}: missing speech {expected!r}", timeout=8)
|
|
wait_for(lambda: server.snapshot()[-1]["finished"], f"{action}: speech did not settle")
|
|
rows = server.snapshot()[offset:]
|
|
assert not any("Could not" in row["text"] for row in rows), (action, rows)
|
|
if node and role in ("tree item", "list item"):
|
|
if selected or role == "tree item":
|
|
assert node.getState().contains(pyatspi.STATE_SELECTED), (action, "native selection missing")
|
|
assert any(row["text"] == role for row in rows), (action, "spoken row role missing", rows)
|
|
report.append({"action": action, "expected": expected,
|
|
"focus": {"role": node.getRoleName(), "name": node.name} if node else None,
|
|
"speech": rows})
|
|
print(f"PASS: {action}", flush=True)
|
|
return node
|
|
|
|
def read_text(node):
|
|
# HTML editors expose their paragraphs as children; plain text can be
|
|
# exposed directly on the document. Ignore embedded-object placeholders.
|
|
try:
|
|
text = node.queryText().getText(0, -1).replace("\ufffc", "")
|
|
except NotImplementedError:
|
|
text = ""
|
|
if text.strip():
|
|
return text
|
|
return "\n".join(read_text(child) for child in node)
|
|
|
|
def selected_text(node):
|
|
for child in walk(node):
|
|
try:
|
|
text = child.queryText()
|
|
if text.getNSelections():
|
|
start, end = text.getSelection(0)
|
|
selected = text.getText(start, end).replace("\ufffc", "")
|
|
if selected.strip():
|
|
return selected
|
|
except NotImplementedError:
|
|
continue
|
|
return ""
|
|
|
|
def selected_messages():
|
|
return [subject for subject in SUBJECTS
|
|
if (node := find_node("list item", subject)) is not None
|
|
and node.getState().contains(pyatspi.STATE_SELECTED)]
|
|
|
|
def discard_composer():
|
|
key("ctrl+w")
|
|
discard = wait_for(lambda: find_node("button", "Discard changes"),
|
|
"discard confirmation missing")
|
|
check("focus discard button", lambda: discard.queryComponent().grabFocus(),
|
|
"Discard changes", "button", "Discard changes")
|
|
check("discard unsent message", lambda: key("Return"), SUBJECTS[1],
|
|
"list item", SUBJECTS[1], selected=True)
|
|
|
|
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([
|
|
executable, "--no-remote", "--profile", str(profile)],
|
|
env=dict(os.environ, MOZ_ACCESSIBILITY_ENABLED="1"),
|
|
stdout=subprocess.DEVNULL, stderr=(directory / "application.log").open("w")))
|
|
root = wait_for(lambda: find_node("tree item", "Local Folders"), "folder tree missing", timeout=25)
|
|
window = subprocess.check_output([
|
|
"xdotool", "search", "--onlyvisible", "--name", "Local Folders - Mozilla Thunderbird"],
|
|
text=True).splitlines()[0]
|
|
subprocess.run(["xdotool", "windowactivate", "--sync", window], check=True)
|
|
check("focus folder tree", lambda: root.queryComponent().grabFocus(),
|
|
"Local Folders", "tree item", "Local Folders")
|
|
key("Right")
|
|
check("next folder", lambda: key("Down"), "Inbox", "tree item", "Inbox")
|
|
check("last folder", lambda: key("End"), "Archive", "tree item", "Archive")
|
|
check("first folder", lambda: key("Home"), "Local Folders", "tree item", "Local Folders")
|
|
check("return to Inbox", lambda: key("Down"), "Inbox", "tree item", "Inbox")
|
|
check("folder to messages", lambda: key("F6"), SUBJECTS[0], "list item", SUBJECTS[0])
|
|
check("next message", lambda: key("Down"), SUBJECTS[1], "list item", SUBJECTS[1], selected=True)
|
|
check("previous message", lambda: key("Up"), SUBJECTS[0], "list item", SUBJECTS[0], selected=True)
|
|
check("extend message selection", lambda: key("shift+Down"), SUBJECTS[1],
|
|
"list item", SUBJECTS[1], selected=True)
|
|
wait_for(lambda: selected_messages() == SUBJECTS, "two-message selection missing")
|
|
report[-1]["selectedMessages"] = selected_messages()
|
|
check("shrink message selection", lambda: key("shift+Up"), SUBJECTS[0],
|
|
"list item", SUBJECTS[0], selected=True)
|
|
wait_for(lambda: selected_messages() == SUBJECTS[:1], "selection did not shrink")
|
|
report[-1]["selectedMessages"] = selected_messages()
|
|
check("move without extending selection", lambda: key("ctrl+Down"), SUBJECTS[1],
|
|
"list item", SUBJECTS[1])
|
|
wait_for(lambda: selected_messages() == SUBJECTS[:1], "Ctrl+Down changed selection")
|
|
check("add focused message to selection", lambda: key("ctrl+space"), "selected", exact=True)
|
|
wait_for(lambda: selected_messages() == SUBJECTS, "Ctrl+Space did not add message")
|
|
report[-1]["selectedMessages"] = selected_messages()
|
|
check("remove focused message from selection", lambda: key("ctrl+space"), "not selected", exact=True)
|
|
wait_for(lambda: selected_messages() == SUBJECTS[:1], "Ctrl+Space did not remove message")
|
|
report[-1]["selectedMessages"] = selected_messages()
|
|
check("return to first selected message", lambda: key("Up"), SUBJECTS[0],
|
|
"list item", SUBJECTS[0], selected=True)
|
|
check("messages to folders", lambda: key("shift+F6"), "Inbox", "tree item", "Inbox")
|
|
check("folders to retained message", lambda: key("F6"), SUBJECTS[0], "list item", SUBJECTS[0])
|
|
|
|
check("open selected message", lambda: key("Return"), SUBJECTS[0])
|
|
check("read opened message", lambda: key("ctrl+Home"), f"Body of {SUBJECTS[0]}.")
|
|
check("focus received attachment bar", lambda: key("Tab"),
|
|
"test-attachment.txt", "button", "1 attachment")
|
|
check("expand received attachments", lambda: key("Return"), "test-attachment.txt",
|
|
"list item", "test-attachment", selected=True)
|
|
check("received attachment menu", lambda: key("shift+F10"), "menu", "menu")
|
|
check("attachment open menu item", lambda: key("Down"), "Open", "menu item", "Open")
|
|
check("attachment save menu item", lambda: key("Down"), "Save As", "menu item", "Save As")
|
|
check("open attachment save dialog", lambda: key("Return"), "Save Attachment")
|
|
key("ctrl+l")
|
|
location = expect_focus("text")
|
|
savedAttachment = directory / "saved-attachment.txt"
|
|
key("ctrl+a")
|
|
type_text(str(savedAttachment))
|
|
wait_for(lambda: str(savedAttachment) in location.queryText().getText(0, -1),
|
|
"save attachment location missing")
|
|
check("save received attachment", lambda: key("Return"), "test-attachment",
|
|
"list item", "test-attachment", selected=True)
|
|
wait_for(savedAttachment.exists, "attachment was not saved")
|
|
assert savedAttachment.read_text().strip() == "Synthetic attachment contents."
|
|
check("return to message list", lambda: key("ctrl+w"), SUBJECTS[0],
|
|
"list item", SUBJECTS[0], selected=True)
|
|
check("next message after return", lambda: key("Down"), SUBJECTS[1],
|
|
"list item", SUBJECTS[1], selected=True)
|
|
|
|
for action, chord, prefix in (("reply", "ctrl+r", "Re: "),
|
|
("reply all", "ctrl+shift+r", "Re: "),
|
|
("forward", "ctrl+l", "Fwd: ")):
|
|
replying = action != "forward"
|
|
check(f"open {action}", lambda: key(chord), "Message body" if replying else "To",
|
|
"document web" if replying else "entry",
|
|
"Message body" if replying else "To")
|
|
subject = find_node("entry", "Subject")
|
|
assert subject.queryText().getText(0, -1) == prefix + SUBJECTS[1], (action, "subject")
|
|
report[-1]["subject"] = subject.queryText().getText(0, -1)
|
|
if replying:
|
|
assert find_node("section", "Sender <sender@example.invalid>: press Enter"), (action, "recipient")
|
|
if action == "reply all":
|
|
assert find_node("section", "Other Recipient <other@example.invalid>: press Enter"), (action, "Cc")
|
|
if action == "forward":
|
|
type_text("recipient@example.invalid")
|
|
check("commit forward recipient", lambda: key("Tab"), "recipient@example.invalid",
|
|
"entry", "To")
|
|
check("forward recipient to subject", lambda: key("Tab"), "Subject", "entry", "Subject")
|
|
check("forward subject to body", lambda: key("Tab"), "Message body",
|
|
"document web", "Message body")
|
|
body = expect_focus("document web", "Message body")
|
|
wait_for(lambda: f"Body of {SUBJECTS[1]}." in read_text(body), f"{action}: quoted body missing")
|
|
originalText = read_text(body)
|
|
check(f"{action} character echo", lambda: key("x"), "x", exact=True)
|
|
wait_for(lambda: read_text(body) != originalText, f"{action}: typed text missing")
|
|
discard_composer()
|
|
|
|
check("open composer", lambda: key("ctrl+m"), "To", "entry", "To")
|
|
recipient = expect_focus("entry", "To")
|
|
type_text("recipient@example.invalid")
|
|
wait_for(lambda: recipient.queryText().getText(0, -1) == "recipient@example.invalid",
|
|
"recipient text did not reach Thunderbird")
|
|
check("commit recipient", lambda: key("Tab"), "recipient@example.invalid", "entry", "To")
|
|
subject = check("recipient to subject", lambda: key("Tab"), "Subject", "entry", "Subject")
|
|
offset = len(server.snapshot())
|
|
type_text("Draft subject")
|
|
wait_for(lambda: subject.queryText().getText(0, -1) == "Draft subject", "subject typing missing")
|
|
# Let the title coalescer and SSIP recorder drain after the last key.
|
|
# The check must include the final title, not just the typing interval.
|
|
time.sleep(0.4)
|
|
rows = server.snapshot()[offset:]
|
|
assert any(row["text"] == "D" for row in rows), ("subject character echo missing", rows)
|
|
assert not any(row["text"].startswith("Write:") for row in rows), ("subject title echo", rows)
|
|
report.append({"action": "subject typing without title echo", "speech": rows})
|
|
print("PASS: subject typing without title echo", flush=True)
|
|
offset = len(server.snapshot())
|
|
check("subject character deletion", lambda: key("BackSpace"), "t", exact=True)
|
|
wait_for(lambda: subject.queryText().getText(0, -1) == "Draft subjec", "subject deletion missing")
|
|
time.sleep(0.4)
|
|
rows = server.snapshot()[offset:]
|
|
assert not any(row["text"].startswith("Write:") for row in rows), ("subject deletion title echo", rows)
|
|
type_text("t")
|
|
check("subject to body", lambda: key("Tab"), "Message body", "document web", "Message body")
|
|
body = expect_focus("document web")
|
|
assert body.getState().contains(pyatspi.STATE_EDITABLE), "body is not editable"
|
|
check("body character echo", lambda: key("h"), "h", exact=True)
|
|
check("body character deletion", lambda: key("BackSpace"), "h", exact=True)
|
|
wait_for(lambda: not read_text(body).strip(), "body character deletion missing")
|
|
type_text("Draft first line.")
|
|
key("Return")
|
|
type_text("Draft second line.")
|
|
wait_for(lambda: "Draft first line." in read_text(body) and "Draft second line." in read_text(body),
|
|
"body text did not reach Thunderbird")
|
|
check("body first line", lambda: key("ctrl+Home"), "Draft first line.")
|
|
check("body next line", lambda: key("Down"), "Draft second line.")
|
|
check("select body word", lambda: (key("Home"), key("ctrl+shift+Right")), "Draft selected", exact=True)
|
|
wait_for(lambda: selected_text(body).strip() == "Draft", "native body word selection missing")
|
|
check("delete selected word", lambda: key("BackSpace"), "Draft")
|
|
wait_for(lambda: "Draft second line." not in read_text(body) and "second line." in read_text(body),
|
|
"body selection/deletion did not reach Thunderbird")
|
|
check("body back to subject", lambda: key("shift+Tab"), "Subject", "entry", "Subject")
|
|
check("subject back to body", lambda: key("Tab"), "Message body", "document web", "Message body")
|
|
attachment = directory / "outgoing-attachment.txt"
|
|
attachment.write_text("Unsent synthetic attachment.\n")
|
|
check("open attachment picker", lambda: key("ctrl+shift+a"), "Attach File")
|
|
key("ctrl+l")
|
|
location = expect_focus("text")
|
|
type_text(str(attachment))
|
|
wait_for(lambda: str(attachment) in location.queryText().getText(0, -1),
|
|
"attachment picker location missing")
|
|
check("attach local file", lambda: key("Return"), attachment.name,
|
|
"list item", attachment.stem, selected=True)
|
|
check("clear attachment selection", lambda: key("Escape"), "not selected", exact=True)
|
|
wait_for(lambda: not find_node("list item", attachment.stem).getState().contains(pyatspi.STATE_SELECTED),
|
|
"attachment selection did not clear")
|
|
check("attachment back to body", lambda: key("Escape"), "Message body",
|
|
"document web", "Message body")
|
|
# No send command is used. Termination below leaves the unsent editor
|
|
# in this disposable profile without interacting with a real mailbox.
|
|
print(f"PASS: Thunderbird {'HTML' if html else 'plain-text'} local mail workflow", flush=True)
|
|
finally:
|
|
tree = []
|
|
try:
|
|
try:
|
|
for app in pyatspi.Registry.getDesktop(0):
|
|
if app.name.lower() in APPLICATIONS:
|
|
for node in walk(app):
|
|
tree.append({"role": node.getRoleName(), "name": node.name,
|
|
"showing": node.getState().contains(pyatspi.STATE_SHOWING),
|
|
"focused": node.getState().contains(pyatspi.STATE_FOCUSED),
|
|
"selected": node.getState().contains(pyatspi.STATE_SELECTED)})
|
|
except Exception as error:
|
|
tree.append({"error": str(error)})
|
|
(directory / "tree.json").write_text(json.dumps(tree, indent=2))
|
|
(directory / "report.json").write_text(json.dumps(report, indent=2))
|
|
(directory / "speech.json").write_text(json.dumps(server.snapshot(), indent=2))
|
|
finally:
|
|
for process in reversed(processes):
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
server.close()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--format", choices=("html", "plain", "both"), 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.format == "html")
|
|
return
|
|
formats = ("html", "plain") if args.format == "both" else (args.format,)
|
|
for composeFormat in formats:
|
|
directory = Path(tempfile.mkdtemp(prefix=f"cthulhu-thunderbird-{composeFormat}-"))
|
|
print(f"Thunderbird {composeFormat} evidence: {directory}", flush=True)
|
|
environment = dict(os.environ, XDG_SESSION_TYPE="x11", CTHULHU_THUNDERBIRD_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()), "--format", composeFormat, "--inside"],
|
|
env=environment, stderr=log, check=True, timeout=120)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|