Initial support for application specific overrides.

tests/native_application_overrides.py

@@ -0,0 +1,184 @@
+#!/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, ["<Username> 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, ["<Username> old", "<Username> 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", "Save unavailable")
+                assert remote("SETTING SAVE") == "Save unavailable."
+                assert remote("SETTING SAVEAS " + str(root / "other.toml")) == "Save unavailable."
+                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", "Ninjam application override selected")
+                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, save guards and 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()