"""Несетевые проверки учебного клиента с синтетическими credentials и аудио.""" import base64 import importlib.util import json import subprocess import sys import tempfile import types import unittest import wave from pathlib import Path from unittest.mock import patch HERE = Path(__file__).resolve().parent spec = importlib.util.spec_from_file_location("stt", HERE / "realtime-transcribe.py") stt = importlib.util.module_from_spec(spec) spec.loader.exec_module(stt) def wav(path, rate=24000, frames=24000): with wave.open(str(path), "wb") as audio: audio.setnchannels(1) audio.setsampwidth(2) audio.setframerate(rate) audio.writeframes(b"\x01\x00" * frames) class Socket: def __init__(self, events): self.events = iter(events) self.sent = [] self.closed = False def recv(self): return json.dumps(next(self.events)) def send(self, text): self.sent.append(json.loads(text)) def settimeout(self, seconds): assert seconds > 0 def close(self): self.closed = True class WebSocketError(Exception): pass class HandshakeError(WebSocketError): status_code = 403 class Tests(unittest.TestCase): def module(self, factory): return types.SimpleNamespace(create_connection=factory, WebSocketException=WebSocketError, WebSocketBadStatusException=HandshakeError) def test_pcm_validation(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "audio.wav" wav(path) self.assertEqual(len(stt.read_audio(path)), 48000) for rate, frames in [(16000, 24000), (24000, 100), (24000, 24000 * 61)]: wav(path, rate, frames) with self.assertRaises(RuntimeError): stt.read_audio(path) def test_truncated_wav_rejected(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "audio.wav" wav(path) path.write_bytes(path.read_bytes()[:-4]) with self.assertRaisesRegex(RuntimeError, "обрезан"): stt.read_audio(path) def test_credentials_validated_not_mutated(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "auth.json" for mode, token in [("apiKey", "synthetic"), ("chatgpt", "0000"), ("chatgpt", "a\nb"), ("chatgpt", "")]: path.write_text(json.dumps({"auth_mode": mode, "tokens": {"access_token": token}})) before = path.read_bytes() with self.assertRaises(RuntimeError): stt.read_credential(path, None) self.assertEqual(before, path.read_bytes()) path.write_text(json.dumps({"auth_mode": "chatgpt", "tokens": {"access_token": "[скрыто]"}})) self.assertEqual(stt.read_credential(path, None), "synthetic") key = Path(directory) / "key" key.write_text("synthetic-platform\n") self.assertEqual(stt.read_credential(None, key), "synthetic-platform") def run_events(self, events, pcm=b"\x01\x00" * 36000): socket = Socket(events) with patch.dict(sys.modules, {"websocket": self.module(lambda *a, **kw: socket)}): result = stt.transcribe(pcm, "synthetic", "gpt-transcribe", 5) return result, socket def test_flow_chunks_and_completed_transcript(self): result, socket = self.run_events([ {"type": "session.created"}, {"type": "session.updated"}, {"type": "input_audio_buffer.committed", "item_id": "one"}, {"type": "conversation.item.input_audio_transcription.delta", "item_id": "one", "delta": "Провер"}, {"type": "conversation.item.input_audio_transcription.completed", "item_id": "other", "transcript": "не наш"}, {"type": "conversation.item.input_audio_transcription.completed", "item_id": "one", "transcript": "Проверка"}]) self.assertEqual(result, "Проверка") self.assertEqual([x["type"] for x in socket.sent], ["session.update", "input_audio_buffer.append", "input_audio_buffer.append", "input_audio_buffer.commit"]) self.assertEqual(b"".join(base64.b64decode(x["audio"]) for x in socket.sent if "audio" in x), b"\x01\x00" * 36000) self.assertEqual(socket.sent[0]["session"]["type"], "transcription") self.assertTrue(socket.closed) def test_completed_before_commit_ack_preserved(self): result, socket = self.run_events([ {"type": "session.created"}, {"type": "session.updated"}, {"type": "conversation.item.input_audio_transcription.completed", "item_id": "one", "transcript": "Текст"}, {"type": "input_audio_buffer.committed", "item_id": "one"}]) self.assertEqual(result, "Текст") self.assertTrue(socket.closed) def test_failure_and_empty_transcript_close_socket(self): for event in [{"type": "error", "error": {"message": "SECRET"}}, {"type": "conversation.item.input_audio_transcription.failed", "error": "SECRET"}, {"type": "conversation.item.input_audio_transcription.completed", "item_id": "one", "transcript": " "}]: socket = Socket([{"type": "session.created"}, {"type": "session.updated"}, {"type": "input_audio_buffer.committed", "item_id": "one"}, event]) with patch.dict(sys.modules, {"websocket": self.module(lambda *a, **kw: socket)}): with self.assertRaises(RuntimeError) as error: stt.transcribe(b"\x00\x00" * 24000, "synthetic", "gpt-transcribe", 5) self.assertNotIn("SECRET", str(error.exception)) self.assertTrue(socket.closed) def test_handshake_hides_private_body_and_headers(self): def factory(*args, **kwargs): raise HandshakeError("SECRET headers or body") with patch.dict(sys.modules, {"websocket": self.module(factory)}): with self.assertRaisesRegex(RuntimeError, "HTTP 403") as error: stt.transcribe(b"\x00\x00", "synthetic", "gpt-transcribe", 5) self.assertNotIn("SECRET", str(error.exception)) def test_authorization_fixed_host_and_no_proxy(self): def factory(url, **kw): self.assertEqual(url, stt.ENDPOINT) self.assertEqual(kw["header"], ["Authorization: Bearer synthetic"]) self.assertEqual(kw["http_no_proxy"], ["*"]) raise HandshakeError() with patch.dict(sys.modules, {"websocket": self.module(factory)}): with self.assertRaises(RuntimeError): stt.transcribe(b"\x00\x00", "synthetic", "gpt-transcribe", 5) def test_common_deadline_closes_socket(self): socket = Socket([]) with patch.dict(sys.modules, {"websocket": self.module(lambda *a, **kw: socket)}), patch.object(stt.time, "monotonic", side_effect=[0, 6]): with self.assertRaisesRegex(RuntimeError, "timeout"): stt.transcribe(b"\x00\x00", "synthetic", "gpt-transcribe", 5) self.assertTrue(socket.closed) def test_single_file_copy_dry_run_needs_no_auth_or_repo(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) script = root / "realtime-transcribe.py" script.write_bytes((HERE / script.name).read_bytes()) path = root / "audio.wav" wav(path) result = subprocess.run([sys.executable, str(script), str(path), "--auth-file", str(root / "missing"), "--dry-run"], cwd=root, capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(result.stdout) self.assertFalse(payload["credentials_read"]) self.assertFalse(payload["network"]) def test_cli_errors_do_not_dump_auth(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) path = root / "audio.wav" wav(path) auth = root / "auth.json" auth.write_text("SECRET-invalid-json") result = subprocess.run([sys.executable, str(HERE / "realtime-transcribe.py"), str(path), "--auth-file", str(auth)], capture_output=True, text=True) self.assertEqual(result.returncode, 1) self.assertNotIn("SECRET", result.stderr) self.assertEqual(result.stdout, "") def test_invalid_timeout_rejected_offline(self): for value in ["nan", "inf", "0", "-1"]: result = subprocess.run([sys.executable, str(HERE / "realtime-transcribe.py"), "/not/read", "--timeout", value, "--dry-run"], capture_output=True, text=True) self.assertEqual(result.returncode, 2) if __name__ == "__main__": unittest.main()