190 lines
9.0 KiB
Python
190 lines
9.0 KiB
Python
#!/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", "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()
|