320 lines
17 KiB
Python
320 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""Private X11 clipboard/PTY fixture; never opens live speech or console input.
|
|
|
|
Build fenrir first, then run:
|
|
xvfb-run -a env FENRIR_PRIVATE_X11=1 python3 tests/native_clipboard.py
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
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("Run this fixture under xvfb-run with FENRIR_PRIVATE_X11=1")
|
|
repo = Path(__file__).resolve().parents[1]
|
|
binary = str(repo / "fenrir")
|
|
terminalScript = str(repo / "tests/native_terminal.py")
|
|
processes = []
|
|
|
|
def owner(data, target="UTF8_STRING"):
|
|
process = subprocess.Popen(
|
|
["xclip", "-selection", "clipboard", "-in", "-quiet", "-target", target],
|
|
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
processes.append(process)
|
|
process.stdin.write(data)
|
|
process.stdin.close()
|
|
return process
|
|
|
|
def x_text():
|
|
result = subprocess.run(["xclip", "-selection", "clipboard", "-out"], capture_output=True, timeout=2)
|
|
return result.stdout.decode() if result.returncode == 0 else ""
|
|
|
|
with tempfile.TemporaryDirectory(prefix="fenrir-clipboard-", dir="/tmp") as directory:
|
|
root = Path(directory)
|
|
server = SpeechServer(root / "speech.sock")
|
|
keyboard = KeyboardFixture()
|
|
try:
|
|
keyboardDir = root / "keyboard"
|
|
keyboardDir.mkdir()
|
|
for filename in ["desktop.toml", "laptop.toml"]:
|
|
(keyboardDir / filename).write_bytes((repo / "config/keyboard" / filename).read_bytes())
|
|
(keyboardDir / "custom.toml").write_bytes((keyboardDir / "desktop.toml").read_bytes())
|
|
clipFile = root / "clipboard.txt"
|
|
(root / "settings.toml").write_text(
|
|
'version=3\n[presentation.sound]\nenabled=false\n'
|
|
f'[clipboard]\nsync=false\nexport_path="{clipFile}"\n'
|
|
f'[remote]\nsocket_file="{root / "remote.sock"}"\n'
|
|
f'[general]\nscript_path="{root / "scripts"}"\n'
|
|
)
|
|
shell = root / "shell"
|
|
shell.write_text(
|
|
f'#!{sys.executable}\nimport runpy, sys\nsys.argv = [{terminalScript!r}, "--fixture-shell"]\n'
|
|
f'runpy.run_path({terminalScript!r}, run_name="__main__")\n'
|
|
)
|
|
shell.chmod(0o700)
|
|
(root / "home").mkdir()
|
|
environment = dict(os.environ, SHELL=str(shell), FENRIR_FIXTURE_DIR=str(root), HOME=str(root / "home"),
|
|
TMPDIR=str(root),
|
|
SPEECHD_ADDRESS="unix_socket:" + str(root / "speech.sock"))
|
|
|
|
def key(*names):
|
|
keyboard.key(*names)
|
|
|
|
def spoken(text, start):
|
|
wait_for(lambda: text in server.events[start:], f"speech {text[:80]!r}: {[item[:80] for item in server.events[-10:]]}")
|
|
|
|
def action(chord, text):
|
|
start = len(server.events)
|
|
key(chord)
|
|
spoken(text, start)
|
|
|
|
def start(layout):
|
|
for filename in ["started", "finished", "size", "input"]:
|
|
(root / filename).unlink(missing_ok=True)
|
|
(root / "mode").write_text("laptop" if layout == "laptop" else "fixture")
|
|
process = subprocess.Popen(["xterm", "-geometry", "80x25", "-T", "ClipboardFixture", "-e",
|
|
sys.executable, terminalScript, "--outer", str(root)], env=environment)
|
|
processes.append(process)
|
|
wait_for(lambda: (root / "started").exists(), "reader launch")
|
|
state = json.loads((root / "started").read_text())
|
|
subprocess.run(["xdotool", "windowfocus", "--sync", state["window"]], check=True)
|
|
wait_for(lambda: (root / "size").exists(), "fixture shell")
|
|
return state
|
|
|
|
def finish():
|
|
key("Insert+q")
|
|
wait_for(lambda: (root / "finished").exists(), "reader quit")
|
|
assert json.loads((root / "finished").read_text()) == {"status": 0, "restored": True, "flagsRestored": True}
|
|
|
|
def emoji_trial():
|
|
baselineInput = (root / "input").read_bytes()
|
|
clipFile.write_text("original clipboard")
|
|
action("Insert+F5", "original clipboard")
|
|
action("Insert+Control_L+e", "Emoji Menu. All tab")
|
|
key("Right")
|
|
key("Tab", "Down", "Right", "Up", "Left", "Shift_L+Tab")
|
|
action("Right", "Nature tab")
|
|
key("Escape")
|
|
action("Insert+F6", "clipboard exported to file")
|
|
assert clipFile.read_text() == "original clipboard", "category browsing changed clipboard"
|
|
|
|
action("Insert+Control_L+e", "Emoji Menu. All tab")
|
|
key("Tab", "z", "z", "z")
|
|
wait_for(lambda: any("No matches" in text for text in server.events[-8:]), "empty emoji results")
|
|
key("Return")
|
|
# Still captured: clear the filter and select the first J result.
|
|
key("Control_L+BackSpace", "j", "Return")
|
|
wait_for(lambda: "Emoji Menu closed" in server.events[-8:], "emoji selection closes menu")
|
|
action("Insert+F6", "clipboard exported to file")
|
|
assert clipFile.read_text() == "🎃", clipFile.read_text()
|
|
action("Insert+Next", "original clipboard")
|
|
action("Insert+Prior", "🎃")
|
|
|
|
# A ZWJ sequence must survive selection, export and PTY paste.
|
|
action("Insert+Control_L+e", "Emoji Menu. All tab")
|
|
key("Tab", "b", "l", "a", "c", "k", "space", "c", "a", "t", "Return")
|
|
wait_for(lambda: "Emoji Menu closed" in server.events[-8:], "ZWJ emoji selection")
|
|
action("Insert+F6", "clipboard exported to file")
|
|
assert clipFile.read_text() == "🐈⬛", clipFile.read_text()
|
|
assert (root / "input").read_bytes() == baselineInput, "emoji menu keys reached PTY"
|
|
action("Insert+v", "paste clipboard")
|
|
expectedInput = baselineInput + "🐈⬛".encode()
|
|
wait_for(lambda: (root / "input").read_bytes() == expectedInput, "ZWJ emoji paste")
|
|
# Cancel a changed filter and ensure selection/history stay intact.
|
|
action("Insert+Control_L+e", "Emoji Menu. All tab")
|
|
key("Tab", "j", "BackSpace", "j", "Escape")
|
|
action("Insert+F6", "clipboard exported to file")
|
|
assert clipFile.read_text() == "🐈⬛"
|
|
(root / "input").write_bytes(baselineInput)
|
|
print("PASS: emoji shortcut, category/grid navigation, prefix/empty search, cancellation, history, exact ZWJ export/paste and no modal key leakage")
|
|
|
|
state = start("desktop")
|
|
emoji_trial()
|
|
action("Insert+Control_L+Shift_L+c", "clipboard cleared")
|
|
action("Insert+Shift_L+c", "clipboard empty")
|
|
key("Insert+KP_Home", "Insert+KP_Left")
|
|
action("Insert+x", "set mark")
|
|
key("KP_Next", "KP_Next", "Insert+x")
|
|
action("Insert+Shift_L+x", "alp")
|
|
action("Insert+c", "alp")
|
|
action("Insert+Shift_L+c", "alp")
|
|
action("Insert+v", "paste clipboard")
|
|
wait_for(lambda: (root / "input").read_bytes() == b"alp", "clipboard paste delivered once")
|
|
action("KP_Prior", "bravo")
|
|
action("Insert+Control_L+c", "bravo")
|
|
action("Insert+Next", "alp")
|
|
action("Insert+Prior", "bravo")
|
|
action("Insert+End", "alp")
|
|
action("Insert+Home", "bravo")
|
|
action("Insert+F6", "clipboard exported to file")
|
|
assert clipFile.read_text() == "bravo" and clipFile.stat().st_mode & 0o777 == 0o600
|
|
clipFile.write_text("from file é界")
|
|
action("Insert+F5", "from file é界")
|
|
action("Insert+F8", "exported to the X session.")
|
|
assert x_text() == "from file é界"
|
|
owner("from X é界".encode())
|
|
wait_for(lambda: x_text() == "from X é界", "external owner")
|
|
action("Insert+F7", "from X é界")
|
|
largeText = "large é界\n" * 150000
|
|
owner(largeText.encode())
|
|
wait_for(lambda: x_text() == largeText, "large external selection")
|
|
action("Insert+F7", largeText)
|
|
action("Insert+F8", "exported to the X session.")
|
|
assert x_text() == largeText
|
|
owner("from X é界".encode())
|
|
wait_for(lambda: x_text() == "from X é界", "external owner after INCR")
|
|
action("Insert+F7", "from X é界")
|
|
owner(b"not-text", "image/png")
|
|
time.sleep(0.1)
|
|
action("Insert+F7", "No text found in clipboard or no accessible display")
|
|
action("Insert+Shift_L+c", "from X é界")
|
|
print("PASS: desktop mark/preview/copy, last echo, history navigation, clear, PTY paste and file/X import/export shortcuts")
|
|
|
|
def set_sync(enabled):
|
|
action("Insert+f", "Fenrir Menu. Presentation tab")
|
|
key("Right", "Tab")
|
|
startIndex = len(server.events)
|
|
key("space")
|
|
spoken("Clipboard sync " + ("checked" if enabled else "unchecked"), startIndex)
|
|
key("Tab", "Tab", "Tab", "Return")
|
|
wait_for(lambda: tomllib.loads((root / "settings.toml").read_text())["clipboard"]["sync"] == enabled, "sync setting saved")
|
|
|
|
set_sync(True)
|
|
owner(b"automatic external")
|
|
wait_for(lambda: x_text() == "automatic external", "external owner before sync")
|
|
time.sleep(0.7)
|
|
action("Insert+Shift_L+c", "automatic external")
|
|
action("KP_Up", "bravo")
|
|
action("Insert+Control_L+c", "bravo")
|
|
wait_for(lambda: x_text() == "bravo", "automatic Fenrir export")
|
|
# Kill only this fixture reader's persistent xclip owner.
|
|
ownerPids = subprocess.check_output(["pgrep", "-P", str(state["pid"]), "-f", "[x]clip.*-quiet"], text=True).split()
|
|
assert len(ownerPids) == 1, ownerPids
|
|
os.kill(int(ownerPids[0]), signal.SIGKILL)
|
|
owner(b"after owner restart")
|
|
wait_for(lambda: x_text() == "after owner restart", "new owner after owner death")
|
|
time.sleep(1.3)
|
|
action("Insert+Shift_L+c", "after owner restart")
|
|
action("Insert+Control_L+Shift_L+c", "clipboard cleared")
|
|
time.sleep(0.7)
|
|
action("Insert+Shift_L+c", "clipboard empty")
|
|
set_sync(False)
|
|
owner(b"sync is off")
|
|
time.sleep(0.7)
|
|
action("Insert+Shift_L+c", "clipboard empty")
|
|
finish()
|
|
print("PASS: live sync checkbox, both sync directions, owner restart, clear without reimport and disabled sync")
|
|
|
|
start("laptop")
|
|
emoji_trial()
|
|
owner(b"laptop clipboard")
|
|
wait_for(lambda: x_text() == "laptop clipboard", "laptop owner")
|
|
action("Insert+F7", "laptop clipboard")
|
|
action("Insert+Shift_L+c", "laptop clipboard")
|
|
action("Insert+Control_L+u", "laptop clipboard")
|
|
action("Insert+v", "paste clipboard")
|
|
wait_for(lambda: (root / "input").read_bytes() == b"laptop clipboard", "laptop paste")
|
|
finish()
|
|
historyPath = root / "home/.local/state/fenrir/clipboard/terminal.json"
|
|
assert not historyPath.exists(), "default-off clipboard created a history file"
|
|
for layout in ["desktop", "laptop"]:
|
|
(root / "settings.toml").write_text(
|
|
'version=3\n[presentation.sound]\nenabled=false\n'
|
|
f'[clipboard]\nsync=false\npersistent=false\nnumber_of_clipboards=3\nexport_path="{clipFile}"\n'
|
|
f'[remote]\nsocket_file="{root / "remote.sock"}"\n'
|
|
)
|
|
start(layout)
|
|
wait_for(lambda: any("prompt" in item for item in server.events[-10:]), "persistence trial readiness")
|
|
for word in ["old", "middle é界", "new 🎃"]:
|
|
clipFile.write_text(word)
|
|
action("Insert+F5", word)
|
|
action("Insert+Next", "middle é界")
|
|
# Save persistence and a smaller history limit through the form.
|
|
action("Insert+f", "Fenrir Menu. Presentation tab")
|
|
key("Right", "Tab", "Tab")
|
|
startIndex = len(server.events)
|
|
key("space")
|
|
spoken("Enable Persistent Clipboard checked", startIndex)
|
|
key("Tab")
|
|
startIndex = len(server.events)
|
|
key("Down")
|
|
spoken("History items 2", startIndex)
|
|
key("Tab", "Return")
|
|
wait_for(lambda: historyPath.exists(), "enabled history write")
|
|
wait_for(lambda: json.loads(historyPath.read_text())["entries"] == ["new 🎃", "middle é界"], "live history resize")
|
|
assert historyPath.stat().st_mode & 0o777 == 0o600
|
|
assert historyPath.parent.stat().st_mode & 0o777 == 0o700
|
|
# Cancel must not apply a smaller limit or change disk history.
|
|
action("Insert+f", "Fenrir Menu. Presentation tab")
|
|
key("Right", "Tab", "Tab", "Tab")
|
|
startIndex = len(server.events)
|
|
key("Down")
|
|
spoken("History items 1", startIndex)
|
|
key("Escape")
|
|
action("Insert+Shift_L+c", "middle é界")
|
|
assert len(json.loads(historyPath.read_text())["entries"]) == 2
|
|
assert tomllib.loads((root / "settings.toml").read_text())["clipboard"]["number_of_clipboards"] == 2
|
|
finish()
|
|
start(layout)
|
|
action("Insert+Shift_L+c", "middle é界")
|
|
action("Insert+Prior", "new 🎃")
|
|
action("Insert+Next", "middle é界")
|
|
action("Insert+Control_L+Shift_L+c", "clipboard cleared")
|
|
wait_for(lambda: not historyPath.exists(), "clear deletes saved history")
|
|
finish()
|
|
start(layout)
|
|
action("Insert+Shift_L+c", "clipboard empty")
|
|
clipFile.write_text("memory retained")
|
|
action("Insert+F5", "memory retained")
|
|
wait_for(lambda: historyPath.exists(), "new copy persists after clear")
|
|
action("Insert+f", "Fenrir Menu. Presentation tab")
|
|
key("Right", "Tab", "Tab")
|
|
startIndex = len(server.events)
|
|
key("space")
|
|
spoken("Enable Persistent Clipboard unchecked", startIndex)
|
|
key("Tab", "Tab", "Return")
|
|
wait_for(lambda: not historyPath.exists(), "disabling deletes saved history")
|
|
action("Insert+Shift_L+c", "memory retained")
|
|
finish()
|
|
start(layout)
|
|
action("Insert+Shift_L+c", "clipboard empty")
|
|
clipFile.write_text("disabled copy")
|
|
action("Insert+F5", "disabled copy")
|
|
finish()
|
|
assert not historyPath.exists(), "disabled copy was persisted"
|
|
print("PASS: persistent clipboard checkbox and live limit in both layouts, Unicode/order/selection across restarts, private files, clear/restart, disable cleanup and memory-only copies")
|
|
assert not server.errors, server.errors
|
|
print("PASS: laptop clipboard import, current entry, distinct last-echo shortcut and paste; terminal restoration in both layouts")
|
|
except BaseException:
|
|
print("Recent speech:", [item[:80] for item in server.events[-30:]], file=sys.stderr)
|
|
for filename in ["reader-errors", "timing.log"]:
|
|
if (root / filename).exists():
|
|
print(filename + ":", (root / filename).read_text()[-4000:], file=sys.stderr)
|
|
raise
|
|
finally:
|
|
for process in reversed(processes):
|
|
if process.poll() is None:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=2)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|
|
server.close()
|
|
keyboard.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|