131 lines
6.2 KiB
Python
131 lines
6.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Two real readers with private X11, speech, sockets and shared fixture HOME.
|
|
|
|
Run sequentially after the other native fixtures, under xvfb-run with
|
|
FENRIR_PRIVATE_X11=1. Does not access live readers or personal clipboard data.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
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 under xvfb-run with FENRIR_PRIVATE_X11=1")
|
|
repo = Path(__file__).resolve().parents[1]
|
|
script = repo / "tests/native_terminal.py"
|
|
processes, servers = [], []
|
|
keyboard = KeyboardFixture()
|
|
with tempfile.TemporaryDirectory(prefix="fenrir-shared-clipboard-", dir="/tmp") as directory:
|
|
base = Path(directory)
|
|
home = base / "home"
|
|
home.mkdir()
|
|
historyPath = home / ".local/state/fenrir/clipboard/terminal.json"
|
|
|
|
def start(name):
|
|
root = base / name
|
|
root.mkdir()
|
|
(root / "keyboard").mkdir()
|
|
(root / "keyboard/custom.toml").write_bytes((repo / "config/keyboard/desktop.toml").read_bytes())
|
|
(root / "settings.toml").write_text(
|
|
'version=3\n[presentation.sound]\nenabled=false\n'
|
|
f'[clipboard]\npersistent=true\nsync=false\nexport_path="{root / "export.txt"}"\n'
|
|
f'[remote]\nsocket_file="{root / "remote.sock"}"\n'
|
|
f'[general]\nscript_path="{root / "scripts"}"\n'
|
|
)
|
|
(root / "shell").write_text(
|
|
f'#!{sys.executable}\nimport runpy, sys\nsys.argv = [{str(script)!r}, "--fixture-shell"]\n'
|
|
f'runpy.run_path({str(script)!r}, run_name="__main__")\n'
|
|
)
|
|
(root / "shell").chmod(0o700)
|
|
server = SpeechServer(root / "speech.sock")
|
|
servers.append(server)
|
|
environment = dict(os.environ, HOME=str(home), TMPDIR=str(root),
|
|
FENRIR_FIXTURE_DIR=str(root), SPEECHD_ADDRESS="unix_socket:" + str(root / "speech.sock"))
|
|
process = subprocess.Popen(["xterm", "-geometry", "80x25", "-T", name, "-e",
|
|
sys.executable, str(script), "--outer", str(root)], env=environment)
|
|
processes.append(process)
|
|
wait_for(lambda: (root / "size").exists(), "reader shell ready")
|
|
state = json.loads((root / "started").read_text())
|
|
return root, server, state
|
|
|
|
def remote(reader, text):
|
|
with socket.socket(socket.AF_UNIX) as connection:
|
|
connection.settimeout(2)
|
|
connection.connect(str(reader[0] / "remote.sock"))
|
|
connection.sendall(("COMMAND " + text + "\n").encode())
|
|
connection.shutdown(socket.SHUT_WR)
|
|
reply = connection.recv(4096)
|
|
assert not reply.strip(), reply
|
|
|
|
def action(reader, chord, expected):
|
|
subprocess.run(["xdotool", "windowfocus", "--sync", reader[2]["window"]], check=True)
|
|
startIndex = len(reader[1].events)
|
|
keyboard.key(chord)
|
|
wait_for(lambda: expected in reader[1].events[startIndex:], "speech " + expected)
|
|
|
|
def export_matches(reader, expected):
|
|
remote(reader, "EXPORTCLIPBOARD")
|
|
time.sleep(0.03)
|
|
path = reader[0] / "export.txt"
|
|
return path.exists() and path.read_text() == expected
|
|
|
|
try:
|
|
first, second = start("FirstFixture"), start("SecondFixture")
|
|
remote(first, "CLIPBOARD first copy")
|
|
wait_for(lambda: export_matches(second, "first copy"), "first copy reaches second reader")
|
|
remote(second, "CLIPBOARD second copy")
|
|
wait_for(lambda: json.loads(historyPath.read_text())["entries"] == ["second copy", "first copy"],
|
|
"both copies retained")
|
|
wait_for(lambda: export_matches(first, "first copy"), "first reader retains selection")
|
|
action(first, "Insert+Home", "second copy")
|
|
action(first, "Insert+Next", "first copy")
|
|
action(second, "Insert+Control_L+Shift_L+c", "clipboard cleared")
|
|
wait_for(lambda: not historyPath.exists(), "shared snapshot removed")
|
|
time.sleep(0.3)
|
|
action(first, "Insert+Shift_L+c", "clipboard empty")
|
|
remote(first, "CLIPBOARD survivor copy")
|
|
wait_for(lambda: export_matches(second, "survivor copy"), "copy after peer Clear")
|
|
remote(first, "QUITAPPLICATION")
|
|
wait_for(lambda: (first[0] / "finished").exists(), "first reader exits")
|
|
assert json.loads((first[0] / "finished").read_text())["status"] == 0
|
|
remote(second, "CLIPBOARD after exit")
|
|
wait_for(lambda: json.loads(historyPath.read_text())["entries"] == ["after exit", "survivor copy"],
|
|
"remaining reader persists after peer exit")
|
|
remote(second, "QUITAPPLICATION")
|
|
wait_for(lambda: (second[0] / "finished").exists(), "second reader exits")
|
|
for reader in [first, second]:
|
|
finished = json.loads((reader[0] / "finished").read_text())
|
|
assert finished == {"status": 0, "restored": True, "flagsRestored": True}, finished
|
|
assert not reader[1].errors, reader[1].errors
|
|
assert "clipboard history" not in (reader[0] / "reader-errors").read_text().lower()
|
|
print("PASS: two real readers share persistent copies and Clear, retain local selection, survive peer exit and restore terminals without clipboard errors")
|
|
except BaseException:
|
|
for server in servers:
|
|
print("Recent speech:", server.events[-12:], 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()
|
|
for server in servers:
|
|
server.close()
|
|
keyboard.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|