#!/usr/bin/env python3 """Private X11 application-override fixture using source-derived Ninjam output. Run: xvfb-run -a env FENRIR_PRIVATE_X11=1 FENRIR_TEST_BINARY=/tmp/fenrir-test \ python3 tests/native_application_overrides.py Uses fake speech and a private PTY; does not establish audible/physical acceptance. """ import json import os from pathlib import Path import signal import socket import subprocess import sys import tempfile import tomllib import tty def draw(selected, messages): output = "\x1b[0m\x1b[2J\x1b[HLOCAL\x1b[2;1H" field = 0 for char in " master: [ ]mute [0.0dB center] | metronome: [ ]mute [0.0dB center]": if char == "]": output += "\x1b[0m" output += char if char == "[": output += "\x1b[0;30;47m" if selected == field else "\x1b[1m" field += 1 output += "\x1b[17;1HCHAT" for row, message in enumerate(messages[-4:], 21 - min(len(messages), 4)): output += f"\x1b[{row + 1};1H{message}" return (output + "\x1b[23;1H#####\x1b[24;1H[QUIT NINJAM] : fixture.invalid : " "120.0BPM 16BPI\x1b[24;1H").encode() def fixture_shell(): root = Path(os.environ["FENRIR_FIXTURE_DIR"]) tty.setraw(0) (root / "shell-pid").write_text(str(os.getpid())) os.write(1, draw(0, [" old"])) with (root / "input").open("ab", buffering=0) as record: while True: data = os.read(0, 4096) if not data: return record.write(data) if b"q" in data: os.write(1, b"\x1b[0m\x1b[2J\x1b[H[Username@host directory] $ ") elif data in (b"\x1b[C", b"\x1bOC"): os.write(1, draw(1, [" old", " incoming"])) def main(): from native_terminal import KeyboardFixture, SpeechServer, wait_for if os.environ.get("FENRIR_PRIVATE_X11") != "1" or not os.environ.get("DISPLAY"): raise SystemExit("Use xvfb-run -a env FENRIR_PRIVATE_X11=1") repo = Path(__file__).resolve().parents[1] script = Path(__file__).resolve() binary = os.environ.get("FENRIR_TEST_BINARY", str(repo / "fenrir")) for layout in ("desktop", "laptop"): with tempfile.TemporaryDirectory(prefix="fenrir-overrides-", dir="/tmp") as directory: root = Path(directory) (root / "keyboard").mkdir() for name in ("custom", "laptop"): source = "desktop" if name == "custom" else "laptop" (root / "keyboard" / (name + ".toml")).write_bytes( (repo / "config/keyboard" / (source + ".toml")).read_bytes()) (root / "mode").write_text("laptop" if layout == "laptop" else "fixture") (root / "home").mkdir() (root / "settings.toml").write_text( 'version=3\n[presentation.sound]\nenabled=false\n' '[presentation.speech]\nrate=0\n' f'[remote]\nsocket_file="{root / "remote.sock"}"\n') shell = root / "shell" shell.write_text(f'#!{sys.executable}\nimport runpy, sys\nsys.argv=[{str(script)!r}, "--shell"]\n' f'runpy.run_path({str(script)!r}, run_name="__main__")\n') shell.chmod(0o700) server = SpeechServer(root / "speech.sock") keyboard = KeyboardFixture() env = dict(os.environ, HOME=str(root / "home"), TMPDIR=str(root), FENRIR_FIXTURE_DIR=str(root), FENRIR_TEST_BINARY=binary, SPEECHD_ADDRESS="unix_socket:" + str(root / "speech.sock"), LOGNAME="Username", USER="Username", LC_ALL="C", LANGUAGE="C") process = subprocess.Popen(["xterm", "-geometry", "100x24", "-e", sys.executable, str(repo / "tests/native_terminal.py"), "--outer", str(root)], env=env) state = None try: wait_for(lambda: (root / "started").exists(), "reader startup") state = json.loads((root / "started").read_text()) subprocess.run(["xdotool", "windowfocus", "--sync", str(state["window"])], check=True) def spoken(text, start): wait_for(lambda: any(text in item for item in server.events[start:]), f"speech {text!r}; recent={server.events[-12:]}") def action(chord, expected): start = len(server.events) keyboard.key(chord) spoken(expected, start) def remote(text): with socket.socket(socket.AF_UNIX) as connection: connection.settimeout(3) connection.connect(str(root / "remote.sock")) connection.sendall((text + "\n").encode()) connection.shutdown(socket.SHUT_WR) return connection.recv(4096).decode().strip() wait_for(lambda: (root / "remote.sock").exists(), "remote ready") action("Insert+Up", "10") action("Insert+a", "Application Default") action("Down", "NetworkManager") action("Down", "Ninjam") action("Escape", "Applications closed") action("Insert+a", "Application Default") action("Down", "NetworkManager") action("Down", "Ninjam") action("Return", "Master unmuted") original = (root / "settings.toml").read_bytes() action("Insert+Up", "20") action("Insert+ctrl+s", "Ninjam application settings saved") profile = root / "applications" / "ninjam.toml" assert tomllib.loads(profile.read_text())["presentation"]["speech"]["rate"] == 20 assert remote("SETTING SAVE") == "" assert remote("SETTING SAVEAS " + str(root / "other.toml")) == "SAVEAS is unavailable in application settings" action("Insert+f", "Ninjam application settings") action("Escape", "Ninjam application settings closed") assert not (root / "other.toml").exists() assert (root / "settings.toml").read_bytes() == original # Start detached review, then deliver actual XTest application # navigation. The fixture redraw includes focus and chat together. action("Insert+i" if layout == "laptop" else "KP_Up", "QUIT NINJAM") start = len(server.events) keyboard.key("Right") spoken("Master Volume and pan", start) spoken("incoming", start) action("Insert+1", "incoming") action("Insert+2", "Master Volume and pan") action("Insert+3", "120.0BPM") action("Insert+a", "Application Ninjam") action("Down", "Default") action("Return", "Default settings restored") action("Insert+space", "10") action("Insert+a", "Application Default") action("Down", "NetworkManager") action("Down", "Ninjam") action("Return", "Master Volume and pan") action("Insert+space", "20") action("q", "Default settings restored") action("Insert+ctrl+s", "Settings saved") saved = tomllib.loads((root / "settings.toml").read_text()) assert saved["presentation"]["speech"]["rate"] == 10, saved assert (root / "input").read_bytes() in (b"\x1b[Cq", b"\x1bOCq"), (root / "input").read_bytes() keyboard.key("Insert+q") wait_for(lambda: (root / "finished").exists(), "reader exit") finished = json.loads((root / "finished").read_text()) assert finished == {"status": 0, "restored": True, "flagsRestored": True}, finished assert not server.errors, server.errors print(f"PASS: {layout} application menu, cancellation, semantic navigation/chat, region reads, application saves/reloads and default restoration") except BaseException: print("SSIP events:", server.events[-25:], file=sys.stderr) for name in ("reader-errors", "timing.log"): if (root / name).exists(): print(name + ":", (root / name).read_text()[-4000:], file=sys.stderr) raise finally: if state: try: os.kill(state["pid"], signal.SIGTERM) except ProcessLookupError: pass process.terminate() try: process.wait(timeout=3) except subprocess.TimeoutExpired: process.kill() process.wait() server.close() keyboard.close() if __name__ == "__main__": if len(sys.argv) > 1 and sys.argv[1] == "--shell": fixture_shell() else: main()