#!/usr/bin/env python3 """Private-X11 driver-dependent settings and hardware preview safety trial.""" import fcntl import json import os from pathlib import Path import select import subprocess import sys import tempfile import tomllib from native_terminal import KeyboardFixture, SpeechServer, wait_for def main(): if not os.environ.get("DISPLAY") or os.environ.get("FENRIR_PRIVATE_X11") != "1": raise SystemExit("Use xvfb-run -a env FENRIR_PRIVATE_X11=1 python3 tests/native_hardware_settings.py") repository = Path(__file__).resolve().parents[1] fixture = str(repository / "tests/native_terminal.py") binary = os.environ.get("FENRIR_TEST_BINARY", str(repository / "fenrir")) with tempfile.TemporaryDirectory(prefix="fenrir-hardware-settings-", dir="/tmp") as directory: suiteRoot = Path(directory) libraryRoot = suiteRoot / "usb" libraryRoot.mkdir() subprocess.run(["cc", "-shared", "-fPIC", "-o", str(libraryRoot / "libusb-1.0.so.0"), str(repository / "internal/speech/testdata/usb_fixture.c")], check=True) keyboard = KeyboardFixture() try: for layout in ("desktop", "laptop"): root = suiteRoot / layout root.mkdir() (root / "home").mkdir() (root / "keyboard").mkdir() layoutName = "laptop.toml" if layout == "laptop" else "custom.toml" (root / "keyboard" / layoutName).write_bytes((repository / "config/keyboard" / (layout + ".toml")).read_bytes()) if layout == "laptop": (root / "mode").write_text("laptop") shell = root / "shell" shell.write_text(f'#!{sys.executable}\nimport runpy, sys\nsys.argv = [{fixture!r}, "--fixture-shell"]\nrunpy.run_path({fixture!r}, run_name="__main__")\n') shell.chmod(0o700) settings = root / "settings.toml" settings.write_text('version=3\n[presentation.speech]\nrate=0\n' '[presentation.sound]\nenabled=false\n' '[reading]\npunctuation_level="none"\nspellcheck=false\n' f'[remote]\nsocket_file="{root / "remote.sock"}"\n') original = settings.read_bytes() master, slave = os.openpty() fcntl.fcntl(master, fcntl.F_SETFL, os.O_NONBLOCK) server = SpeechServer(root / "speech.sock") environment = dict(os.environ, FENRIR_TEST_BINARY=binary, FENRIR_FIXTURE_DIR=str(root), FENRIR_TEST_READER_TIMEOUT="120", HOME=str(root / "home"), TMPDIR=str(root), LOGNAME="Username", USER="Username", SPEECHD_ADDRESS="unix_socket:" + str(root / "speech.sock"), LD_LIBRARY_PATH=str(libraryRoot), USB_FIXTURE_MODE="ok", USB_FIXTURE_LOG=str(root / "usb.log")) terminal = subprocess.Popen(["xterm", "-geometry", "80x25", "-e", sys.executable, fixture, "--outer", str(root)], env=environment) captured = bytearray() def spoken(text, start=0): wait_for(lambda: any(text in item for item in server.events[start:]), f"speech {text!r}; recent={server.events[-8:]}") def focus(label): for _ in range(40): start = len(server.events) keyboard.key("Tab") text = wait_for(lambda: next((item for item in reversed(server.events[start:]) if item != "CANCEL"), None), "next control") if label in text: return text raise AssertionError("Control not reachable: " + label) def output(driver): if driver == "dectalk-usb": log = root / "usb.log" return b"".join(bytes.fromhex(line.partition(":")[2]) for line in log.read_text().splitlines() if line.startswith("write-02:")) if log.exists() else b"" while select.select([master], [], [], 0)[0]: captured.extend(os.read(master, 65536)) return bytes(captured) try: wait_for(lambda: (root / "started").exists(), "reader startup") state = json.loads((root / "started").read_text()) subprocess.run(["xdotool", "windowfocus", "--sync", state["window"]], check=True) spoken("prompt") for index, (driver, driverLabel) in enumerate((("dectalk", "DECtalk"), ("dectalk-usb", "DECtalk USB"), ("litetalk", "LiteTalk"), ("doubletalk", "DoubleTalk LT"), ("tripletalk", "TripleTalk")), 1): start = len(server.events) keyboard.key("Insert+f") spoken("Fenrir Menu. Presentation tab", start) focus("Driver combo box Speech Dispatcher") keyboard.key("Home", *("Down",) * index) start = len(server.events) keyboard.key("Tab") spoken("Pitch" if driver == "dectalk-usb" else "Serial port entry", start) if driver != "dectalk-usb": keyboard.key("ctrl+a") subprocess.run(["xdotool", "type", "--delay", "1", os.ttyname(slave)], check=True) start = len(server.events) keyboard.key("Tab") spoken("Baud rate combo box default", start) keyboard.key("Home", *("Down",) * 5) focus("Test synthesizer button") if driver == "dectalk": # A missing port must report through the still-working # Speech Dispatcher client, without saving anything. focus("Serial port entry") keyboard.key("ctrl+a") subprocess.run(["xdotool", "type", "--delay", "1", str(root / "missing-port")], check=True) focus("Test synthesizer button") start = len(server.events) keyboard.key("Return") spoken("Voice test failed", start) assert settings.read_bytes() == original focus("Serial port entry") keyboard.key("ctrl+a") subprocess.run(["xdotool", "type", "--delay", "1", os.ttyname(slave)], check=True) focus("Test synthesizer button") before = len(output(driver)) keyboard.key("Return") phrase = b"Voice test: The quick brown fox jumps over the lazy dog." wait_for(lambda: phrase in output(driver)[before:], f"{driver} temporary preview bytes") assert settings.read_bytes() == original, "test wrote settings" assert not (root / "input").exists() or not (root / "input").read_bytes(), "form keys leaked" if driver == "tripletalk": focus("OK button") start = len(server.events) keyboard.key("Return") spoken("Fenrir Menu closed", start) saved = tomllib.loads(settings.read_text())["presentation"]["speech"] assert saved["driver"] == driver and saved["hardware_device"] == os.ttyname(slave) and saved["hardware_baud_rate"] == 19200, saved else: focus("Driver combo box " + driverLabel) keyboard.key("Home") start = len(server.events) keyboard.key("Tab") spoken("Speech Dispatcher module combo box", start) keyboard.key("Shift_L+Tab") keyboard.key("Escape") assert settings.read_bytes() == original, "cancel wrote settings" print(f"PASS: {driver}/{layout}: dynamic fields, temporary hardware test, active speech preserved and save/cancel", flush=True) start = len(server.events) keyboard.key("Insert+i" if layout == "laptop" else "KP_Up") spoken("prompt", start) keyboard.key("Insert+q") wait_for(lambda: (root / "finished").exists(), "reader shutdown") assert json.loads((root / "finished").read_text()) == {"status": 0, "restored": True, "flagsRestored": True} assert not server.errors, server.errors finally: terminal.terminate() try: terminal.wait(timeout=3) except subprocess.TimeoutExpired: terminal.kill() terminal.wait() server.close() os.close(master) os.close(slave) finally: keyboard.close() if __name__ == "__main__": main()