test_telegram_reply_mode.py
1 """Tests for Telegram reply_to_mode functionality. 2 3 Covers the threading behavior control for multi-chunk replies: 4 - "off": Never thread replies to original message 5 - "first": Only first chunk threads (default) 6 - "all": All chunks thread to original message 7 """ 8 import os 9 import sys 10 from unittest.mock import MagicMock, AsyncMock, patch 11 12 import pytest 13 14 from gateway.config import PlatformConfig, GatewayConfig, Platform, _apply_env_overrides 15 16 17 def _ensure_telegram_mock(): 18 """Mock the telegram package if it's not installed.""" 19 if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): 20 return 21 mod = MagicMock() 22 mod.ext.ContextTypes.DEFAULT_TYPE = type(None) 23 mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" 24 mod.constants.ChatType.GROUP = "group" 25 mod.constants.ChatType.SUPERGROUP = "supergroup" 26 mod.constants.ChatType.CHANNEL = "channel" 27 mod.constants.ChatType.PRIVATE = "private" 28 for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): 29 sys.modules.setdefault(name, mod) 30 31 32 _ensure_telegram_mock() 33 34 from gateway.platforms.telegram import TelegramAdapter # noqa: E402 35 36 37 @pytest.fixture() 38 def adapter_factory(): 39 """Factory to create TelegramAdapter with custom reply_to_mode.""" 40 def create(reply_to_mode: str = "first"): 41 config = PlatformConfig(enabled=True, token="test-token", reply_to_mode=reply_to_mode) 42 return TelegramAdapter(config) 43 return create 44 45 46 class TestReplyToModeConfig: 47 """Tests for reply_to_mode configuration loading.""" 48 49 def test_default_mode_is_first(self, adapter_factory): 50 adapter = adapter_factory() 51 assert adapter._reply_to_mode == "first" 52 53 def test_off_mode(self, adapter_factory): 54 adapter = adapter_factory(reply_to_mode="off") 55 assert adapter._reply_to_mode == "off" 56 57 def test_first_mode(self, adapter_factory): 58 adapter = adapter_factory(reply_to_mode="first") 59 assert adapter._reply_to_mode == "first" 60 61 def test_all_mode(self, adapter_factory): 62 adapter = adapter_factory(reply_to_mode="all") 63 assert adapter._reply_to_mode == "all" 64 65 def test_invalid_mode_stored_as_is(self, adapter_factory): 66 """Invalid modes are stored but _should_thread_reply handles them.""" 67 adapter = adapter_factory(reply_to_mode="invalid") 68 assert adapter._reply_to_mode == "invalid" 69 70 def test_none_mode_defaults_to_first(self): 71 config = PlatformConfig(enabled=True, token="test-token") 72 adapter = TelegramAdapter(config) 73 assert adapter._reply_to_mode == "first" 74 75 def test_empty_string_mode_defaults_to_first(self): 76 config = PlatformConfig(enabled=True, token="test-token", reply_to_mode="") 77 adapter = TelegramAdapter(config) 78 assert adapter._reply_to_mode == "first" 79 80 81 class TestShouldThreadReply: 82 """Tests for _should_thread_reply method.""" 83 84 def test_no_reply_to_returns_false(self, adapter_factory): 85 adapter = adapter_factory(reply_to_mode="first") 86 assert adapter._should_thread_reply(None, 0) is False 87 assert adapter._should_thread_reply("", 0) is False 88 89 def test_off_mode_never_threads(self, adapter_factory): 90 adapter = adapter_factory(reply_to_mode="off") 91 assert adapter._should_thread_reply("msg-123", 0) is False 92 assert adapter._should_thread_reply("msg-123", 1) is False 93 assert adapter._should_thread_reply("msg-123", 5) is False 94 95 def test_first_mode_only_first_chunk(self, adapter_factory): 96 adapter = adapter_factory(reply_to_mode="first") 97 assert adapter._should_thread_reply("msg-123", 0) is True 98 assert adapter._should_thread_reply("msg-123", 1) is False 99 assert adapter._should_thread_reply("msg-123", 2) is False 100 assert adapter._should_thread_reply("msg-123", 10) is False 101 102 def test_all_mode_all_chunks(self, adapter_factory): 103 adapter = adapter_factory(reply_to_mode="all") 104 assert adapter._should_thread_reply("msg-123", 0) is True 105 assert adapter._should_thread_reply("msg-123", 1) is True 106 assert adapter._should_thread_reply("msg-123", 2) is True 107 assert adapter._should_thread_reply("msg-123", 10) is True 108 109 def test_invalid_mode_falls_back_to_first(self, adapter_factory): 110 """Invalid mode behaves like 'first' - only first chunk threads.""" 111 adapter = adapter_factory(reply_to_mode="invalid") 112 assert adapter._should_thread_reply("msg-123", 0) is True 113 assert adapter._should_thread_reply("msg-123", 1) is False 114 115 116 class TestSendWithReplyToMode: 117 """Tests for send() method respecting reply_to_mode.""" 118 119 @pytest.mark.asyncio 120 async def test_off_mode_no_reply_threading(self, adapter_factory): 121 adapter = adapter_factory(reply_to_mode="off") 122 adapter._bot = MagicMock() 123 adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1)) 124 adapter.truncate_message = lambda content, max_len, **kw: ["chunk1", "chunk2", "chunk3"] 125 126 await adapter.send("12345", "test content", reply_to="999") 127 128 for call in adapter._bot.send_message.call_args_list: 129 assert call.kwargs.get("reply_to_message_id") is None 130 131 @pytest.mark.asyncio 132 async def test_first_mode_only_first_chunk_threads(self, adapter_factory): 133 adapter = adapter_factory(reply_to_mode="first") 134 adapter._bot = MagicMock() 135 adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1)) 136 adapter.truncate_message = lambda content, max_len, **kw: ["chunk1", "chunk2", "chunk3"] 137 138 await adapter.send("12345", "test content", reply_to="999") 139 140 calls = adapter._bot.send_message.call_args_list 141 assert len(calls) == 3 142 assert calls[0].kwargs.get("reply_to_message_id") == 999 143 assert calls[1].kwargs.get("reply_to_message_id") is None 144 assert calls[2].kwargs.get("reply_to_message_id") is None 145 146 @pytest.mark.asyncio 147 async def test_all_mode_all_chunks_thread(self, adapter_factory): 148 adapter = adapter_factory(reply_to_mode="all") 149 adapter._bot = MagicMock() 150 adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1)) 151 adapter.truncate_message = lambda content, max_len, **kw: ["chunk1", "chunk2", "chunk3"] 152 153 await adapter.send("12345", "test content", reply_to="999") 154 155 calls = adapter._bot.send_message.call_args_list 156 assert len(calls) == 3 157 for call in calls: 158 assert call.kwargs.get("reply_to_message_id") == 999 159 160 @pytest.mark.asyncio 161 async def test_no_reply_to_param_no_threading(self, adapter_factory): 162 adapter = adapter_factory(reply_to_mode="all") 163 adapter._bot = MagicMock() 164 adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1)) 165 adapter.truncate_message = lambda content, max_len, **kw: ["chunk1", "chunk2"] 166 167 await adapter.send("12345", "test content", reply_to=None) 168 169 calls = adapter._bot.send_message.call_args_list 170 for call in calls: 171 assert call.kwargs.get("reply_to_message_id") is None 172 173 @pytest.mark.asyncio 174 async def test_single_chunk_respects_mode(self, adapter_factory): 175 adapter = adapter_factory(reply_to_mode="first") 176 adapter._bot = MagicMock() 177 adapter._bot.send_message = AsyncMock(return_value=MagicMock(message_id=1)) 178 adapter.truncate_message = lambda content, max_len, **kw: ["single chunk"] 179 180 await adapter.send("12345", "test", reply_to="999") 181 182 calls = adapter._bot.send_message.call_args_list 183 assert len(calls) == 1 184 assert calls[0].kwargs.get("reply_to_message_id") == 999 185 186 187 class TestConfigSerialization: 188 """Tests for reply_to_mode serialization.""" 189 190 def test_to_dict_includes_reply_to_mode(self): 191 config = PlatformConfig(enabled=True, token="test", reply_to_mode="all") 192 result = config.to_dict() 193 assert result["reply_to_mode"] == "all" 194 195 def test_from_dict_loads_reply_to_mode(self): 196 data = {"enabled": True, "token": "test", "reply_to_mode": "off"} 197 config = PlatformConfig.from_dict(data) 198 assert config.reply_to_mode == "off" 199 200 def test_from_dict_defaults_to_first(self): 201 data = {"enabled": True, "token": "test"} 202 config = PlatformConfig.from_dict(data) 203 assert config.reply_to_mode == "first" 204 205 206 class TestEnvVarOverride: 207 """Tests for TELEGRAM_REPLY_TO_MODE environment variable override.""" 208 209 def _make_config(self): 210 config = GatewayConfig() 211 config.platforms[Platform.TELEGRAM] = PlatformConfig(enabled=True, token="test") 212 return config 213 214 def test_env_var_sets_off_mode(self): 215 config = self._make_config() 216 with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": "off"}, clear=False): 217 _apply_env_overrides(config) 218 assert config.platforms[Platform.TELEGRAM].reply_to_mode == "off" 219 220 def test_env_var_sets_all_mode(self): 221 config = self._make_config() 222 with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": "all"}, clear=False): 223 _apply_env_overrides(config) 224 assert config.platforms[Platform.TELEGRAM].reply_to_mode == "all" 225 226 def test_env_var_case_insensitive(self): 227 config = self._make_config() 228 with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": "ALL"}, clear=False): 229 _apply_env_overrides(config) 230 assert config.platforms[Platform.TELEGRAM].reply_to_mode == "all" 231 232 def test_env_var_invalid_value_ignored(self): 233 config = self._make_config() 234 with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": "banana"}, clear=False): 235 _apply_env_overrides(config) 236 assert config.platforms[Platform.TELEGRAM].reply_to_mode == "first" 237 238 def test_env_var_empty_value_ignored(self): 239 config = self._make_config() 240 with patch.dict(os.environ, {"TELEGRAM_REPLY_TO_MODE": ""}, clear=False): 241 _apply_env_overrides(config) 242 assert config.platforms[Platform.TELEGRAM].reply_to_mode == "first"