"""Несетевые тесты самодостаточных демо. Все credentials синтетические.""" import base64 import contextlib import importlib.util import io import json import os import subprocess import sys import tempfile import types import unittest import wave from pathlib import Path from unittest.mock import patch ROOT = Path(__file__).resolve().parent def load(name): spec = importlib.util.spec_from_file_location(name, ROOT / (name + ".py")) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module rt = load("realtime-tts") chat = load("chatgpt-read-aloud") app = load("appserver-tts") class FileTests(unittest.TestCase): def test_auth_readers_reject_wrong_mode_missing_token_and_header_injection(self): with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "auth.json" for auth in [{"auth_mode": "apikey", "OPENAI_API_KEY": "0000"}, {"auth_mode": "chatgpt", "tokens": {}}, {"auth_mode": "chatgpt", "tokens": {"access_token": "[скрыто]"}}]: path.write_text(json.dumps(auth)) for reader in [rt.read_token, app.read_token, chat.read_auth]: with self.subTest(reader=reader.__module__, auth=auth): with self.assertRaises(RuntimeError): reader(path) path.write_text(json.dumps({"auth_mode": "chatgpt", "tokens": {"access_token": "[скрыто]"}})) self.assertEqual(rt.read_token(path), "synthetic-token") self.assertEqual(app.read_token(path), "synthetic-token") self.assertEqual(chat.read_auth(path)[0]["Authorization"], "Bearer synthetic-token") def test_outputs_private_complete_and_never_overwritten(self): with tempfile.TemporaryDirectory() as folder: root = Path(folder) for name, save in [("realtime.wav", lambda p: rt.save_wav(p, b"\x01\x02" * 40)), ("app.wav", lambda p: app.save_wav(p, b"\x01\x02" * 40, (24000, 1))), ("chat.mp3", lambda p: chat.save_audio(p, b"ID3-synthetic-audio"))]: output = root / name save(output) original = output.read_bytes() self.assertEqual(output.stat().st_mode & 0o777, 0o600) with self.assertRaises(FileExistsError): save(output) self.assertEqual(output.read_bytes(), original) self.assertFalse(list(root.glob(".remote-tts-*"))) if name.endswith(".wav"): with wave.open(str(output), "rb") as audio: self.assertEqual((audio.getnchannels(), audio.getframerate(), audio.getnframes()), (1, 24000, 40)) def test_three_scripts_run_after_copy_without_repo_imports(self): with tempfile.TemporaryDirectory() as folder: root = Path(folder) text = root / "text.txt" text.write_text("Проверка") for name in ["realtime-tts", "appserver-tts", "chatgpt-read-aloud"]: script = root / (name + ".py") script.write_bytes((ROOT / script.name).read_bytes()) args = [sys.executable, str(script)] if name != "chatgpt-read-aloud": args += [str(text)] args += [str(root / (name + ".out")), "--auth-file", str(root / "missing-auth"), "--dry-run"] if name == "appserver-tts": args += ["--codex", str(root / "missing-codex")] result = subprocess.run(args, cwd=folder, text=True, capture_output=True, timeout=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(json.loads(result.stdout)["credentials_read"]) def test_bad_auth_not_dumped_by_cli(self): with tempfile.TemporaryDirectory() as folder: root = Path(folder) text = root / "text.txt" text.write_text("Проверка") auth = root / "auth.json" auth.write_text('{"private-synthetic-token": invalid}') for name in ["realtime-tts", "chatgpt-read-aloud"]: args = [sys.executable, str(ROOT / (name + ".py"))] if name != "chatgpt-read-aloud": args += [str(text)] args += [str(root / "output"), "--auth-file", str(auth)] result = subprocess.run(args, text=True, capture_output=True, timeout=5) self.assertEqual(result.returncode, 1) self.assertNotIn("private-synthetic-token", result.stdout + result.stderr) self.assertNotIn("Traceback", result.stderr) class ChatTests(unittest.TestCase): def test_current_account_only(self): with tempfile.TemporaryDirectory() as folder: state = Path(folder) / "state.json" state.write_text(json.dumps({"electron-persisted-atom-state": {"chatgpt-sidebar-state-v1": { "own": {"pinnedConversations": [{"conversation": {"id": "own-conversation"}}]}, "other": {"pinnedConversations": [{"conversation": {"id": "other-conversation"}}]}}}})) self.assertEqual(chat.pinned_conversation(state, "own"), "own-conversation") with self.assertRaises(RuntimeError): chat.pinned_conversation(state, "missing-account") def test_active_branch_ignores_hidden_analysis_and_tool_messages(self): def node(parent, **extra): return {"parent": parent, "message": {"id": "visible-id", "author": {"role": "assistant"}, "content": {"content_type": "text"}, **extra}} conversation = {"current_node": "tool", "mapping": { "tool": node("hidden", recipient="tool"), "hidden": node("analysis", metadata={"is_visually_hidden_from_conversation": True}), "analysis": node("final", channel="analysis"), "final": node(None, channel="final"), "other-branch": node(None, id="unrelated-id")}} self.assertEqual(chat.assistant_message(conversation), "visible-id") with self.assertRaises(RuntimeError): chat.assistant_message({"current_node": "cycle", "mapping": {"cycle": {"parent": "cycle"}}}) def test_voice_catalog_does_not_return_preview_urls_or_extra_fields(self): data = json.dumps({"voices": [{"voice": "glimmer", "name": "Sol", "preview_url": "private"}]}).encode() with patch.object(chat, "get", return_value=(data, "application/json")): self.assertEqual(chat.voice_catalog(None, {}, "ru", 1), [{"id": "glimmer", "name": "Sol"}]) def test_cli_friendly_voice_is_encoded_as_backend_id(self): with tempfile.TemporaryDirectory() as folder: root = Path(folder) auth = root / "auth.json" auth.write_text(json.dumps({"auth_mode": "chatgpt", "tokens": {"access_token": "[скрыто]"}})) urls = [] def get(opener, url, headers, timeout, limit): urls.append(url) return b"ID3-synthetic-audio", "audio/mpeg" args = ["demo", str(root / "out.mp3"), "--auth-file", str(auth), "--voice", "Sol", "--conversation-id", "chat&id", "--message-id", "message&id"] capture = io.StringIO() with patch("sys.argv", args), patch.object(chat, "voice_catalog", return_value=[{"id": "glimmer", "name": "Sol"}]), patch.object(chat, "get", side_effect=get), contextlib.redirect_stdout(capture): chat.main() self.assertIn("voice=glimmer", urls[0]) self.assertIn("conversation_id=chat%26id", urls[0]) self.assertNotIn("synthetic-token", capture.getvalue()) self.assertNotIn("chat&id", capture.getvalue()) def test_redirect_refused(self): self.assertIsNone(chat.NoRedirect().redirect_request(None, None, 302, None, {}, "https://untrusted.example/")) class RealtimeTests(unittest.TestCase): def socket_module(self, events): class Socket: def __init__(self): self.sent = [] self.closed = False def recv(self): return json.dumps(events.pop(0)) def send(self, data): self.sent.append(json.loads(data)) def settimeout(self, timeout): self.timeout = timeout def close(self): self.closed = True socket = Socket() class WebSocketError(Exception): pass module = types.SimpleNamespace(create_connection=lambda *a, **k: socket, WebSocketException=WebSocketError, WebSocketBadStatusException=type("Handshake", (WebSocketError,), {})) return socket, module def test_chosen_voice_and_completed_pcm(self): events = [{"type": "session.created"}, {"type": "session.updated"}, {"type": "response.created", "response": {"id": "own"}}, {"type": "response.output_audio.delta", "response_id": "foreign", "delta": base64.b64encode(b"wrong").decode()}, {"type": "response.output_audio.delta", "response_id": "own", "delta": base64.b64encode(b"\x01\x02").decode()}, {"type": "response.done", "response": {"id": "own", "status": "completed"}}] socket, module = self.socket_module(events) with patch.dict("sys.modules", {"websocket": module}): self.assertEqual(rt.synthesize("Проверка", "synthetic-token", "gpt-realtime", "cedar", 5), b"\x01\x02") self.assertEqual(socket.sent[0]["session"]["audio"]["output"]["voice"], "cedar") self.assertTrue(socket.closed) def test_incomplete_response_and_server_errors_not_success(self): for events in [[{"type": "session.created"}, {"type": "session.updated"}, {"type": "response.done", "response": {"status": "failed"}}], [{"type": "error", "error": {"message": "private-synthetic-token"}}]]: socket, module = self.socket_module(events) with patch.dict("sys.modules", {"websocket": module}): with self.assertRaises(RuntimeError) as error: rt.synthesize("Проверка", "synthetic-token", "gpt-realtime", "marin", 5) self.assertNotIn("private-synthetic-token", str(error.exception)) self.assertTrue(socket.closed) def test_list_voices_needs_no_files_or_credentials(self): r = subprocess.run([sys.executable, str(ROOT / "realtime-tts.py"), "--list-voices", "--auth-file", "/missing-auth"], text=True, capture_output=True, timeout=5) self.assertEqual(r.returncode, 0, r.stderr) self.assertEqual(len(json.loads(r.stdout)["voices"]), 10) self.assertFalse(json.loads(r.stdout)["account_access_checked"]) def test_bad_base64_odd_pcm_empty_and_foreign_done_rejected(self): variants = [ [{"type": "response.output_audio.delta", "response_id": "own", "delta": "!bad-base64"}], [{"type": "response.output_audio.delta", "response_id": "own", "delta": base64.b64encode(b"\x01").decode()}, {"type": "response.done", "response": {"id": "own", "status": "completed"}}], [{"type": "response.done", "response": {"id": "own", "status": "completed"}}], [{"type": "response.done", "response": {"id": "foreign", "status": "completed"}}], ] for tail in variants: socket, module = self.socket_module([{"type": "session.created"}, {"type": "session.updated"}, {"type": "response.created", "response": {"id": "own"}}, *tail]) with patch.dict("sys.modules", {"websocket": module}), self.assertRaises((RuntimeError, ValueError)): rt.synthesize("Проверка", "synthetic", "gpt-realtime", "cedar", 5) self.assertTrue(socket.closed) FAKE_SERVER = '''#!/usr/bin/python3 import sys,json,os,base64 from pathlib import Path root=Path(__file__).parent home=Path(os.environ["CODEX_HOME"]) root.joinpath("home-path").write_text(str(home)) root.joinpath("env-names").write_text(json.dumps(list(os.environ))) for line in sys.stdin: msg=json.loads(line);method=msg["method"] if "id" not in msg:continue result={} if method=="thread/realtime/listVoices":result={"voices":{"v1":["cove"],"v2":["ash","marin"],"defaultV1":"cove","defaultV2":"marin"}} if method=="account/login/start":home.joinpath("auth.json").write_text(json.dumps({"OPENAI_API_KEY":msg["params"]["apiKey"]})) if method=="thread/start":result={"thread":{"id":"test-thread"}};root.joinpath("cwd-path").write_text(msg["params"]["cwd"]) if method=="thread/realtime/start":root.joinpath("chosen-voice").write_text(msg["params"]["voice"]) if method=="thread/realtime/appendSpeech" and root.joinpath("emit-early").exists(): print(json.dumps({"method":"thread/realtime/outputAudio/delta","params":{"threadId":"test-thread","audio":{"sampleRate":24000,"numChannels":1,"data":base64.b64encode(b"\\x03\\x04"*20).decode()}}}),flush=True) print(json.dumps({"id":msg["id"],"result":result}),flush=True) if method=="thread/realtime/appendSpeech": if root.joinpath("emit-error").exists():print(json.dumps({"method":"thread/realtime/error","params":{"message":"private synthetic-token detail"}}),flush=True) elif not root.joinpath("emit-timeout").exists():print(json.dumps({"method":"thread/realtime/outputAudio/delta","params":{"threadId":"test-thread","audio":{"sampleRate":24000,"numChannels":1,"data":base64.b64encode(b"\\x01\\x02"*40).decode()}}}),flush=True) ''' class AppServerTests(unittest.TestCase): def server(self, root): server = root / "fake-codex" server.write_text(FAKE_SERVER) server.chmod(0o700) return server def test_separate_home_cwd_selected_voice_and_clean_environment(self): with tempfile.TemporaryDirectory() as folder: root = Path(folder) with patch.dict("os.environ", {"OPENAI_API_KEY": "[скрыто]", "HTTPS_PROXY": "untrusted", "CODEX_BASE_URL": "untrusted"}): pcm, shape = app.synthesize("Проверка", "synthetic-token", self.server(root), 6, "ash") self.assertEqual((pcm, shape), (b"\x01\x02" * 40, (24000, 1))) self.assertEqual((root / "chosen-voice").read_text(), "ash") home = Path((root / "home-path").read_text()) self.assertEqual(Path((root / "cwd-path").read_text()).parent, home) self.assertFalse(home.exists()) names = json.loads((root / "env-names").read_text()) for name in ["OPENAI_API_KEY", "HTTPS_PROXY", "CODEX_BASE_URL"]: self.assertNotIn(name, names) def test_catalog_needs_no_token_and_cleans_home(self): with tempfile.TemporaryDirectory() as folder: root = Path(folder) result = app.synthesize("", None, self.server(root), 2, list_voices=True) self.assertEqual(result["defaultV2"], "marin") self.assertFalse(Path((root / "home-path").read_text()).exists()) self.assertFalse((root / "cwd-path").exists()) def test_invalid_voice_and_server_error_always_clean_home(self): with tempfile.TemporaryDirectory() as folder: root = Path(folder) server = self.server(root) with self.assertRaisesRegex(RuntimeError, "Голос"): app.synthesize("Проверка", "synthetic-token", server, 2, "not-supported") self.assertFalse(Path((root / "home-path").read_text()).exists()) (root / "emit-error").touch() with self.assertRaises(RuntimeError) as error: app.synthesize("Проверка", "synthetic-token", server, 2, "ash") self.assertNotIn("synthetic-token", str(error.exception)) self.assertFalse(Path((root / "home-path").read_text()).exists()) def test_timeout_always_clean_home(self): with tempfile.TemporaryDirectory() as folder: root = Path(folder) (root / "emit-timeout").touch() with self.assertRaises(RuntimeError): app.synthesize("Проверка", "synthetic-token", self.server(root), .2, "ash") self.assertFalse(Path((root / "home-path").read_text()).exists()) def test_audio_before_rpc_ack_not_lost(self): with tempfile.TemporaryDirectory() as folder: root = Path(folder) (root / "emit-early").touch() pcm, _ = app.synthesize("Проверка", "synthetic-token", self.server(root), 6, "ash") self.assertEqual(pcm, b"\x03\x04" * 20 + b"\x01\x02" * 40) self.assertFalse(Path((root / "home-path").read_text()).exists()) def test_wrong_pcm_parameters_and_base64_clean_home(self): for old, new in [('"sampleRate":24000', '"sampleRate":0'), ('"numChannels":1', '"numChannels":0'), ('base64.b64encode(b"\\x01\\x02"*40).decode()', '"!invalid-base64"')]: with tempfile.TemporaryDirectory() as folder: root = Path(folder) server = root / "fake-codex" server.write_text(FAKE_SERVER.replace(old, new)) server.chmod(0o700) with self.assertRaises((RuntimeError, ValueError)): app.synthesize("Проверка", "synthetic", server, 2, "ash") self.assertFalse(Path((root / "home-path").read_text()).exists()) if __name__ == "__main__": unittest.main()