"""Проверки HTTP-контрактов и живого MCP-процесса без Desktop и без удалённого TTS.""" import argparse import importlib.util import json import os import socket import subprocess import tempfile import threading import time import unittest from unittest.mock import patch from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parent def load(name): spec = importlib.util.spec_from_file_location(name.replace("-", "_"), ROOT / (name + ".py")) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module http = load("http-tts") mcp = load("mcp-client") realtime = load("realtime-text") BINARY = ROOT.parents[1] / "target/debug/codex-read-aloud-linux" class RequestTests(unittest.TestCase): def args(self, mode): return argparse.Namespace(mode=mode, language="ru", voice=None, model="gpt-4o-mini-tts", conversation_id="chat/id", message_id="message&x", summary_format="podcast") def test_arbitrary_chatgpt_text_contract(self): url, method, body = http.build_request(self.args("chatgpt-text"), "Привет") self.assertEqual(url, "https://chatgpt.com/backend-api/pronunciation/synthesize?format=mp3") self.assertEqual(method, "POST") self.assertEqual(body, {"text": "Привет", "pronunciation_language": "ru", "source": "translation"}) def test_existing_message_ids_encoded(self): url, method, body = http.build_request(self.args("chatgpt-message"), "") self.assertIn("conversation_id=chat%2Fid", url) self.assertIn("message_id=message%26x", url) self.assertEqual((method, body), ("GET", None)) def test_summary_uses_conversation_path(self): url, method, body = http.build_request(self.args("chatgpt-summary"), "Brief") self.assertIn("/conversation/chat%2Fid/audio_summary/synthesize", url) self.assertEqual(body["custom_instructions"], "Brief") self.assertEqual(body["format"], "podcast") def test_summary_format_is_not_audio_codec(self): args = self.args("chatgpt-summary") for kind in ("podcast", "executive_briefing", "study_guide"): args.summary_format = kind self.assertEqual(http.build_request(args, "")[2], {"format": kind, "custom_instructions": None}) def test_realtime_explicit_chatgpt_file_and_sentinel_rejected(self): with tempfile.TemporaryDirectory() as root: auth = Path(root) / "auth.json" auth.write_text(json.dumps({"auth_mode": "chatgpt", "tokens": {"access_token": "[скрыто]"}})) self.assertEqual(realtime.read_credential(chatgpt_auth_file=auth), "synthetic-token") auth.write_text('{"auth_mode":"apikey","OPENAI_API_KEY":"0000"}') with self.assertRaises(ValueError): realtime.read_credential(chatgpt_auth_file=auth) key = Path(root) / "key" key.write_text("0000") with self.assertRaises(ValueError): realtime.read_credential(api_key_file=key) def test_platform_speech_contract_and_limit(self): url, method, body = http.build_request(self.args("openai"), "Speech") self.assertEqual(url, "https://api.openai.com/v1/audio/speech") self.assertEqual(body["response_format"], "wav") self.assertEqual(body["model"], "gpt-4o-mini-tts") with self.assertRaises(ValueError): http.build_request(self.args("openai"), "x" * 4097) def test_auth_read_from_file_and_api_mode_rejected(self): with tempfile.TemporaryDirectory() as root: path = Path(root) / "auth.json" path.write_text(json.dumps({"auth_mode": "chatgpt", "tokens": { "access_token": "[скрыто]", "account_id": "test-account"}})) self.assertEqual(http.chatgpt_headers(path)["Authorization"], "Bearer local-test-token") path.write_text('{"auth_mode":"apikey","OPENAI_API_KEY":"0000"}') with self.assertRaises(ValueError): http.chatgpt_headers(path) class HttpTests(unittest.TestCase): def setUp(self): self.visited = [] visited = self.visited class Handler(BaseHTTPRequestHandler): def do_GET(self): visited.append(self.path) if self.path == "/redirect": self.send_response(302) self.send_header("Location", "/audio") self.end_headers() return if self.path == "/error": self.send_response(403) self.send_header("cf-mitigated", "challenge") self.end_headers() self.wfile.write(b"private response body") return self.send_response(200) self.send_header("Content-Type", "application/json" if self.path == "/json" else "audio/mpeg") self.end_headers() self.wfile.write(b'{"error":true}' if self.path in {"/json", "/fakeaudio"} else b"ID3-test-audio") def log_message(self, *args): pass self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) threading.Thread(target=self.server.serve_forever, daemon=True).start() self.root = tempfile.TemporaryDirectory() self.output = Path(self.root.name) / "audio.mp3" def tearDown(self): self.server.shutdown() self.server.server_close() self.root.cleanup() def call(self, path): return http.download_audio(f"http://127.0.0.1:{self.server.server_port}{path}", "GET", None, {"Authorization": "[скрыто]"}, self.output, 5) def test_audio_saved_and_existing_file_preserved(self): self.assertEqual(self.call("/audio")["http_status"], 200) original = self.output.read_bytes() with self.assertRaises(FileExistsError): self.call("/audio") self.assertEqual(self.output.read_bytes(), original) def test_json_and_fake_audio_never_saved(self): for path in ["/json", "/fakeaudio"]: with self.assertRaises(RuntimeError): self.call(path) self.assertFalse(self.output.exists()) def test_redirect_not_followed_with_credentials(self): with self.assertRaisesRegex(RuntimeError, "HTTP 302"): self.call("/redirect") self.assertEqual(self.visited, ["/redirect"]) def test_challenge_report_omits_body(self): with self.assertRaisesRegex(RuntimeError, "Cloudflare challenge=True") as caught: self.call("/error") self.assertNotIn("private response body", str(caught.exception)) class McpProcessTests(unittest.TestCase): def setUp(self): if not BINARY.is_file(): self.fail("Сначала cargo build -p codex-read-aloud-linux; интеграционные проверки не пропускаются") self.root = tempfile.TemporaryDirectory() self.capture = Path(self.root.name) / "text.txt" helper = Path(self.root.name) / "stdin-helper" helper.write_text('#!/usr/bin/python3\nimport os,sys,time\nfrom pathlib import Path\n' 'Path(os.environ["TTS_TEST_CAPTURE"]).write_text(sys.stdin.read())\ntime.sleep(60)\n') helper.chmod(0o700) env = {**os.environ, "CODEX_LINUX_READ_ALOUD_COMMAND": str(helper), "TTS_TEST_CAPTURE": str(self.capture)} self.client = mcp.McpClient(BINARY, env=env) def tearDown(self): self.client.close() self.root.cleanup() def test_tools_and_empty_input(self): names = {x["name"] for x in self.client.request("tools/list", {})["tools"]} self.assertEqual(names, {"doctor", "read_aloud", "stop"}) self.assertEqual(self.client.tool("read_aloud", {"text": ""})["reason"], "empty") self.assertFalse(self.client.tool("stop")["stopped"]) def test_cleaned_text_forwarded_to_real_process(self): result = self.client.tool("read_aloud", {"text": "# Привет **мир**\n```sh\nevil\n```", "max_chars": 10}) self.assertTrue(result["started"]) deadline = time.monotonic() + 3 while not self.capture.exists() and time.monotonic() < deadline: time.sleep(.01) self.assertEqual(self.capture.read_text(), "Привет мир") self.assertEqual(self.client.tool("doctor")["active_backend"], "custom") self.assertTrue(self.client.tool("stop")["stopped"]) self.assertIsNone(self.client.tool("doctor").get("active_backend")) def test_busy_interrupt_and_stop(self): self.assertTrue(self.client.tool("read_aloud", {"text": "first"})["started"]) self.assertEqual(self.client.tool("read_aloud", {"text": "second", "interrupt": False})["reason"], "busy") self.assertTrue(self.client.tool("read_aloud", {"text": "third", "interrupt": True})["started"]) self.assertTrue(self.client.tool("stop")["stopped"]) self.assertFalse(self.client.tool("stop")["stopped"]) def test_unknown_tool_and_missing_text_rejected(self): for tool, args in [("missing", {}), ("read_aloud", {})]: with self.assertRaises(RuntimeError): self.client.tool(tool, args) class AdapterTests(unittest.TestCase): def test_stdin_socketpair_from_electron_node(self): with tempfile.TemporaryDirectory() as root: capture = Path(root) / "arguments.json" shim = Path(root) / "spd-say" shim.write_text('#!/usr/bin/python3\nimport os,sys,json\nfrom pathlib import Path\n' 'Path(os.environ["ADAPTER_CAPTURE"]).write_text(json.dumps(sys.argv[1:]))\n') shim.chmod(0o700) env = {**os.environ, "PATH": root + ":" + os.environ.get("PATH", ""), "ADAPTER_CAPTURE": str(capture)} left, right = socket.socketpair() try: process = subprocess.Popen(["bash", str(ROOT / "spd-stdin.sh")], stdin=left, env=env) left.close() text = "Привет $() `текст` без завершающей новой строки" right.sendall(text.encode()) right.shutdown(socket.SHUT_WR) self.assertEqual(process.wait(timeout=5), 0) self.assertEqual(json.loads(capture.read_text())[-1], text) finally: left.close() right.close() class RealtimeTests(unittest.TestCase): def test_text_item_and_audio_events(self): import base64 import types event_list = [{"type": "session.created"}, {"type": "session.updated"}, {"type": "response.output_audio.delta", "delta": base64.b64encode(b"\x01\x02").decode()}, {"type": "response.done", "response": {"status": "completed"}}] class Socket: sent = [] closed = False def send(self, message): self.sent.append(json.loads(message)) def recv(self): return json.dumps(event_list.pop(0)) def close(self): self.closed = True socket = Socket() module = types.SimpleNamespace(create_connection=lambda *args, **kwargs: socket) with patch.dict("sys.modules", {"websocket": module}): self.assertEqual(realtime.synthesize("Привет", "gpt-realtime", "marin", "synthetic-key", 1), b"\x01\x02") self.assertEqual(socket.sent[1]["item"]["content"], [{"type": "input_text", "text": "Привет"}]) self.assertTrue(socket.closed) def test_realtime_error_never_creates_success(self): import types class Socket: def recv(self): return '{"type":"error","error":{"message":"private-detail"}}' def close(self): pass with patch.dict("sys.modules", {"websocket": types.SimpleNamespace(create_connection=lambda *a, **k: Socket())}): with self.assertRaisesRegex(RuntimeError, "вернул error"): realtime.synthesize("hi", "gpt-realtime", "marin", "synthetic-key", 1) if __name__ == "__main__": unittest.main()