import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from unittest.mock import AsyncMock, MagicMock, call, patch import pytest # --------------------------------------------------------------------------- # 헬퍼: Telegram Update / Message / Bot 목 객체 생성 # (test_bot_logic.py 패턴과 일관성 유지) # --------------------------------------------------------------------------- def _make_user(user_id: int, is_bot: bool = False, username: str = "testuser") -> MagicMock: user = MagicMock() user.id = user_id user.is_bot = is_bot user.username = username return user def _make_message( text: str, from_user: MagicMock, chat_id: int = 100, chat_type: str = "group", reply_to_message=None, entities=None, ) -> MagicMock: msg = MagicMock() msg.text = text msg.from_user = from_user msg.chat = MagicMock() msg.chat.id = chat_id msg.chat.type = chat_type msg.reply_to_message = reply_to_message msg.entities = entities or [] msg.reply_text = AsyncMock() return msg def _make_update(message: MagicMock) -> MagicMock: update = MagicMock() update.message = message update.effective_message = message update.effective_user = message.from_user return update def _make_context(bot_username: str = "mybot") -> MagicMock: ctx = MagicMock() ctx.bot = MagicMock() ctx.bot.username = bot_username ctx.bot_data = {} return ctx def _make_async_bot(bot_id: int = 999, username: str = "mybot") -> AsyncMock: """send_message / edit_message_text를 AsyncMock으로 가진 봇 목 객체.""" bot = MagicMock() bot.id = bot_id bot.username = username sent_msg = MagicMock() sent_msg.message_id = 12345 bot.send_message = AsyncMock(return_value=sent_msg) bot.edit_message_text = AsyncMock() return bot # --------------------------------------------------------------------------- # Class 1: TestThinkingMessageHelpers # bot_utils.py에 추가될 헬퍼 함수 테스트 # # 대상 함수 시그니처 (미구현): # async def send_thinking_message(bot, chat_id: int) -> int # async def replace_thinking_message( # bot, chat_id: int, message_id: int, text: str, max_len: int = 4096 # ) -> None # --------------------------------------------------------------------------- class TestThinkingMessageHelpers: """bot_utils.send_thinking_message / replace_thinking_message 테스트""" @pytest.mark.asyncio async def test_send_thinking_message(self): """send_thinking_message() 호출 시 '🤔 생각 중...' 텍스트로 bot.send_message가 호출되고, message_id가 반환되어야 한다.""" from bot_utils import send_thinking_message bot = _make_async_bot() chat_id = 100 result = await send_thinking_message(bot, chat_id) bot.send_message.assert_called_once_with(chat_id=chat_id, text="🤔 생각 중...") assert result == 12345 @pytest.mark.asyncio async def test_replace_thinking_with_short_response(self): """replace_thinking_message()에 4096자 이하 텍스트를 전달하면 bot.edit_message_text가 한 번 호출되어야 한다.""" from bot_utils import replace_thinking_message bot = _make_async_bot() chat_id = 100 message_id = 12345 short_text = "짧은 응답입니다." await replace_thinking_message(bot, chat_id, message_id, short_text) bot.edit_message_text.assert_called_once_with( chat_id=chat_id, message_id=message_id, text=short_text, ) bot.send_message.assert_not_called() @pytest.mark.asyncio async def test_replace_thinking_with_long_response(self): """replace_thinking_message()에 4096자 초과 텍스트를 전달하면 첫 청크는 edit_message_text로, 나머지 청크는 send_message로 전송해야 한다.""" from bot_utils import replace_thinking_message bot = _make_async_bot() chat_id = 100 message_id = 12345 # 4096자 초과: 두 청크로 분할될 텍스트 long_text = "가" * 5000 await replace_thinking_message(bot, chat_id, message_id, long_text) # 첫 번째 청크: edit_message_text로 전송 assert bot.edit_message_text.call_count == 1 first_call_kwargs = bot.edit_message_text.call_args.kwargs assert first_call_kwargs["chat_id"] == chat_id assert first_call_kwargs["message_id"] == message_id assert len(first_call_kwargs["text"]) <= 4096 # 나머지 청크: send_message로 전송 assert bot.send_message.call_count >= 1 for send_call in bot.send_message.call_args_list: sent_kwargs = send_call.kwargs assert sent_kwargs["chat_id"] == chat_id assert len(sent_kwargs["text"]) <= 4096 # --------------------------------------------------------------------------- # Class 2: TestSequentialResponse # 순차 응답 로직 테스트 # # 봇 순서: gemini_view_bot → codex_view_bot → claude_view_bot # --------------------------------------------------------------------------- class TestSequentialResponse: """멀티봇 순차 응답 로직 테스트""" def _make_discussion_manager(self): from discussion_manager import DiscussionManager return DiscussionManager() @pytest.mark.asyncio async def test_user_message_triggers_first_bot_only(self): """유저 메시지 수신 시 첫 번째 봇(gemini_view_bot)의 current_turn이 설정되고, 나머지 봇은 should_bot_respond()에서 False를 반환해야 한다.""" from discussion_manager import DiscussionManager dm = self._make_discussion_manager() chat_id = 200 dm.on_user_message(chat_id, "안녕") first_bot = DiscussionManager.BOT_USERNAMES[0] second_bot = DiscussionManager.BOT_USERNAMES[1] third_bot = DiscussionManager.BOT_USERNAMES[2] # 첫 번째 봇은 현재 턴 → True assert ( dm.should_bot_respond( bot_username=first_bot, sender_username="someuser", sender_is_bot=False, chat_id=chat_id, ) is True ) # 두 번째, 세 번째 봇은 아직 턴이 아님 → False assert ( dm.should_bot_respond( bot_username=second_bot, sender_username="someuser", sender_is_bot=False, chat_id=chat_id, ) is False ) assert ( dm.should_bot_respond( bot_username=third_bot, sender_username="someuser", sender_is_bot=False, chat_id=chat_id, ) is False ) @pytest.mark.asyncio async def test_first_bot_triggers_second(self): """첫 번째 봇이 응답하면 두 번째 봇의 턴으로 넘어가야 한다.""" from discussion_manager import DiscussionManager dm = self._make_discussion_manager() chat_id = 201 dm.on_user_message(chat_id, "안녕") first_bot = DiscussionManager.BOT_USERNAMES[0] second_bot = DiscussionManager.BOT_USERNAMES[1] # 첫 번째 봇 응답 후 next_bot이 두 번째 봇이어야 함 next_bot = dm.on_bot_response(first_bot) assert next_bot == second_bot # 이제 두 번째 봇이 should_bot_respond에서 True를 반환해야 함 assert ( dm.should_bot_respond( bot_username=second_bot, sender_username=first_bot, sender_is_bot=True, chat_id=chat_id, ) is True ) @pytest.mark.asyncio async def test_response_order_includes_all_three(self): """세 봇이 순서대로 모두 응답하는 흐름이 정상 동작해야 한다. 유저 메시지 → gemini → codex → claude → gemini(wrap-around) 순서.""" from discussion_manager import DiscussionManager dm = self._make_discussion_manager() chat_id = 202 dm.on_user_message(chat_id, "안녕") bot_a = DiscussionManager.BOT_USERNAMES[0] # gemini_view_bot bot_b = DiscussionManager.BOT_USERNAMES[1] # codex_view_bot bot_c = DiscussionManager.BOT_USERNAMES[2] # claude_view_bot # 응답 순서 기록 response_order = [bot_a] next_b = dm.on_bot_response(bot_a) assert next_b is not None assert next_b == bot_b response_order.append(next_b) next_c = dm.on_bot_response(bot_b) assert next_c is not None assert next_c == bot_c response_order.append(next_c) # 3개 봇 모두 순차 응답 확인 assert response_order == [bot_a, bot_b, bot_c] # 마지막 봇 응답 후 다시 첫 번째 봇으로 순환 next_wrap = dm.on_bot_response(bot_c) assert next_wrap == bot_a # --------------------------------------------------------------------------- # Class 3: TestContextPrompt # 맥락 기반 프롬프트 테스트 # --------------------------------------------------------------------------- class TestContextPrompt: """봇이 생성하는 프롬프트에 맥락이 올바르게 포함되는지 테스트""" def _make_discussion_prompt( self, previous_bot_name: str, previous_message: str, ) -> str: """main_bot.py의 trigger_next_bot_response에서 사용하는 프롬프트 포맷을 재현한다.""" return ( f"[그룹 토론 중] {previous_bot_name}의 발언: {previous_message}\n\n" "위 내용에 대해 당신의 의견을 간결하게 말해주세요." ) def test_context_includes_previous_messages(self): """프롬프트에 이전 대화 내용(previous_message)이 포함되어야 한다.""" previous_bot = "gemini_view_bot" previous_msg = "인공지능은 인류에 도움이 됩니다." prompt = self._make_discussion_prompt(previous_bot, previous_msg) assert previous_msg in prompt def test_second_bot_sees_first_bot_response(self): """두 번째 봇의 프롬프트에 첫 번째 봇의 응답 내용이 포함되어야 한다.""" first_bot_name = "gemini_view_bot" first_bot_response = "저는 이 문제에 대해 긍정적으로 생각합니다." prompt = self._make_discussion_prompt(first_bot_name, first_bot_response) # 첫 번째 봇 이름과 응답이 모두 포함되어야 함 assert first_bot_name in prompt assert first_bot_response in prompt def test_persona_included_in_prompt(self): """각 봇의 페르소나(이름)가 프롬프트의 발언자 표기에 포함되어야 한다.""" from discussion_manager import DiscussionManager for bot_username in DiscussionManager.BOT_USERNAMES: some_message = "테스트 발언입니다." prompt = self._make_discussion_prompt(bot_username, some_message) # 프롬프트에 해당 봇의 username이 포함되어야 함 assert bot_username in prompt @pytest.mark.asyncio async def test_trigger_next_bot_response_builds_correct_prompt(self): """trigger_next_bot_response()가 memory.format_context()를 통해 맥락 기반 프롬프트를 엔진 호출에 사용해야 한다.""" # main_bot.trigger_next_bot_response의 프롬프트 구성 로직을 검증한다. # 외부 의존성(DiscussionManager, call_codex, app.bot, memory)은 모두 mock 처리한다. # codex_view_bot은 call_codex를 직접 호출하므로 call_codex를 패치한다. captured_prompts: list[str] = [] expected_context_prompt = "코덱스의 관점에서 이전 대화를 참고하여 답변하라." async def fake_call_codex(prompt: str, model: str = "gpt-5.1-codex-mini") -> str: captured_prompts.append(prompt) return "모의 응답" mock_memory = MagicMock() mock_memory.format_context.return_value = expected_context_prompt env_patch = { "GEMINI_BOT_TOKEN": "fake-gemini-token", "CODEX_BOT_TOKEN": "fake-codex-token", "CLAUDE_BOT_TOKEN": "fake-claude-token", } with ( patch.dict("os.environ", env_patch), patch("main_bot.dm") as mock_dm, patch("main_bot.memory", mock_memory), patch("main_bot.bot_apps") as mock_bot_apps, patch("main_bot.call_codex", new=fake_call_codex), patch("asyncio.sleep", new_callable=AsyncMock), ): mock_dm.is_discussion_active.return_value = True mock_dm.on_bot_response.return_value = None # 다음 봇 없음 → 토론 종료 mock_dm.get_current_phase.return_value.value = "diverge" mock_dm.get_codex_model.return_value = "gpt-5.1-codex-mini" mock_app = MagicMock() mock_app.bot.send_message = AsyncMock(return_value=MagicMock(message_id=999)) mock_app.bot.edit_message_text = AsyncMock() mock_bot_apps.__contains__ = MagicMock(return_value=True) mock_bot_apps.__getitem__ = MagicMock(return_value=(mock_app, AsyncMock(return_value="모의 응답"))) from main_bot import trigger_next_bot_response await trigger_next_bot_response( next_bot_username="codex_view_bot", chat_id=300, previous_message="이전 봇의 발언 내용입니다.", previous_bot_name="gemini_view_bot", ) assert len(captured_prompts) == 1 prompt = captured_prompts[0] # 새 구현: memory.format_context()의 반환값이 call_codex에 전달되어야 한다 assert prompt == expected_context_prompt mock_memory.format_context.assert_called_once_with(300, "codex_view_bot", phase="diverge")