163 lines
9.4 KiB
Python
163 lines
9.4 KiB
Python
#!/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()
|