131 lines
7.2 KiB
Python
131 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright (c) 2026 Storm Dragon
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
"""Private bridge fixtures: python3 -m unittest discover -s tests -p test_codex_bridge.py."""
|
|
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import socket
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location("codex_bridge", Path(__file__).resolve().parents[1] / "tools/codex_bridge.py")
|
|
bridge = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(bridge)
|
|
|
|
|
|
def record(kind, payload):
|
|
return (json.dumps({"type": kind, "payload": payload}) + "\n").encode()
|
|
|
|
|
|
class BridgeTests(unittest.TestCase):
|
|
def test_completed_messages_only_and_partial_records(self):
|
|
meta = record("session_meta", {"id": "fixture-session"})
|
|
tool = record("event_msg", {"type": "item_completed", "item": {"type": "CommandExecution", "aggregated_output": "DO NOT READ"}})
|
|
reasoning = record("response_item", {"type": "reasoning", "text": "DO NOT READ"})
|
|
message = record("event_msg", {"type": "item_completed", "item": {"type": "AgentMessage", "content": [{"type": "Text", "text": "Complete reply."}]}})
|
|
stream = io.BytesIO(meta + tool + reasoning + message[:-2])
|
|
follower = bridge.Follower("fixture-session")
|
|
follower.read_available(stream)
|
|
self.assertFalse(follower.history)
|
|
stream.seek(0, io.SEEK_END)
|
|
stream.write(message[-2:])
|
|
follower.read_available(stream)
|
|
self.assertEqual([entry[1] for entry in follower.history], ["Complete reply."])
|
|
follower.read_available(stream)
|
|
self.assertEqual(len(follower.history), 1)
|
|
|
|
def test_baseline_bounded_history_and_session_isolation(self):
|
|
data = record("session_meta", {"id": "fixture-session"})
|
|
for index in range(15):
|
|
data += record("event_msg", {"type": "agent_message", "message": f"Reply {index}"})
|
|
follower = bridge.Follower("fixture-session")
|
|
follower.read_available(io.BytesIO(data), baseline=True)
|
|
self.assertEqual(len(follower.history), 9)
|
|
self.assertTrue(all(entry[2] for entry in follower.history))
|
|
self.assertEqual(follower.history[-1][1], "Reply 14")
|
|
with self.assertRaises(ValueError):
|
|
bridge.Follower("another-session").read_available(io.BytesIO(data))
|
|
|
|
def test_large_unicode_messages_fit_protocol_without_truncation(self):
|
|
text = 'Text 界 "\n\t\\' * 9000
|
|
base = {"version": 1, "window": "42", "terminal": 1, "session": "fixture-session"}
|
|
events = bridge.message_events(base, ["message-id", text, False])
|
|
self.assertEqual("".join(event["text"] for event in events), text)
|
|
self.assertGreater(len(events), 1)
|
|
self.assertTrue(all(len(("APPLICATION CODEX " + json.dumps(event, ensure_ascii=False) + "\n").encode()) <= 8129 for event in events))
|
|
|
|
def test_socket_delivery_and_window_routing(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
socketPath = root / f"fenrirscreenreader-{os.getpid()}.sock"
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as server:
|
|
server.bind(str(socketPath))
|
|
server.listen()
|
|
entry = {"pid": os.getpid(), "updated_at": bridge.time.time(), "screen_driver": "ptyDriver", "x11_window_id": "42", "socket_files": [str(socketPath)]}
|
|
(root / "instance.json").write_text(json.dumps(entry))
|
|
self.assertEqual(list(bridge.reader_sockets({"window": "43"}, root)), [])
|
|
destinations = list(bridge.reader_sockets({"window": "42"}, root))
|
|
self.assertEqual(len(destinations), 1)
|
|
received = []
|
|
def receive():
|
|
connection, _ = server.accept()
|
|
with connection:
|
|
data = b""
|
|
while chunk := connection.recv(8192):
|
|
data += chunk
|
|
received.append(data)
|
|
connection.sendall(b"\n")
|
|
thread = threading.Thread(target=receive)
|
|
thread.start()
|
|
self.assertTrue(bridge.send_event(destinations[0], {"kind": "status", "status": "working"}))
|
|
thread.join(timeout=2)
|
|
self.assertTrue(received[0].startswith(b"APPLICATION CODEX "))
|
|
|
|
def test_question_status_and_neutral_hook(self):
|
|
with tempfile.TemporaryDirectory() as directory, patch.object(bridge, "codex_process", return_value={"pid": 123, "started": "1", "tty": 0}), patch.object(bridge, "private_directory", return_value=Path(directory)), patch.dict(os.environ, {"WINDOWID": "42"}):
|
|
bridge.hook({"hook_event_name": "Stop", "session_id": "fixture-session", "transcript_path": "/unused", "last_assistant_message": "\nI have a question.\nContinue?"})
|
|
files = list(Path(directory).glob("*.json"))
|
|
self.assertEqual(json.loads(files[0].read_text()), {"status": "question"})
|
|
self.assertEqual(files[0].stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_question_record_completion_and_attribution_filter(self):
|
|
follower = bridge.Follower("fixture-session")
|
|
data = record("session_meta", {"id": "fixture-session"})
|
|
data += record("event_msg", {"type": "agent_message", "message": "I have a question.\nContinue?\n<oai-mem-citation>metadata</oai-mem-citation>"})
|
|
data += record("event_msg", {"type": "task_complete"})
|
|
follower.read_available(io.BytesIO(data))
|
|
self.assertEqual(follower.status, "question")
|
|
self.assertNotIn("metadata", follower.history[-1][1])
|
|
|
|
def test_neutral_main_output_on_hook_failure(self):
|
|
output = io.StringIO()
|
|
with patch.object(bridge.sys, "argv", ["codex_bridge.py"]), patch.object(bridge.sys, "stdin", io.StringIO('{"hook_event_name":"Stop"}')), patch.object(bridge.sys, "stdout", output), patch.object(bridge.sys, "stderr", io.StringIO()):
|
|
bridge.main()
|
|
self.assertEqual(json.loads(output.getvalue()), {})
|
|
|
|
def test_conversation_switch_retires_old_terminal_binding(self):
|
|
with tempfile.TemporaryDirectory() as directory, patch.object(bridge, "codex_process", return_value={"pid": 123, "started": "1", "tty": 0}), patch.object(bridge, "private_directory", return_value=Path(directory)), patch.object(bridge.subprocess, "Popen"), patch.dict(os.environ, {"WINDOWID": "42"}):
|
|
payload = {"hook_event_name":"SessionStart", "session_id":"first-session", "transcript_path":"/unused"}
|
|
bridge.hook(payload)
|
|
activePath = next(Path(directory).glob("active-*.json"))
|
|
first = json.loads(activePath.read_text())
|
|
bridge.hook(payload)
|
|
self.assertEqual(json.loads(activePath.read_text()), first)
|
|
bridge.hook(dict(payload, session_id="second-session"))
|
|
second = json.loads(activePath.read_text())
|
|
self.assertNotEqual(second["key"], first["key"])
|
|
self.assertGreater(second["epoch"], first["epoch"])
|
|
bridge.hook(dict(payload, hook_event_name="Stop"))
|
|
self.assertEqual(json.loads(activePath.read_text()), second)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|