#!/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?\nmetadata"}) 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()