test_tts_media_routing.py
1 """ 2 Tests for cross-platform audio/voice media routing. 3 4 These tests pin the expected delivery path for audio media files across 5 Telegram (where Bot-API sendAudio only accepts MP3/M4A and .ogg/.opus 6 only renders as a voice bubble when explicitly flagged) and via 7 ``GatewayRunner._deliver_media_from_response``. 8 """ 9 10 from types import SimpleNamespace 11 from unittest.mock import AsyncMock 12 13 import pytest 14 15 from gateway.config import Platform, PlatformConfig 16 from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType, SendResult 17 from gateway.run import GatewayRunner 18 from gateway.session import SessionSource, build_session_key 19 20 21 class _MediaRoutingAdapter(BasePlatformAdapter): 22 def __init__(self): 23 super().__init__(PlatformConfig(enabled=True, token="test"), Platform.TELEGRAM) 24 25 async def connect(self): 26 return True 27 28 async def disconnect(self): 29 pass 30 31 async def send(self, chat_id, content=None, **kwargs): 32 return SendResult(success=True, message_id="text") 33 34 async def get_chat_info(self, chat_id): 35 return {"id": chat_id, "type": "dm"} 36 37 38 def _event(thread_id=None): 39 source = SessionSource( 40 platform=Platform.TELEGRAM, 41 chat_id="chat-1", 42 chat_type="dm", 43 thread_id=thread_id, 44 ) 45 return MessageEvent( 46 text="make speech", 47 message_type=MessageType.TEXT, 48 source=source, 49 message_id="msg-1", 50 ) 51 52 53 @pytest.mark.asyncio 54 async def test_base_adapter_routes_telegram_flac_media_tag_to_document_sender(): 55 adapter = _MediaRoutingAdapter() 56 event = _event() 57 adapter._message_handler = AsyncMock(return_value="MEDIA:/tmp/speech.flac") 58 adapter.send_voice = AsyncMock(return_value=SendResult(success=True, message_id="voice")) 59 adapter.send_document = AsyncMock(return_value=SendResult(success=True, message_id="doc")) 60 61 await adapter._process_message_background(event, build_session_key(event.source)) 62 63 adapter.send_document.assert_awaited_once_with( 64 chat_id="chat-1", 65 file_path="/tmp/speech.flac", 66 metadata=None, 67 ) 68 adapter.send_voice.assert_not_awaited() 69 70 71 @pytest.mark.asyncio 72 async def test_base_adapter_routes_non_voice_telegram_ogg_media_tag_to_document_sender(): 73 adapter = _MediaRoutingAdapter() 74 event = _event() 75 adapter._message_handler = AsyncMock(return_value="MEDIA:/tmp/speech.ogg") 76 adapter.send_voice = AsyncMock(return_value=SendResult(success=True, message_id="voice")) 77 adapter.send_document = AsyncMock(return_value=SendResult(success=True, message_id="doc")) 78 79 await adapter._process_message_background(event, build_session_key(event.source)) 80 81 adapter.send_document.assert_awaited_once_with( 82 chat_id="chat-1", 83 file_path="/tmp/speech.ogg", 84 metadata=None, 85 ) 86 adapter.send_voice.assert_not_awaited() 87 88 89 @pytest.mark.asyncio 90 async def test_base_adapter_routes_voice_tagged_telegram_ogg_media_tag_to_voice_sender(): 91 adapter = _MediaRoutingAdapter() 92 event = _event() 93 adapter._message_handler = AsyncMock( 94 return_value="[[audio_as_voice]]\nMEDIA:/tmp/speech.ogg" 95 ) 96 adapter.send_voice = AsyncMock(return_value=SendResult(success=True, message_id="voice")) 97 adapter.send_document = AsyncMock(return_value=SendResult(success=True, message_id="doc")) 98 99 await adapter._process_message_background(event, build_session_key(event.source)) 100 101 adapter.send_voice.assert_awaited_once_with( 102 chat_id="chat-1", 103 audio_path="/tmp/speech.ogg", 104 metadata=None, 105 ) 106 adapter.send_document.assert_not_awaited() 107 108 109 @pytest.mark.asyncio 110 async def test_streaming_delivery_routes_telegram_flac_media_tag_to_document_sender(): 111 event = _event(thread_id="topic-1") 112 adapter = SimpleNamespace( 113 name="test", 114 extract_media=BasePlatformAdapter.extract_media, 115 extract_images=BasePlatformAdapter.extract_images, 116 extract_local_files=BasePlatformAdapter.extract_local_files, 117 send_voice=AsyncMock(return_value=SendResult(success=True, message_id="voice")), 118 send_document=AsyncMock(return_value=SendResult(success=True, message_id="doc")), 119 send_image_file=AsyncMock(return_value=SendResult(success=True, message_id="image")), 120 send_video=AsyncMock(return_value=SendResult(success=True, message_id="video")), 121 ) 122 123 await GatewayRunner._deliver_media_from_response( 124 object(), 125 "MEDIA:/tmp/speech.flac", 126 event, 127 adapter, 128 ) 129 130 adapter.send_document.assert_awaited_once_with( 131 chat_id="chat-1", 132 file_path="/tmp/speech.flac", 133 metadata={"thread_id": "topic-1"}, 134 ) 135 adapter.send_voice.assert_not_awaited() 136 137 138 @pytest.mark.asyncio 139 async def test_streaming_delivery_routes_non_voice_telegram_ogg_media_tag_to_document_sender(): 140 event = _event(thread_id="topic-1") 141 adapter = SimpleNamespace( 142 name="test", 143 extract_media=BasePlatformAdapter.extract_media, 144 extract_images=BasePlatformAdapter.extract_images, 145 extract_local_files=BasePlatformAdapter.extract_local_files, 146 send_voice=AsyncMock(return_value=SendResult(success=True, message_id="voice")), 147 send_document=AsyncMock(return_value=SendResult(success=True, message_id="doc")), 148 send_image_file=AsyncMock(return_value=SendResult(success=True, message_id="image")), 149 send_video=AsyncMock(return_value=SendResult(success=True, message_id="video")), 150 ) 151 152 await GatewayRunner._deliver_media_from_response( 153 object(), 154 "MEDIA:/tmp/speech.ogg", 155 event, 156 adapter, 157 ) 158 159 adapter.send_document.assert_awaited_once_with( 160 chat_id="chat-1", 161 file_path="/tmp/speech.ogg", 162 metadata={"thread_id": "topic-1"}, 163 ) 164 adapter.send_voice.assert_not_awaited() 165 166 167 @pytest.mark.asyncio 168 async def test_streaming_delivery_routes_telegram_mp3_media_tag_to_voice_sender(): 169 """MP3 audio on Telegram must go through send_voice (which routes to 170 sendAudio internally); Telegram accepts MP3 for the audio player.""" 171 event = _event(thread_id="topic-1") 172 adapter = SimpleNamespace( 173 name="test", 174 extract_media=BasePlatformAdapter.extract_media, 175 extract_images=BasePlatformAdapter.extract_images, 176 extract_local_files=BasePlatformAdapter.extract_local_files, 177 send_voice=AsyncMock(return_value=SendResult(success=True, message_id="voice")), 178 send_document=AsyncMock(return_value=SendResult(success=True, message_id="doc")), 179 send_image_file=AsyncMock(return_value=SendResult(success=True, message_id="image")), 180 send_video=AsyncMock(return_value=SendResult(success=True, message_id="video")), 181 ) 182 183 await GatewayRunner._deliver_media_from_response( 184 object(), 185 "MEDIA:/tmp/speech.mp3", 186 event, 187 adapter, 188 ) 189 190 adapter.send_voice.assert_awaited_once_with( 191 chat_id="chat-1", 192 audio_path="/tmp/speech.mp3", 193 metadata={"thread_id": "topic-1"}, 194 ) 195 adapter.send_document.assert_not_awaited()