"""Tests for BasePlatformAdapter topic-aware session handling.""" import asyncio import json from types import SimpleNamespace from unittest.mock import AsyncMock, patch import pytest from gateway.config import Platform, PlatformConfig from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType, ProcessingOutcome, SendResult from gateway.session import SessionSource, build_session_key class DummyTelegramAdapter(BasePlatformAdapter): def __init__(self): super().__init__(PlatformConfig(enabled=True, token="fake-token"), Platform.TELEGRAM) self._busy_text_mode = "" self.sent = [] self.typing = [] self.processing_hooks = [] 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, "reply_to": reply_to, "metadata": metadata, } ) return SendResult(success=True, message_id="1") async def send_typing(self, chat_id: str, metadata=None) -> None: self.typing.append({"chat_id": chat_id, "metadata": metadata}) return None async def stop_typing(self, chat_id: str, metadata=None) -> None: self.typing.append({"chat_id": chat_id, "stopped": True, "metadata": metadata}) async def get_chat_info(self, chat_id: str): return {"id": chat_id} async def on_processing_start(self, event: MessageEvent) -> None: self.processing_hooks.append(("start", event.message_id)) async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None: self.processing_hooks.append(("complete", event.message_id, outcome)) def _make_event(chat_id: str, thread_id: str, message_id: str = "1") -> MessageEvent: return MessageEvent( text="hello", source=SessionSource( platform=Platform.TELEGRAM, chat_id=chat_id, chat_type="group", thread_id=thread_id, ), message_id=message_id, ) class TestBasePlatformTopicSessions: @pytest.mark.asyncio async def test_handle_message_interrupts_same_topic(self, monkeypatch): adapter = DummyTelegramAdapter() adapter.set_message_handler(lambda event: asyncio.sleep(0, result=None)) active_event = _make_event("-1001", "10") adapter._active_sessions[build_session_key(active_event.source)] = asyncio.Event() scheduled = [] def fake_create_task(coro): scheduled.append(coro) coro.close() return SimpleNamespace() monkeypatch.setattr(asyncio, "create_task", fake_create_task) pending_event = _make_event("-1001", "10", message_id="2") await adapter.handle_message(pending_event) assert scheduled == [] assert adapter.get_pending_message(build_session_key(pending_event.source)) == pending_event @pytest.mark.asyncio async def test_process_message_background_replies_in_same_topic(self): adapter = DummyTelegramAdapter() typing_calls = [] async def handler(_event): await asyncio.sleep(0) return "ack" async def hold_typing(_chat_id, interval=2.0, metadata=None): typing_calls.append({"chat_id": _chat_id, "metadata": metadata}) await asyncio.Event().wait() adapter.set_message_handler(handler) adapter._keep_typing = hold_typing event = _make_event("-1001", "17585") await adapter._process_message_background(event, build_session_key(event.source)) assert adapter.sent == [ { "chat_id": "-1001", "content": "ack", "reply_to": None, "metadata": {"thread_id": "17585", "notify": True}, } ] assert typing_calls == [ { "chat_id": "-1001", "metadata": {"thread_id": "17585"}, } ] assert { "chat_id": "-1001", "stopped": True, "metadata": {"thread_id": "17585"}, } in adapter.typing assert adapter.processing_hooks == [ ("start", "1"), ("complete", "1", ProcessingOutcome.SUCCESS), ] class TestTelegramAutoTtsCaptionDelivery: @staticmethod def _make_voice_event(chat_id: str = "-1001", thread_id: str = "17585") -> MessageEvent: return MessageEvent( text="hello", message_type=MessageType.VOICE, source=SessionSource( platform=Platform.TELEGRAM, chat_id=chat_id, chat_type="group", thread_id=thread_id, ), message_id="voice-1", ) @staticmethod def _hold_typing(): async def hold(_chat_id, interval=2.0, metadata=None): await asyncio.Event().wait() return hold @pytest.mark.asyncio async def test_long_original_with_short_spoken_script_still_sends_full_reply(self, tmp_path): adapter = DummyTelegramAdapter() adapter._keep_typing = self._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")) # Markdown-heavy reply: over the 1024-char caption limit as written, # but the normalized spoken script (markdown and URLs removed) is far # below it. Caption eligibility must follow the ORIGINAL reply, so the # full formatted text is still delivered as its own message instead of # being swallowed into a lossy caption. long_reply = "\n".join( f"- **item {i}** [details](https://example.com/some/very/long/path/{i:04d})" for i in range(20) ) assert len(long_reply) > 1024 assert len(adapter.prepare_tts_text(long_reply)) <= 1024 adapter.set_message_handler(lambda _event: asyncio.sleep(0, result=long_reply)) tts_path = tmp_path / "reply.ogg" tts_path.write_text("audio", encoding="utf-8") event = self._make_voice_event() with patch("tools.tts_tool.check_tts_requirements", return_value=True), patch( "tools.tts_tool.text_to_speech_tool", return_value=json.dumps({"file_path": str(tts_path)}), ): await adapter._process_message_background(event, build_session_key(event.source)) adapter.play_tts.assert_awaited_once() assert adapter.play_tts.await_args.kwargs["caption"] is None assert adapter.sent == [ { "chat_id": "-1001", "content": long_reply, "reply_to": None, "metadata": {"thread_id": "17585", "notify": True}, } ]