From 2008d80a9e6266e73e80cd8ff4a4a9a837d63a05 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 28 Jul 2026 09:41:48 -0700 Subject: [PATCH] test(gateway): regression coverage for platform-aware voice delivery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tests/gateway/test_base_auto_tts_output_format.py (new): the base adapter auto-TTS block passes an explicit .ogg output path on every OPUS_VOICE_PLATFORMS member (parametrized from the tts_tool set — the single source of truth), keeps .mp3 on non-opus platforms, honors the tool's success flag, and stays unique/uuid-based. - tests/gateway/test_auto_voice_reply_format.py: runner _send_voice_reply parametrized across Matrix/Feishu/WhatsApp/Signal (.ogg) alongside the existing Telegram/Slack cases; streamed+global-auto-TTS gate regression. - tests/gateway/test_telegram_voice_caption_markdown.py (new): caption MarkdownV2 formatting, entity-rejection plain fallback, overflow skip, and no-caption passthrough (#32029). - test_voice_command.py filename-uuid contract test updated to point at build_auto_tts_output_path (construction moved there). --- tests/gateway/test_auto_voice_reply_format.py | 59 ++++++ .../test_base_auto_tts_output_format.py | 182 ++++++++++++++++++ .../test_telegram_voice_caption_markdown.py | 130 +++++++++++++ tests/gateway/test_voice_command.py | 17 +- 4 files changed, 384 insertions(+), 4 deletions(-) create mode 100644 tests/gateway/test_base_auto_tts_output_format.py create mode 100644 tests/gateway/test_telegram_voice_caption_markdown.py diff --git a/tests/gateway/test_auto_voice_reply_format.py b/tests/gateway/test_auto_voice_reply_format.py index 13dc0218fe12..8a26c8914010 100644 --- a/tests/gateway/test_auto_voice_reply_format.py +++ b/tests/gateway/test_auto_voice_reply_format.py @@ -70,6 +70,65 @@ class TestAutoVoiceReplyFormat: assert requested_paths[0].endswith(".mp3") adapter.send_voice.assert_awaited_once() assert adapter.send_voice.await_args.kwargs["audio_path"].endswith(".mp3") + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "platform", + [Platform.MATRIX, Platform.FEISHU, Platform.WHATSAPP, Platform.SIGNAL], + ) + async def test_opus_platform_auto_voice_reply_requests_ogg(self, platform): + """Every OPUS_VOICE_PLATFORMS member gets an explicit .ogg output path. + + Regression for #14841 (Matrix) / #45557 (Feishu): _send_voice_reply + hardcoded .ogg for Telegram only, so Matrix/Feishu voice replies were + synthesized as MP3 and delivered as plain attachments instead of + native voice bubbles. + """ + runner = _make_runner() + adapter = _make_adapter(platform) + runner.adapters[platform] = adapter + event = _make_event(platform) + requested_paths = [] + + def fake_tts(*, text, output_path): + requested_paths.append(output_path) + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + Path(output_path).write_bytes(b"fake ogg opus") + return json.dumps({ + "success": True, + "file_path": output_path, + "provider": "gemini", + "voice_compatible": True, + }) + + with patch("tools.tts_tool.text_to_speech_tool", side_effect=fake_tts): + await runner._send_voice_reply(event, "hello from auto tts") + + assert requested_paths and requested_paths[0].endswith(".ogg") + adapter.send_voice.assert_awaited_once() + assert adapter.send_voice.await_args.kwargs["audio_path"].endswith(".ogg") + + def test_should_send_voice_reply_streamed_global_auto_tts_fires(self): + """Streamed reply + global voice.auto_tts (no /voice opt-in) sends voice. + + Regression for the #51867/#23983 remainder: when streaming consumed + the text, the base adapter's auto-TTS gets text_content=None, and the + runner path used to consult only self._voice_mode — so a chat relying + purely on the global voice.auto_tts default silently lost its voice + reply. + """ + runner = _make_runner() + adapter = _make_adapter(Platform.TELEGRAM) + adapter._should_auto_tts_for_chat = MagicMock(return_value=True) + runner.adapters[Platform.TELEGRAM] = adapter + voice_event = _make_event( + Platform.TELEGRAM, chat_id="123", message_type=MessageType.VOICE + ) + + assert runner._should_send_voice_reply( + voice_event, "hello", [], already_sent=True + ) is True + def test_should_send_voice_reply_uses_global_auto_tts_adapter_default(self): """voice.auto_tts=true should make normal text replies get voice too.""" runner = _make_runner() diff --git a/tests/gateway/test_base_auto_tts_output_format.py b/tests/gateway/test_base_auto_tts_output_format.py new file mode 100644 index 000000000000..7bef04ed5cbc --- /dev/null +++ b/tests/gateway/test_base_auto_tts_output_format.py @@ -0,0 +1,182 @@ +"""Base-adapter auto-TTS must pass a platform-aware explicit output path. + +Regression tests for the cleared-contextvar bug (#57049, #36685): the +post-handler auto-TTS block in ``BasePlatformAdapter._process_message_background`` +runs AFTER ``_clear_session_env`` wiped ``HERMES_SESSION_PLATFORM``, so the +TTS tool's contextvar-based ``want_opus`` detection always resolved False on +that path and Opus platforms received MP3 (audio attachment, not a native +voice bubble). The fix passes an explicit output path from +``build_auto_tts_output_path(platform)``, which consults the TTS tool's +``OPUS_VOICE_PLATFORMS`` set — the single source of truth. +""" + +import asyncio +import json +from unittest.mock import AsyncMock, patch + +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import ( + BasePlatformAdapter, + MessageEvent, + MessageType, + SendResult, + build_auto_tts_output_path, +) +from gateway.session import SessionSource, build_session_key +from tools.tts_tool import OPUS_VOICE_PLATFORMS + + +class _DummyAdapter(BasePlatformAdapter): + def __init__(self, platform: Platform): + super().__init__(PlatformConfig(enabled=True, token="fake-token"), platform) + self.sent = [] + + async def connect(self, *, is_reconnect: bool = False) -> bool: + return True + + async def disconnect(self) -> None: + return None + + async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult: + self.sent.append({"chat_id": chat_id, "content": content}) + return SendResult(success=True, message_id="1") + + async def send_typing(self, chat_id: str, metadata=None) -> None: + return None + + async def stop_typing(self, chat_id: str, metadata=None) -> None: + return None + + async def get_chat_info(self, chat_id: str): + return {"id": chat_id} + + +def _make_voice_event(platform: Platform) -> MessageEvent: + return MessageEvent( + text="hello", + message_type=MessageType.VOICE, + source=SessionSource( + platform=platform, + chat_id="-1001", + chat_type="group", + ), + message_id="voice-1", + ) + + +def _hold_typing(): + async def hold(*_args, **_kwargs): + await asyncio.Event().wait() + + return hold + + +# --------------------------------------------------------------------------- +# build_auto_tts_output_path: OPUS_VOICE_PLATFORMS is the single source of truth +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("platform_name", sorted(OPUS_VOICE_PLATFORMS)) +def test_output_path_is_ogg_for_every_opus_voice_platform(platform_name): + path = build_auto_tts_output_path(platform_name) + assert path.endswith(".ogg"), path + + +@pytest.mark.parametrize( + "platform", [Platform.DISCORD, Platform.SLACK, "irc", None] +) +def test_output_path_is_mp3_for_non_opus_platforms(platform): + path = build_auto_tts_output_path(platform) + assert path.endswith(".mp3"), path + + +def test_output_path_accepts_platform_enum(): + assert build_auto_tts_output_path(Platform.TELEGRAM).endswith(".ogg") + assert build_auto_tts_output_path(Platform.MATRIX).endswith(".ogg") + assert build_auto_tts_output_path(Platform.FEISHU).endswith(".ogg") + + +def test_output_paths_are_unique(): + assert build_auto_tts_output_path("telegram") != build_auto_tts_output_path("telegram") + + +# --------------------------------------------------------------------------- +# Base-adapter auto-TTS block: explicit output_path, no contextvar reliance +# --------------------------------------------------------------------------- + +async def _run_auto_tts(adapter: _DummyAdapter, platform: Platform): + adapter._keep_typing = _hold_typing() + adapter._should_auto_tts_for_chat = lambda _chat_id: True + adapter.play_tts = AsyncMock(return_value=SendResult(success=True, message_id="tts-1")) + long_reply = "x" * 2000 # avoid the telegram caption-collapse path + adapter.set_message_handler(lambda _event: asyncio.sleep(0, result=long_reply)) + event = _make_voice_event(platform) + requested = [] + + def fake_tts(*, text, output_path=None): + requested.append(output_path) + from pathlib import Path + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + Path(output_path).write_bytes(b"fake audio") + return json.dumps({"success": True, "file_path": output_path}) + + with patch("tools.tts_tool.check_tts_requirements", return_value=True), patch( + "tools.tts_tool.text_to_speech_tool", side_effect=fake_tts + ): + await adapter._process_message_background( + event, build_session_key(event.source) + ) + return requested, adapter + + +@pytest.mark.asyncio +async def test_base_auto_tts_requests_ogg_on_opus_platform(): + """Telegram (opus platform) gets an explicit .ogg path even though the + HERMES_SESSION_PLATFORM contextvar is cleared by the time the block runs.""" + adapter = _DummyAdapter(Platform.TELEGRAM) + requested, adapter = await _run_auto_tts(adapter, Platform.TELEGRAM) + + assert requested and requested[0] is not None + assert requested[0].endswith(".ogg") + adapter.play_tts.assert_awaited_once() + assert adapter.play_tts.await_args.kwargs["audio_path"].endswith(".ogg") + + +@pytest.mark.asyncio +async def test_base_auto_tts_keeps_mp3_on_non_opus_platform(): + adapter = _DummyAdapter(Platform.DISCORD) + requested, adapter = await _run_auto_tts(adapter, Platform.DISCORD) + + assert requested and requested[0] is not None + assert requested[0].endswith(".mp3") + adapter.play_tts.assert_awaited_once() + assert adapter.play_tts.await_args.kwargs["audio_path"].endswith(".mp3") + + +@pytest.mark.asyncio +async def test_base_auto_tts_skips_playback_when_tool_reports_failure(): + """A success=False tool result must not deliver a stale/partial file.""" + adapter = _DummyAdapter(Platform.TELEGRAM) + adapter._keep_typing = _hold_typing() + adapter._should_auto_tts_for_chat = lambda _chat_id: True + adapter.play_tts = AsyncMock(return_value=SendResult(success=True, message_id="tts-1")) + adapter.set_message_handler(lambda _event: asyncio.sleep(0, result="reply text")) + event = _make_voice_event(Platform.TELEGRAM) + + def fake_tts(*, text, output_path=None): + from pathlib import Path + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + Path(output_path).write_bytes(b"partial") + return json.dumps({"success": False, "error": "backend exploded"}) + + with patch("tools.tts_tool.check_tts_requirements", return_value=True), patch( + "tools.tts_tool.text_to_speech_tool", side_effect=fake_tts + ): + await adapter._process_message_background( + event, build_session_key(event.source) + ) + + adapter.play_tts.assert_not_awaited() + # Text reply still goes out. + assert adapter.sent and adapter.sent[0]["content"] == "reply text" diff --git a/tests/gateway/test_telegram_voice_caption_markdown.py b/tests/gateway/test_telegram_voice_caption_markdown.py new file mode 100644 index 000000000000..c4f50052dae9 --- /dev/null +++ b/tests/gateway/test_telegram_voice_caption_markdown.py @@ -0,0 +1,130 @@ +"""Telegram voice-message captions must render markdown (#32029). + +``TelegramAdapter.send_voice`` used to pass captions raw with no +``parse_mode``, so auto-TTS captions carrying the agent's markdown reply +showed literal *asterisks*, backticks and [links](...). It now formats the +caption to MarkdownV2 (when the formatted text fits the 1024-char caption +cap) and falls back to the plain truncated caption when the Bot API rejects +the entities or formatting overflows. +""" +import sys +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import PlatformConfig + + +def _ensure_telegram_mock(): + if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): + return + mod = MagicMock() + mod.error.NetworkError = type("NetworkError", (OSError,), {}) + mod.error.TimedOut = type("TimedOut", (OSError,), {}) + mod.error.BadRequest = type("BadRequest", (Exception,), {}) + for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): + sys.modules.setdefault(name, mod) + sys.modules.setdefault("telegram.error", mod.error) + + +_ensure_telegram_mock() + +from plugins.platforms.telegram import adapter as telegram_mod # noqa: E402 +from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 + + +def _make_adapter() -> TelegramAdapter: + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***")) + adapter._bot = MagicMock() + adapter._bot.send_voice = AsyncMock(return_value=MagicMock(message_id=1)) + adapter._bot.send_audio = AsyncMock(return_value=MagicMock(message_id=2)) + return adapter + + +def _write_ogg(tmp_path): + audio = tmp_path / "reply.ogg" + audio.write_bytes(b"\x00" * 16) + return audio + + +@pytest.mark.asyncio +async def test_voice_caption_gets_markdown_parse_mode(monkeypatch, tmp_path): + """A markdown caption is MarkdownV2-formatted and sent with parse_mode.""" + monkeypatch.setattr( + telegram_mod, "_probe_voice_duration_seconds", lambda _p: 3 + ) + adapter = _make_adapter() + + result = await adapter.send_voice( + "123", str(_write_ogg(tmp_path)), caption="*bold* reply" + ) + + assert result.success is True + adapter._bot.send_voice.assert_awaited_once() + kwargs = adapter._bot.send_voice.await_args.kwargs + assert kwargs["parse_mode"] is not None + # format_message converts *bold* to MarkdownV2 bold; the raw caption must + # have gone through formatting rather than being passed verbatim. + assert kwargs["caption"] == adapter.format_message("*bold* reply") + + +@pytest.mark.asyncio +async def test_voice_caption_falls_back_to_plain_on_entity_rejection( + monkeypatch, tmp_path +): + """A Bot API entity-parse rejection retries with the plain caption.""" + monkeypatch.setattr( + telegram_mod, "_probe_voice_duration_seconds", lambda _p: 3 + ) + adapter = _make_adapter() + adapter._bot.send_voice = AsyncMock( + side_effect=[ + Exception("Bad Request: can't parse entities"), + MagicMock(message_id=7), + ] + ) + + result = await adapter.send_voice( + "123", str(_write_ogg(tmp_path)), caption="*bold* reply" + ) + + assert result.success is True + assert adapter._bot.send_voice.await_count == 2 + retry_kwargs = adapter._bot.send_voice.await_args_list[1].kwargs + assert retry_kwargs["parse_mode"] is None + assert retry_kwargs["caption"] == "*bold* reply" + + +@pytest.mark.asyncio +async def test_voice_caption_overflow_skips_formatting(monkeypatch, tmp_path): + """When the formatted caption exceeds 1024 UTF-16 units, send plain + (truncated) — MarkdownV2 escaping inflates length, and a truncated + formatted caption could cut an entity in half.""" + monkeypatch.setattr( + telegram_mod, "_probe_voice_duration_seconds", lambda _p: 3 + ) + adapter = _make_adapter() + # Enough special chars that escaping pushes the formatted text over 1024. + caption = ("hello. " * 160)[:1020] + + result = await adapter.send_voice("123", str(_write_ogg(tmp_path)), caption=caption) + + assert result.success is True + kwargs = adapter._bot.send_voice.await_args.kwargs + assert kwargs["parse_mode"] is None + assert kwargs["caption"] == caption[:1024] + + +@pytest.mark.asyncio +async def test_voice_without_caption_unchanged(monkeypatch, tmp_path): + monkeypatch.setattr( + telegram_mod, "_probe_voice_duration_seconds", lambda _p: 3 + ) + adapter = _make_adapter() + + result = await adapter.send_voice("123", str(_write_ogg(tmp_path))) + + assert result.success is True + kwargs = adapter._bot.send_voice.await_args.kwargs + assert kwargs["caption"] is None + assert kwargs["parse_mode"] is None diff --git a/tests/gateway/test_voice_command.py b/tests/gateway/test_voice_command.py index 539648886daf..6f5ca0164143 100644 --- a/tests/gateway/test_voice_command.py +++ b/tests/gateway/test_voice_command.py @@ -1872,14 +1872,23 @@ class TestSendVoiceReplyFilename: """_send_voice_reply uses uuid for unique filenames.""" def test_filename_uses_uuid(self): - """The method uses uuid in the filename, not time-based.""" + """The path builder uses uuid in the filename, not time-based. + + Filename construction moved into build_auto_tts_output_path + (gateway/platforms/base.py) when the path became platform-aware; + the uniqueness contract lives there now. + """ import inspect + from gateway.platforms.base import build_auto_tts_output_path from gateway.run import GatewayRunner - source = inspect.getsource(GatewayRunner._send_voice_reply) + source = inspect.getsource(build_auto_tts_output_path) assert "uuid" in source, \ - "_send_voice_reply should use uuid for unique filenames" + "build_auto_tts_output_path should use uuid for unique filenames" assert "int(time.time())" not in source, \ - "_send_voice_reply should not use int(time.time()) — collision risk" + "build_auto_tts_output_path should not use int(time.time()) — collision risk" + runner_source = inspect.getsource(GatewayRunner._send_voice_reply) + assert "build_auto_tts_output_path" in runner_source, \ + "_send_voice_reply should build its path via build_auto_tts_output_path" def test_filenames_are_unique(self): """Two calls produce different filenames."""