""" test_bot_tts.py — TTS: Fish Audio SSE и inline_tts (квоты, фолбэк на Gemini). Выделено из test_bot.py (P2 аудита); общие фейки — в bot_test_helpers.py. """ from unittest.mock import MagicMock, AsyncMock import asyncio import base64 from types import SimpleNamespace import bot import pytest from tests.bot_test_helpers import ( _FakeIncomingMessage, _FakeSSEResponse, _FakeSessionForSSE, _FakeVoiceBot, _fake_tts_response, ) def test_inline_tts_records_quota_usage_on_success(): # У TTS всего 10 запросов/сутки на модель (дашборд) — расход учитываем в GLOBAL_QUOTA["gemini"] как у текстовых. # Минимальный RIFF/WAV-заголовок — достаточно, чтобы код распознал формат как # WAV и не пытался обернуть его заново через pcm_to_wav; реальная конвертация # через ffmpeg в этом окружении не установлена и ожидаемо упадёт — это штатно # ловится внутри inline_tts (тест проверяет учёт квоты, а не качество звука). fake_wav_bytes = b"RIFF" + b"\x00" * 4 + b"WAVEfmt " + b"\x00" * 64 def fake_generate_content(*, model, contents, config=None): return _fake_tts_response(fake_wav_bytes) fake_client = MagicMock() # Асинхронный путь (lumen_tts): синхронный вызов в потоке убран — зависший # Google держал лок чата бесконечно (аудит 26.09.2026). fake_client.aio.models.generate_content = AsyncMock(side_effect=fake_generate_content) incoming = _FakeIncomingMessage(999801) incoming.message_id = 12345 # inline_tts использует его для reply_to_message_id original_client = bot.client original_bot = bot.bot bot.client = fake_client bot.bot = _FakeVoiceBot() bot.GLOBAL_QUOTA["gemini"].pop("gemini-3.1-flash-tts-preview", None) try: asyncio.run(bot.inline_tts(incoming, "Привет, мир")) entry = bot.GLOBAL_QUOTA["gemini"].get("gemini-3.1-flash-tts-preview") assert entry is not None assert entry["used"] >= 1 finally: bot.client = original_client bot.bot = original_bot bot.GLOBAL_QUOTA["gemini"].pop("gemini-3.1-flash-tts-preview", None) def test_inline_tts_marks_quota_exhausted_on_daily_quota(): # Суточная квота первой TTS-модели — метка до утра через _mark_quota_exhausted, # а синтез продолжается со второй моделью цепочки. class _RateLimitExc(Exception): status_code = 429 fake_wav_bytes = b"RIFF" + b"\x00" * 4 + b"WAVEfmt " + b"\x00" * 64 calls = [] def fake_generate_content(*, model, contents, config=None): calls.append(model) if model == "gemini-3.1-flash-tts-preview": raise _RateLimitExc("Quota exceeded for quota metric 'GenerateRequestsPerDayPerProjectPerModel'") return _fake_tts_response(fake_wav_bytes) fake_client = MagicMock() fake_client.aio.models.generate_content = AsyncMock(side_effect=fake_generate_content) incoming = _FakeIncomingMessage(999802) incoming.message_id = 12346 original_client = bot.client original_bot = bot.bot bot.client = fake_client bot.bot = _FakeVoiceBot() bot.GLOBAL_QUOTA["gemini"].pop("gemini-3.1-flash-tts-preview", None) bot.GLOBAL_QUOTA["gemini"].pop("gemini-2.5-flash-preview-tts", None) try: asyncio.run(bot.inline_tts(incoming, "Привет, мир")) assert calls == ["gemini-3.1-flash-tts-preview", "gemini-2.5-flash-preview-tts"] exhausted = bot.GLOBAL_QUOTA["gemini"].get("gemini-3.1-flash-tts-preview") assert exhausted is not None and exhausted.get("exhausted_at") is not None succeeded = bot.GLOBAL_QUOTA["gemini"].get("gemini-2.5-flash-preview-tts") assert succeeded is not None and succeeded["used"] >= 1 finally: bot.client = original_client bot.bot = original_bot bot.GLOBAL_QUOTA["gemini"].pop("gemini-3.1-flash-tts-preview", None) bot.GLOBAL_QUOTA["gemini"].pop("gemini-2.5-flash-preview-tts", None) def test_inline_tts_burst_rate_limit_only_cools_down(): # Минутный всплеск у первой TTS-модели не должен ставить суточную метку: # раньше любой 429 убирал её до полуночи, хотя текстовый маршрут уже получил # короткую остывку. Синтез всё равно продолжается со второй моделью. import time as _time class _RateLimitExc(Exception): status_code = 429 fake_wav_bytes = b"RIFF" + b"\x00" * 4 + b"WAVEfmt " + b"\x00" * 64 calls = [] def fake_generate_content(*, model, contents, config=None): calls.append(model) if model == "gemini-3.1-flash-tts-preview": raise _RateLimitExc("rate limit exceeded") return _fake_tts_response(fake_wav_bytes) fake_client = MagicMock() fake_client.aio.models.generate_content = AsyncMock(side_effect=fake_generate_content) incoming = _FakeIncomingMessage(999803) incoming.message_id = 12347 original_client = bot.client original_bot = bot.bot bot.client = fake_client bot.bot = _FakeVoiceBot() bot.GLOBAL_QUOTA["gemini"].pop("gemini-3.1-flash-tts-preview", None) bot.GLOBAL_QUOTA["gemini"].pop("gemini-2.5-flash-preview-tts", None) try: asyncio.run(bot.inline_tts(incoming, "Привет, мир")) assert calls == ["gemini-3.1-flash-tts-preview", "gemini-2.5-flash-preview-tts"] entry = bot.GLOBAL_QUOTA["gemini"].get("gemini-3.1-flash-tts-preview") assert entry is not None and entry.get("exhausted_at") is None assert entry.get("cooldown_until", 0) > _time.time() finally: bot.client = original_client bot.bot = original_bot bot.GLOBAL_QUOTA["gemini"].pop("gemini-3.1-flash-tts-preview", None) bot.GLOBAL_QUOTA["gemini"].pop("gemini-2.5-flash-preview-tts", None) def test_inline_tts_skips_fish_audio_when_disabled(monkeypatch): # FISH_AUDIO_ENABLED=False (аудит моделей, 17.09.2026 — зеркало снято с # бесплатного каталога): inline_tts не должен вообще трогать _fish_audio_tts_bytes. async def _fail_if_called(text): raise AssertionError("fish attempt must be skipped while disabled") monkeypatch.setattr(bot, "_fish_audio_tts_bytes", _fail_if_called) fake_wav_bytes = b"RIFF" + b"\x00" * 4 + b"WAVEfmt " + b"\x00" * 64 def fake_generate_content(*, model, contents, config=None): return _fake_tts_response(fake_wav_bytes) fake_client = MagicMock() # Асинхронный путь (lumen_tts): синхронный вызов в потоке убран — зависший # Google держал лок чата бесконечно (аудит 26.09.2026). fake_client.aio.models.generate_content = AsyncMock(side_effect=fake_generate_content) incoming = _FakeIncomingMessage(999803) incoming.message_id = 12347 original_client = bot.client original_bot = bot.bot bot.client = fake_client bot.bot = _FakeVoiceBot() try: asyncio.run(bot.inline_tts(incoming, "Привет, мир")) assert bot.bot.sent_voice is not None finally: bot.client = original_client bot.bot = original_bot def test_fish_audio_tts_bytes_parses_sse_audio_chunks(): raw_audio = b"fake-mp3-bytes" b64_whole = base64.b64encode(raw_audio).decode("ascii") half = len(b64_whole) // 2 lines = [ ('data: {"choices":[{"delta":{"audio":{"data":"' + b64_whole[:half] + '"}}}]}\n').encode("utf-8"), ('data: {"choices":[{"delta":{"audio":{"data":"' + b64_whole[half:] + '"}}}]}\n').encode("utf-8"), b"data: [DONE]\n", ] fake_resp = _FakeSSEResponse(lines) fake_session = _FakeSessionForSSE(fake_resp) async def fake_get_http_session(): return fake_session original_get_session = bot._get_http_session original_key = bot.OPENROUTER_API_KEY bot._get_http_session = fake_get_http_session bot.OPENROUTER_API_KEY = "fake-key" try: result = asyncio.run(bot._fish_audio_tts_bytes("Привет, мир")) assert result == raw_audio finally: bot._get_http_session = original_get_session bot.OPENROUTER_API_KEY = original_key def test_fish_audio_tts_bytes_returns_none_on_http_error(): fake_resp = _FakeSSEResponse([], status=500) fake_session = _FakeSessionForSSE(fake_resp) async def fake_get_http_session(): return fake_session original_get_session = bot._get_http_session original_key = bot.OPENROUTER_API_KEY bot._get_http_session = fake_get_http_session bot.OPENROUTER_API_KEY = "fake-key" try: assert asyncio.run(bot._fish_audio_tts_bytes("Привет")) is None finally: bot._get_http_session = original_get_session bot.OPENROUTER_API_KEY = original_key def test_fish_audio_tts_bytes_returns_none_without_api_key(): original_key = bot.OPENROUTER_API_KEY bot.OPENROUTER_API_KEY = "" try: assert asyncio.run(bot._fish_audio_tts_bytes("Привет")) is None finally: bot.OPENROUTER_API_KEY = original_key def test_gemini_tts_synthesis_has_timeout_and_does_not_use_threads(): # Регрессия (аудит 26.09.2026): синхронный вызов в asyncio.to_thread без # таймаута — зависший Google держал лок чата бесконечно и терял поток. import asyncio as _asyncio from lumen_tts import _gemini_tts_bytes as synth calls = [] class _HangingAio: class models: @staticmethod async def generate_content(**kwargs): calls.append(kwargs) await _asyncio.sleep(30) client = SimpleNamespace(aio=_HangingAio) def never(*args, **kwargs): raise AssertionError("sync client path must not be used") client.models = SimpleNamespace(generate_content=never) with pytest.raises(Exception): _asyncio.run(synth( client, "текст", tts_models=["m1"], is_rate_limit_error=lambda e: False, on_model_exhausted=lambda m: None, on_model_success=lambda m: None, request_timeout_sec=0.3, )) assert calls, "async client must be used" def test_gemini_tts_passes_http_timeout_to_sdk(): # Второй рубеж: таймаут задан и в http_options, даже если внешний wait_for не сработал. import asyncio as _asyncio from lumen_tts import _gemini_tts_bytes as synth seen = {} def fake_generate_content(*, model, contents, config=None): seen["config"] = config return _fake_tts_response(b"RIFF" + b"\x00" * 4 + b"WAVEfmt " + b"\x00" * 64) class _Aio: class models: @staticmethod async def generate_content(**kwargs): return fake_generate_content(model=kwargs["model"], contents=kwargs["contents"], config=kwargs["config"]) _asyncio.run(synth( SimpleNamespace(aio=_Aio), "текст", tts_models=["m1"], is_rate_limit_error=lambda e: False, on_model_exhausted=lambda m: None, on_model_success=lambda m: None, request_timeout_sec=7.0, )) http_options = getattr(seen["config"], "http_options", None) assert http_options is not None and http_options.timeout == 7000 def test_fish_audio_tts_bytes_returns_none_on_empty_stream(): # Поток отдал валидный SSE, но ни одного audio-чанка (например, если формат # ответа модели когда-нибудь изменится) — должны тихо откатиться на Gemini, # а не упасть с исключением или вернуть пустые байты как будто это успех. lines = [b"data: [DONE]\n"] fake_resp = _FakeSSEResponse(lines) fake_session = _FakeSessionForSSE(fake_resp) async def fake_get_http_session(): return fake_session original_get_session = bot._get_http_session original_key = bot.OPENROUTER_API_KEY bot._get_http_session = fake_get_http_session bot.OPENROUTER_API_KEY = "fake-key" try: assert asyncio.run(bot._fish_audio_tts_bytes("Привет")) is None finally: bot._get_http_session = original_get_session bot.OPENROUTER_API_KEY = original_key class _CountingVoiceBot(_FakeVoiceBot): def __init__(self): super().__init__() self.voices = [] async def send_voice(self, **kwargs): self.voices.append(kwargs) return SimpleNamespace() def _wav_client(): fake_wav_bytes = b"RIFF" + b"\x00" * 4 + b"WAVEfmt " + b"\x00" * 64 def fake_generate_content(*, model, contents, config=None): return _fake_tts_response(fake_wav_bytes) fake_client = MagicMock() # Асинхронный путь (lumen_tts): синхронный вызов в потоке убран — зависший # Google держал лок чата бесконечно (аудит 26.09.2026). fake_client.aio.models.generate_content = AsyncMock(side_effect=fake_generate_content) return fake_client def test_split_tts_chunks_short_text_unchanged(): from lumen_commands import _split_tts_chunks assert _split_tts_chunks("Привет, мир", 800) == ["Привет, мир"] assert _split_tts_chunks(" ", 800) == [] def test_split_tts_chunks_packs_sentences_within_limit(): from lumen_commands import _split_tts_chunks s1 = "Первое предложение про котиков. " s2 = "Второе предложение про погоду за окном. " s3 = "Третье предложение про смысл жизни и всё такое." chunks = _split_tts_chunks(s1 + s2 + s3, 80) assert all(len(c) <= 80 for c in chunks) assert len(chunks) == 2 assert chunks[0].startswith("Первое") and "Второе" in chunks[0] assert chunks[1].startswith("Третье") def test_split_tts_chunks_hard_cuts_single_long_sentence(): from lumen_commands import _split_tts_chunks long_single = "Слово " * 200 chunks = _split_tts_chunks(long_single.strip(), 800) assert all(len(c) <= 800 for c in chunks) assert len(chunks) >= 2 def test_inline_tts_sends_long_text_in_parts(): # Длинный текст бьём на части вместо отказа: два голосовых, первый в ответ. from lumen_commands import TTS_MAX_PARTS assert TTS_MAX_PARTS >= 2 limit = bot.TTS_MAX_CHARS part = "Предложение номер раз про интересные вещи. " text = part * ((limit // len(part)) + 2) assert len(text) > limit incoming = _FakeIncomingMessage(999804) incoming.message_id = 12348 original_client = bot.client original_bot = bot.bot bot.client = _wav_client() counting = _CountingVoiceBot() bot.bot = counting try: asyncio.run(bot.inline_tts(incoming, text)) assert len(counting.voices) == 2 assert counting.voices[0].get("reply_to_message_id") == 12348 assert counting.voices[1].get("reply_to_message_id") is None finally: bot.client = original_client bot.bot = original_bot def test_inline_tts_refuses_beyond_parts_cap_without_sending(): from lumen_commands import TTS_MAX_PARTS limit = bot.TTS_MAX_CHARS text = "Слово " * ((limit * TTS_MAX_PARTS) // 6 + 10) incoming = _FakeIncomingMessage(999805) incoming.message_id = 12349 original_client = bot.client original_bot = bot.bot bot.client = _wav_client() counting = _CountingVoiceBot() bot.bot = counting try: asyncio.run(bot.inline_tts(incoming, text)) assert counting.voices == [] assert len(incoming.sent) >= 1 # текст tts_too_long ушёл finally: bot.client = original_client bot.bot = original_bot def test_pcm_to_wav_rejects_non_wav_riff(): # Аудит A6-11: префиксу RIFF верили вслепую — у WEBP/AVI он тоже есть. import lumen_tts out = lumen_tts.pcm_to_wav(b"RIFF....WEBP....") assert out.startswith(b"RIFF") and out[8:12] == b"WAVE" assert lumen_tts.pcm_to_wav(b"RIFFxxxxWAVE....") == b"RIFFxxxxWAVE...." def test_inline_tts_zero_budget_never_synthesizes(monkeypatch): # Аудит D2: нулевой общий бюджет — синтез не стартует вообще, сразу честная ошибка. import lumen_commands async def must_not_synthesize(text): raise AssertionError("synthesis must not start on zero budget") monkeypatch.setattr(bot, "TTS_TOTAL_BUDGET_SEC", 0) monkeypatch.setattr(lumen_commands, "_synthesize_tts_voice", must_not_synthesize) incoming = _FakeIncomingMessage(999986) incoming.message_id = 12350 original_bot = bot.bot bot.bot = _FakeVoiceBot() try: asyncio.run(bot.inline_tts(incoming, "Привет, мир")) shown = [text for msg in incoming.sent for text, _ in msg.edits] assert shown, "пользователь должен увидеть ошибку, а не тишину" finally: bot.bot = original_bot bot.chat_state.pop(999986, None) def test_inline_tts_sends_partial_with_shortened_note(monkeypatch): # Аудит D2: второй кусок завис — первый уходит, пользователь узнаёт об обрезке. import lumen_commands class _CountingVoiceBot(_FakeVoiceBot): def __init__(self): super().__init__() self.voice_count = 0 async def send_voice(self, **kwargs): self.voice_count += 1 return await super().send_voice(**kwargs) calls = {"n": 0} async def flaky_synth(text): calls["n"] += 1 if calls["n"] > 1: await asyncio.sleep(30) return (b"audio1", "speech.ogg", 5) sent = {} async def fake_reply(message, text, **kwargs): sent["text"] = text monkeypatch.setattr(bot, "TTS_TOTAL_BUDGET_SEC", 0.3) monkeypatch.setattr(lumen_commands, "_synthesize_tts_voice", flaky_synth) monkeypatch.setattr(bot, "_safe_reply", fake_reply) voice_bot = _CountingVoiceBot() monkeypatch.setattr(bot, "bot", voice_bot) incoming = _FakeIncomingMessage(999987) incoming.message_id = 12351 bot.get_state(999987)["lang"] = "ru" try: asyncio.run(bot.inline_tts(incoming, "x" * 900)) assert voice_bot.voice_count == 1 assert "сокращ" in sent.get("text", "") finally: bot.chat_state.pop(999987, None) def test_inline_tts_refuses_when_user_tts_limit_exhausted(monkeypatch): # Новое: исчерпанный TTS-лимит — отказ до синтеза, модель не вызывается. incoming = _FakeIncomingMessage(999988) incoming.message_id = 12352 incoming.from_user = SimpleNamespace(id=777013) monkeypatch.setattr(bot, "DAILY_USER_TTS_LIMIT", 1) bot._record_user_daily(777013, tts=True) async def fail_generate(*args, **kwargs): raise AssertionError("synthesis must not run for an exhausted user") fake_client = MagicMock() fake_client.aio.models.generate_content = AsyncMock(side_effect=fail_generate) monkeypatch.setattr(bot, "client", fake_client) monkeypatch.setattr(bot, "bot", _FakeVoiceBot()) replies = [] async def fake_safe_reply(msg, text, **kwargs): replies.append(text) monkeypatch.setattr(bot, "_safe_reply", fake_safe_reply) try: asyncio.run(bot.inline_tts(incoming, "Привет, мир")) assert len(replies) == 1 and "1/1" in replies[0] # Отказ не считается: счётчик не вырос. assert bot._user_daily_entry(777013)["tts"] == 1 finally: bot.GLOBAL_QUOTA.get("user_daily", {}).pop("777013", None) bot.chat_state.pop(999988, None) def test_inline_tts_records_user_daily_on_success(monkeypatch): # Новое: успешная озвучка растит total+tts (пара к test_inline_tts_records_quota_usage_on_success). fake_wav_bytes = b"RIFF" + b"\x00" * 4 + b"WAVEfmt " + b"\x00" * 64 def fake_generate_content(*, model, contents, config=None): return _fake_tts_response(fake_wav_bytes) fake_client = MagicMock() fake_client.aio.models.generate_content = AsyncMock(side_effect=fake_generate_content) incoming = _FakeIncomingMessage(999989) incoming.message_id = 12353 incoming.from_user = SimpleNamespace(id=777014) monkeypatch.setattr(bot, "client", fake_client) monkeypatch.setattr(bot, "bot", _FakeVoiceBot()) try: asyncio.run(bot.inline_tts(incoming, "Привет, мир")) entry = bot.GLOBAL_QUOTA.get("user_daily", {}).get("777014") assert entry is not None assert entry["total"] == 1 and entry["tts"] == 1 and entry["gemini"] == 0 finally: bot.GLOBAL_QUOTA.get("user_daily", {}).pop("777014", None) bot.chat_state.pop(999989, None) bot.GLOBAL_QUOTA["gemini"].pop("gemini-3.1-flash-tts-preview", None) def test_inline_tts_charges_each_synthesized_chunk(): # Каждый чанк — отдельный синтез за квоту: списываем поштучно, а не раз за сообщение. limit = bot.TTS_MAX_CHARS part = "Предложение номер раз про интересные вещи. " text = part * ((limit // len(part)) + 2) assert len(text) > limit incoming = _FakeIncomingMessage(999810) incoming.message_id = 12360 incoming.from_user = SimpleNamespace(id=777015) original_client = bot.client original_bot = bot.bot bot.client = _wav_client() counting = _CountingVoiceBot() bot.bot = counting try: asyncio.run(bot.inline_tts(incoming, text)) assert len(counting.voices) == 2 entry = bot.GLOBAL_QUOTA.get("user_daily", {}).get("777015") assert entry is not None assert entry["tts"] == 2 and entry["total"] == 2 finally: bot.client = original_client bot.bot = original_bot bot.GLOBAL_QUOTA.get("user_daily", {}).pop("777015", None) bot.chat_state.pop(999810, None) def test_inline_tts_refuses_when_chunks_would_exceed_limit(monkeypatch): # Проверка считает все чанки сразу: 4/5 + текст на 3 чанка — отказ до синтеза. import lumen_commands from types import SimpleNamespace limit = bot.TTS_MAX_CHARS part = "Предложение номер раз про интересные вещи. " text = part * ((limit // len(part)) + 2) assert len(text) > limit incoming = _FakeIncomingMessage(999811) incoming.message_id = 12361 incoming.from_user = SimpleNamespace(id=777016) monkeypatch.setattr(bot, "DAILY_USER_TTS_LIMIT", 5) for _ in range(4): bot._record_user_daily(777016, tts=True) async def fail_synth(chunk_text): raise AssertionError("synthesis must not run when chunks exceed the limit") monkeypatch.setattr(lumen_commands, "_synthesize_tts_voice", fail_synth) replies = [] async def fake_safe_reply(msg, reply_text, **kwargs): replies.append(reply_text) monkeypatch.setattr(bot, "_safe_reply", fake_safe_reply) try: asyncio.run(bot.inline_tts(incoming, text)) assert len(replies) == 1 assert bot._user_daily_entry(777016)["tts"] == 4 finally: bot.GLOBAL_QUOTA.get("user_daily", {}).pop("777016", None) bot.chat_state.pop(999811, None)