"""Real reader/preferences/GTK flow against a private, completion-aware SSIP server.""" import json import os from pathlib import Path import socket import subprocess import tempfile import threading import time from native_x11 import wait_for class PreviewSpeechServer: def __init__(self, path, defaultsSupported=True): self.listener = socket.socket(socket.AF_UNIX) self.listener.bind(str(path)) self.listener.listen() self.lock = threading.RLock() self.clients = {} self.spoken = [] self.defaultsSupported = defaultsSupported threading.Thread(target=self.accept_clients, daemon=True).start() def accept_clients(self): while True: try: connection, _ = self.listener.accept() except OSError: return with self.lock: clientId = len(self.clients) + 1 client = {"socket": connection, "lock": threading.RLock(), "notifications": False, "module": "main", "voice": "", "active": None, "settings": {"RATE": 20, "PITCH": -10, "VOLUME": 100}} self.clients[clientId] = client threading.Thread(target=self.serve_client, args=(clientId, client), daemon=True).start() def reply(self, client, text): with client["lock"]: try: client["socket"].sendall(text.encode()) except OSError: pass def finish(self, clientId, client, record, canceled=False): with self.lock: if client["active"] is not record: return client["active"] = None record["finished"], record["canceled"] = True, canceled if client["notifications"]: code = 703 if canceled else 702 self.reply(client, f"{code}-1\r\n{code}-{clientId}\r\n{code} DONE\r\n") def serve_client(self, clientId, client): try: with client["socket"].makefile("rb") as stream: for raw in stream: command = raw.decode().strip() response = "200 OK\r\n" if command == "HISTORY GET CLIENT_ID": response = f"245-{clientId}\r\n245 OK\r\n" elif command == "LIST OUTPUT_MODULES": response = "200-echo-a\r\n200-echo-b\r\n200-main\r\n200 OK\r\n" elif command == "LIST SYNTHESIS_VOICES": voice = {"echo-a": "Echo A", "echo-b": "Echo B", "main": "Main"}[client["module"]] response = f"200-{voice}\ten\tnone\r\n200 OK\r\n" elif command.startswith("GET "): field = command.removeprefix("GET ") response = (f"200-{client['settings'][field]}\r\n200 OK\r\n" if self.defaultsSupported and field in client["settings"] else "500 UNSUPPORTED\r\n") elif any(command.startswith("SET SELF " + field + " ") for field in client["settings"]): _, _, field, value = command.split() client["settings"][field] = int(value) elif command.startswith("SET SELF OUTPUT_MODULE "): client["module"] = command.removeprefix("SET SELF OUTPUT_MODULE ") elif command.startswith("SET SELF SYNTHESIS_VOICE "): client["voice"] = command.removeprefix("SET SELF SYNTHESIS_VOICE ") elif command.startswith("SET SELF NOTIFICATION ALL "): client["notifications"] = command.endswith(" on") elif command.startswith("CANCEL "): with self.lock: targetId = int(command.split()[1]) target = self.clients[targetId] if target["active"] is not None: self.finish(targetId, target, target["active"], canceled=True) elif command == "SPEAK" or command.startswith("CHAR "): parts = [] if command == "SPEAK": for line in stream: if line == b".\r\n": break parts.append(line.decode().rstrip("\r\n")) else: parts.append(command.removeprefix("CHAR ")) record = {"text": "\n".join(parts), "module": client["module"], "voice": client["voice"], "finished": False, "canceled": False, "settings": dict(client["settings"])} with self.lock: self.spoken.append(record) client["active"] = record self.reply(client, "230 OK\r\n200 QUEUED\r\n" if command == "SPEAK" else "225 QUEUED\r\n") if client["notifications"]: self.reply(client, f"701-1\r\n701-{clientId}\r\n701 BEGIN\r\n") timer = threading.Timer(0.25, self.finish, args=(clientId, client, record)) timer.daemon = True timer.start() continue self.reply(client, response) except (OSError, ValueError): pass def snapshot(self): with self.lock: return [dict(record) for record in self.spoken] def close(self): self.listener.close() for client in self.clients.values(): client["socket"].close() def check_radio_buttons(controls, tabs, server, readerLog): import pyatspi def radio_cues(): records = [json.loads(line) for line in readerLog.read_text().splitlines() if line.startswith("{")] return [record["cue"] for record in records if record.get("message") == "playing object sound" and record.get("cue", "").startswith("radiobutton_")] assert tabs.querySelection().selectChild(0) assert controls["Desktop"].queryComponent().grabFocus() wait_for(lambda: radio_cues(), "initial radio sound did not play") time.sleep(0.35) for keyName, target in (("Right", "Laptop"), ("Left", "Desktop"), ("Right", "Laptop")): cueOffset = len(radio_cues()) speechOffset = len(server.snapshot()) subprocess.run(["xdotool", "key", "--clearmodifiers", keyName], check=True) wait_for(lambda: controls[target].getState().contains(pyatspi.STATE_CHECKED), target + " was not selected") wait_for(lambda: len(radio_cues()) > cueOffset, target + " selection sound did not play") # Cover the deferred focus cue and duplicate focus event after checking. time.sleep(0.35) cues = radio_cues()[cueOffset:] assert cues == ["radiobutton_checked"], (keyName, target, cues) spoken = [record["text"] for record in server.snapshot()[speechOffset:]] assert any(target in text for text in spoken), target + " label was lost" assert not any("radio button" in text for text in spoken), (target, spoken) records = [json.loads(line) for line in readerLog.read_text().splitlines() if line.startswith("{")] errors = [record for record in records if record.get("message") == "play sound" and record.get("cue", "").startswith("radiobutton_") and record.get("error") != "context canceled"] assert not errors, errors def check_flat_review(tabs, server, readerLog): def records(): return [json.loads(line) for line in readerLog.read_text().splitlines() if line.startswith("{")] def key(*keys): subprocess.run(["xdotool", "key", "--clearmodifiers", *keys], check=True) assert tabs.querySelection().selectChild(0) key("ctrl+Page_Down", "ctrl+Page_Up") wait_for(lambda: any(record["text"] == "General" for record in server.snapshot()) and server.snapshot()[-1]["finished"], "General tab was not announced") time.sleep(0.4) for command in ("KP_Up", "KP_Page_Up", "KP_Right", "KP_Page_Down", "KP_End", "KP_Home"): logOffset = len(records()) speechOffset = len(server.snapshot()) key(command) def review_delivered(): return any(record.get("message") == "delivering accessible presentation" and record.get("cause") == "flat-review" for record in records()[logOffset:]) wait_for(review_delivered, "flat review did not speak: " + command) wait_for(lambda: len(server.snapshot()) > speechOffset and server.snapshot()[-1]["finished"], "review speech did not finish: " + command) # Include the focus replay after the key grab has been released. time.sleep(0.4) deliveries = [record for record in records()[logOffset:] if record.get("message") == "delivering accessible presentation"] assert deliveries and all(record.get("cause") == "flat-review" for record in deliveries), deliveries assert not any(record["canceled"] for record in server.snapshot()[speechOffset:]) assert tabs.querySelection().getSelectedChild(0).name == "General" speechOffset = len(server.snapshot()) nextTab = tabs[1].name key("Right") wait_for(lambda: any(record["text"] == nextTab for record in server.snapshot()[speechOffset:]) and server.snapshot()[-1]["finished"], "new tab focus was suppressed") speechOffset = len(server.snapshot()) key("Tab") wait_for(lambda: len(server.snapshot()) > speechOffset and server.snapshot()[-1]["finished"], "Tab into the page did not speak") speechOffset = len(server.snapshot()) key("shift+Tab") wait_for(lambda: any(record["text"] == nextTab for record in server.snapshot()[speechOffset:]) and server.snapshot()[-1]["finished"], "return to tab header was suppressed") def check_interruption_summary(controls, tabs, server): import pyatspi def key(*keys): subprocess.run(["xdotool", "key", "--clearmodifiers", *keys], check=True) assert tabs.querySelection().selectChild(2) summary = controls["Defined interruption keys"] assert summary.queryText().getText(0, -1) == "Tab, Enter" assert not summary.getState().contains(pyatspi.STATE_EDITABLE) assert controls["Record interruption keys"].queryComponent().grabFocus() offset = len(server.snapshot()) key("Tab") wait_for(lambda: any(record["text"] == "Tab, Enter" and record["finished"] for record in server.snapshot()[offset:]), "read-only summary contents were not spoken on focus") assert any(record["text"] == "Defined interruption keys" for record in server.snapshot()[offset:]) key("Home") for keyName, character, caret in (("Right", "a", 1), ("Left", "T", 0)): offset = len(server.snapshot()) key(keyName) wait_for(lambda: summary.queryText().caretOffset == caret, "summary caret did not move for " + keyName) wait_for(lambda: any(record["text"] == character and record["finished"] for record in server.snapshot()[offset:]), "summary character was not spoken for " + keyName) key("shift+Tab") wait_for(lambda: controls["Record interruption keys"].getState().contains(pyatspi.STATE_FOCUSED), "read-only summary trapped Shift+Tab") def check_runtime(radioOnly=False, flatReviewOnly=False, interruptionOnly=False, noConfigOnly=False): import pyatspi directory = Path(tempfile.mkdtemp(prefix="cthulhu-echo-runtime-")) binary = os.environ["CTHULHU_BINARY"] wmConfig = directory / "i3.config" wmConfig.write_text(f"font pango:monospace 10\nipc-socket {directory / 'i3.sock'}\n") server = PreviewSpeechServer(directory / "speech.sock", defaultsSupported=os.environ.get("CTHULHU_PREFS_DEFAULTS_UNSUPPORTED") != "1") wm = subprocess.Popen(["i3", "-c", str(wmConfig)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) reader = None try: wait_for(lambda: "0x" in subprocess.check_output(["xprop", "-root", "_NET_SUPPORTING_WM_CHECK"], text=True), "private WM missing") configPath = directory / "settings.toml" if not noConfigOnly: configPath.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 custom-voice = true custom-speech-server = true speech-server = "echo-a" [profiles.default.key-echo.voice] rate = 50 average-pitch = 5 gain = 5 [profiles.default.key-echo.voice.family] name = "Echo A" lang = "en" ''') if interruptionOnly: with configPath.open("a") as settingsFile: settingsFile.write('\n[profiles.default.speech.interruption]\nmode = "custom"\nkeys = ["Tab", "Return"]\n') if radioOnly: configPath.write_text(configPath.read_text().replace( "[profiles.default.sound]\nenabled = false", '[profiles.default.sound]\nenabled = true\nsink = "auto"\nplay-for-state = true\nrole-presentation-strategy = "sound_only"')) readerEnv = os.environ.copy() if noConfigOnly: # Preferences checks must not launch Wine Access in live prefixes. readerEnv["CTHULHU_WINE_ACCESS_HELPER"] = "/usr/bin/true" if radioOnly or noConfigOnly: import gi gi.require_version("Gst", "1.0") from gi.repository import Gst Gst.init(None) assert Gst.ElementFactory.find("fakeaudiosink"), "silent GStreamer audio sink is unavailable" # Only the child reader sees these ranks. Auto audio can choose # the fake sink; every real audio sink is excluded from discovery. ranks = [] for factory in Gst.Registry.get().get_feature_list(Gst.ElementFactory): kind = factory.get_metadata("klass") or "" if "Audio" in kind and "Sink" in kind: rank = "MAX" if factory.get_name() == "fakeaudiosink" else "0" ranks.append(factory.get_name() + ":" + rank) readerEnv["GST_PLUGIN_FEATURE_RANK"] = ",".join(ranks) reader = subprocess.Popen([binary, "--config", str(configPath), "--settings", "--no-tentacles", "--speech-socket", str(directory / "speech.sock"), "--debug-file", str(directory / "reader.log")], env=readerEnv, stdout=subprocess.DEVNULL, stderr=(directory / "stderr.log").open("w")) def find_window(): for application in pyatspi.Registry.getDesktop(0): for window in application: if window and window.name == "Cthulhu Preferences": return window window = wait_for(find_window, "reader's preferences did not open") def walk(node): yield node for child in node: if child is not None: yield from walk(child) controls = {node.name: node for node in walk(window) if node.name} xid = subprocess.check_output(["xdotool", "search", "--onlyvisible", "--name", "^Cthulhu Preferences$"], text=True).splitlines()[0] subprocess.run(["xdotool", "windowactivate", "--sync", xid], check=True) tabs = next(node for node in walk(window) if node.getRole() == pyatspi.ROLE_PAGE_TAB_LIST) if noConfigOnly: from native_speech_defaults import check_no_config check_no_config(window, tabs, server, configPath) print(f"PASS: missing config, editable speech sliders, server defaults/fallback, inheritance, Apply/save/reopen; evidence: {directory}") return if interruptionOnly: check_interruption_summary(controls, tabs, server) print(f"PASS: real reader speaks read-only interruption summary on focus and character arrows; Tab traversal preserved; evidence: {directory}") return if flatReviewOnly: check_flat_review(tabs, server, directory / "reader.log") print(f"PASS: notebook flat review has no repeated tab speech; tab changes and focus returns speak; evidence: {directory}") return if radioOnly: check_radio_buttons(controls, tabs, server, directory / "reader.log") print(f"PASS: real GTK Desktop/Laptop arrows play only the final selected sound and retain labels; evidence: {directory}") return assert tabs.querySelection().selectChild(4) assert controls["Use custom echo voice"].queryComponent().grabFocus() wait_for(lambda: any(record["text"] == "checked" and record["finished"] for record in server.snapshot()), "initial checkbox announcement did not finish") time.sleep(0.3) def key(*keys): subprocess.run(["xdotool", "key", "--clearmodifiers", *keys], check=True) def check_output(keys, expected): offset = len(server.snapshot()) key(*keys) def completed(): records = server.snapshot()[offset:] matching = [record for record in records if record["text"] in {text for text, _, _ in expected}] return len(matching) >= len(expected) and all(record["finished"] for record in matching) wait_for(completed, f"preview did not complete: {keys}") # The original delayed menu-item event followed its preview by up # to 160 ms. Require quiescence beyond that delay, then inspect all # speech requests, including duplicates in the main voice. time.sleep(0.25) records = server.snapshot()[offset:] got = [(record["text"], record["module"], record["voice"]) for record in records] assert got == expected, (keys, got, expected) assert not any(record["canceled"] for record in records), (keys, records) check_output(("Tab",), [("Key echo speech synthesizer", "main", ""), ("echo-a", "echo-a", "Echo A")]) check_output(("Down",), [("echo-b", "echo-b", "")]) check_output(("Tab",), [("Key echo person", "main", ""), ("Default", "echo-b", "")]) check_output(("Down",), [("Echo B (en)", "echo-b", "Echo B")]) for control in ("rate", "pitch", "volume"): value = "50.0" if control == "rate" else "5.0" check_output(("Tab",), [("Key echo " + control, "main", ""), (value, "echo-b", "Echo B")]) for control in ("pitch", "rate", "person", "synthesizer"): value = {"pitch": "5.0", "rate": "50.0", "person": "Echo B (en)", "synthesizer": "echo-b"}[control] label = "Key echo speech synthesizer" if control == "synthesizer" else "Key echo " + control check_output(("shift+Tab",), [(label, "main", ""), (value, "echo-b", "Echo B")]) print(f"PASS: real reader + GTK + SSIP focus/selection previews, voice completion, no duplicate menu-item speech; evidence: {directory}") finally: if reader is not None: reader.terminate() try: reader.wait(timeout=5) except subprocess.TimeoutExpired: reader.kill() reader.wait(timeout=5) wm.terminate() wm.wait(timeout=5) server.close() (directory / "speech.json").write_text(json.dumps(server.snapshot(), indent=2))