167 lines
8.8 KiB
Python
167 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
||
# Copyright (c) 2026 Storm Dragon
|
||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||
"""Private X11 bridge/PTY test, with fake speech and no live Codex requests."""
|
||
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import tty
|
||
|
||
|
||
def fixture_shell():
|
||
tty.setraw(0)
|
||
os.write(1, "\x1b[2J\x1b[HOpenAI Codex\x1b[22;1H› ".encode())
|
||
while True:
|
||
data = os.read(0, 4096)
|
||
if not data:
|
||
return
|
||
if data == b"\x04":
|
||
os.write(1, b"\x1b[24;1H\x1b[2K[Username@host directory] $ ")
|
||
continue
|
||
os.write(1, data)
|
||
|
||
|
||
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]
|
||
spec = importlib.util.spec_from_file_location("bridge", repo / "tools/codex_bridge.py")
|
||
bridge = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(bridge)
|
||
with tempfile.TemporaryDirectory(prefix="fenrir-codex-native-") as directory:
|
||
root = Path(directory)
|
||
(root / "keyboard").mkdir()
|
||
(root / "keyboard/custom.toml").write_bytes((repo / "config/keyboard/desktop.toml").read_bytes())
|
||
(root / "home").mkdir()
|
||
(root / "settings.toml").write_text('version=3\n[presentation.sound]\nenabled=false\n'
|
||
f'[remote]\nsocket_file="{root / "remote.sock"}"\n')
|
||
shell = root / "shell"
|
||
shell.write_text(f'#!{sys.executable}\nimport runpy, sys\nsys.argv=[{str(Path(__file__).resolve())!r}, "--shell"]\nrunpy.run_path(sys.argv[0], 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_TEST_BINARY=os.environ.get("FENRIR_TEST_BINARY", str(repo / "fenrir")),
|
||
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)
|
||
watcher = 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)
|
||
wait_for(lambda: any("OpenAI Codex" in item for item in server.events), "fixture screen ready")
|
||
def spoken(text, start):
|
||
wait_for(lambda: any(text in item for item in server.events[start:]), f"speech {text!r}")
|
||
def action(chord, expected):
|
||
start = len(server.events)
|
||
keyboard.key(chord)
|
||
spoken(expected, start)
|
||
action("Insert+a", "Application Default")
|
||
action("c", "Codex")
|
||
action("Return", "Codex application override selected")
|
||
action("Insert+0", "Waiting for Codex bridge")
|
||
session = "fixture-session"
|
||
transcript = root / "session.jsonl"
|
||
def append_record(kind, payload):
|
||
with transcript.open("ab") as stream:
|
||
stream.write((json.dumps({"type": kind, "payload": payload}) + "\n").encode())
|
||
append_record("session_meta", {"id": session})
|
||
append_record("event_msg", {"type": "agent_message", "message": "Existing reply"})
|
||
statusPath = root / "status.json"
|
||
bridge.write_status(statusPath, "working")
|
||
activePath = root / "active.json"
|
||
bridge.write_json(activePath, {"key":"fixture", "epoch":1})
|
||
config = {"pid": os.getpid(), "started": bridge.process_info(os.getpid())[2],
|
||
"window": str(state["window"]), "terminal": 1, "session": session,
|
||
"transcript": str(transcript), "status_path": str(statusPath), "lock_path": str(root / "watch.lock"),
|
||
"active_path": str(activePath), "key":"fixture"}
|
||
watcher = subprocess.Popen([sys.executable, str(repo / "tools/codex_bridge.py"), "--watch", json.dumps(config)], env=env)
|
||
wait_for(lambda: (root / "watch.lock").exists(), "bridge startup")
|
||
# Wait for baseline delivery by polling the manual history command.
|
||
start = len(server.events)
|
||
for _ in range(20):
|
||
keyboard.key("Insert+1")
|
||
time.sleep(0.2)
|
||
if "Existing reply" in server.events[start:]:
|
||
break
|
||
else:
|
||
raise AssertionError("baseline history unavailable")
|
||
action("Insert+0", "Codex working")
|
||
# The raw fixture echoes terminal cursor escapes just like a TUI.
|
||
# Actual XTest arrows must speak the unchanged composer character.
|
||
keyboard.key("h", "e", "l", "l", "o")
|
||
action("Left", "o")
|
||
action("Left", "l")
|
||
action("Right", "o")
|
||
start = len(server.events)
|
||
append_record("event_msg", {"type": "item_completed", "item": {"type": "CommandExecution", "aggregated_output": "Private tool result"}})
|
||
message = json.dumps({"type": "event_msg", "payload": {"type": "item_completed", "item": {"type": "AgentMessage", "content": [{"type": "Text", "text": "A completed progress update."}]}}}) + "\n"
|
||
with transcript.open("ab") as stream:
|
||
stream.write(message[:-3].encode())
|
||
time.sleep(0.5)
|
||
assert not any("completed progress" in item or "Private tool result" in item for item in server.events[start:])
|
||
with transcript.open("ab") as stream:
|
||
stream.write(message[-3:].encode())
|
||
spoken("A completed progress update.", start)
|
||
time.sleep(2.3) # Include a complete heartbeat/retry interval.
|
||
assert server.events[start:].count("A completed progress update.") == 1, server.events[start:]
|
||
action("Insert+1", "A completed progress update.")
|
||
action("Insert+2", "Existing reply")
|
||
start = len(server.events)
|
||
longReply = "Long reply " + "word " * 2000
|
||
append_record("event_msg", {"type": "agent_message", "message": longReply})
|
||
spoken("Long reply", start)
|
||
assert any(item.strip() == longReply.strip() for item in server.events[start:]), "long reply truncated"
|
||
watcher.terminate()
|
||
watcher.wait(timeout=3)
|
||
bridge.write_status(statusPath, "working")
|
||
start = len(server.events)
|
||
watcher = subprocess.Popen([sys.executable, str(repo / "tools/codex_bridge.py"), "--watch", json.dumps(config)], env=env)
|
||
time.sleep(0.6)
|
||
assert not any("Long reply" in item for item in server.events[start:]), "replay spoke old history"
|
||
bridge.write_status(statusPath, "done")
|
||
time.sleep(0.4)
|
||
action("Insert+0", "Codex finished")
|
||
# Leave old Codex/composer text visible and keep the bridge alive:
|
||
# the fresh shell prompt itself must restore ordinary behavior.
|
||
action("ctrl+d", "Default settings restored")
|
||
keyboard.key("h", "e", "l", "l", "o")
|
||
action("Left", "o")
|
||
action("Left", "l")
|
||
action("Right", "o")
|
||
keyboard.key("Insert+q")
|
||
wait_for(lambda: (root / "finished").exists(), "reader exit")
|
||
assert not server.errors, server.errors
|
||
print("PASS: real private PTY/override/socket bridge, composer Left/Right feedback, baseline history, completed messages only, one announcement, long replies, replay, status, shell exit without bridge closure and restored shell arrows")
|
||
except BaseException:
|
||
print("Recent private speech:", server.events[-15:], file=sys.stderr)
|
||
if (root / "status.json").exists():
|
||
print("Bridge status:", (root / "status.json").read_text(), file=sys.stderr)
|
||
if (root / "reader-errors").exists():
|
||
print("Reader errors:", (root / "reader-errors").read_text()[-2500:], file=sys.stderr)
|
||
raise
|
||
finally:
|
||
if watcher is not None and watcher.poll() is None:
|
||
watcher.terminate()
|
||
watcher.wait(timeout=3)
|
||
process.terminate()
|
||
try:
|
||
process.wait(timeout=3)
|
||
except subprocess.TimeoutExpired:
|
||
process.kill()
|
||
keyboard.close()
|
||
server.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
fixture_shell() if "--shell" in sys.argv else main()
|