from __future__ import annotations import itertools from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest from telegram import ForceReply, ReplyParameters from gateway.config import PlatformConfig from gateway.platforms.telegram import TelegramAdapter from gateway.platforms.telegram_nutrition_onboarding_preview import ( PreviewSession, ) OWNER_ID = 8_693_203_710 PREVIEW_CHAT_ID = -1_004_290_459_350 PREVIEW_TOPIC_ID = 777 PREVIEW_TOPIC_NAME = "[PREVIEW] 영양 온보딩" VALID_ANSWERS = ( "1990-01-01", "남성", "180", "80", "보통", "주 3회 운동", "감량", "75", "2026-12-01", "없음", "없음", "없음", "없음", "균형식", "없음", "없음", "아니오", "아니오", "가능", "보통", "3", "없음", ) def _adapter() -> TelegramAdapter: config = PlatformConfig( enabled=True, token="test-token", extra={ "nutrition_onboarding_preview": { "enabled": True, "chat_id": str(PREVIEW_CHAT_ID), "topic_name": PREVIEW_TOPIC_NAME, "topic_id": str(PREVIEW_TOPIC_ID), }, }, ) adapter = TelegramAdapter(config) receipt_ids = itertools.count(901) async def _send_message(**_kwargs: object) -> SimpleNamespace: return SimpleNamespace(message_id=next(receipt_ids)) adapter._bot = SimpleNamespace( send_message=AsyncMock(side_effect=_send_message) ) coordinator = SimpleNamespace( owner=SimpleNamespace( user_id=str(OWNER_ID), chat_id=str(OWNER_ID), topic_id="0", ), resolve=MagicMock(return_value=None), open_customer_onboarding=None, resolve_stopped=None, owns_space=MagicMock(return_value=False), ) adapter._get_nutrition_coaching = MagicMock(return_value=coordinator) adapter._room_bootstrap_reserves_ingress = MagicMock(return_value=False) adapter._reserve_adaptive_review_update = AsyncMock(return_value=False) adapter._should_process_message = MagicMock(return_value=True) adapter._ensure_forum_commands = AsyncMock() adapter._build_message_event = MagicMock( return_value=SimpleNamespace(text="/nutritionpreview") ) adapter._clean_bot_trigger_text = MagicMock( return_value="/nutritionpreview" ) adapter._cache_replied_media = AsyncMock() adapter._apply_telegram_group_observe_attribution = lambda event: event adapter._get_physique_checkin = MagicMock(return_value=None) adapter._enqueue_text_event = MagicMock() adapter.handle_message = AsyncMock() return adapter def _message(text: str) -> SimpleNamespace: return SimpleNamespace( text=text, message_id=900, message_thread_id=PREVIEW_TOPIC_ID, is_topic_message=True, from_user=SimpleNamespace(id=OWNER_ID), chat=SimpleNamespace(id=PREVIEW_CHAT_ID, type="supergroup"), reply_text=AsyncMock(return_value=SimpleNamespace(message_id=901)), ) def _reply( text: str, *, actor_id: int = OWNER_ID, topic_id: int = PREVIEW_TOPIC_ID, prompt_id: int = 901, receipt_id: int = 902, ) -> SimpleNamespace: return SimpleNamespace( text=text, message_id=receipt_id - 1, message_thread_id=topic_id, is_topic_message=True, from_user=SimpleNamespace(id=actor_id), chat=SimpleNamespace(id=PREVIEW_CHAT_ID, type="supergroup"), reply_to_message=SimpleNamespace(message_id=prompt_id), reply_text=AsyncMock( return_value=SimpleNamespace(message_id=receipt_id) ), ) def _update(message: SimpleNamespace, update_id: int) -> SimpleNamespace: return SimpleNamespace( message=message, effective_message=message, update_id=update_id, ) @pytest.mark.asyncio async def test_owner_preview_command_is_reserved_before_generic_routing() -> None: adapter = _adapter() message = _message("/nutritionpreview") await adapter._handle_command(_update(message, 1), SimpleNamespace()) adapter._bot.send_message.assert_awaited_once() sent = adapter._bot.send_message.await_args.kwargs sent_text = sent["text"] assert sent_text.startswith("[PREVIEW][NUTRITION ONBOARDING]") assert "생년월일을 YYYY-MM-DD 형식으로 입력해 주세요." in sent_text assert sent["message_thread_id"] == PREVIEW_TOPIC_ID assert "direct_messages_topic_id" not in sent force_reply = sent["reply_markup"] assert isinstance(force_reply, ForceReply) assert force_reply.selective is True assert force_reply.input_field_placeholder == "답변을 입력하세요" reply_parameters = sent["reply_parameters"] assert isinstance(reply_parameters, ReplyParameters) assert reply_parameters.message_id == 900 adapter.handle_message.assert_not_awaited() @pytest.mark.asyncio async def test_exact_anchored_reply_advances_and_duplicate_is_consumed() -> None: adapter = _adapter() start = _message("/nutritionpreview") await adapter._handle_command(_update(start, 1), SimpleNamespace()) reply = _reply("1990-01-01") await adapter._handle_text_message(_update(reply, 2), SimpleNamespace()) await adapter._handle_text_message(_update(reply, 3), SimpleNamespace()) assert adapter._bot.send_message.await_count == 2 sent_text = adapter._bot.send_message.await_args.kwargs["text"] assert "계산 기준을 입력해 주세요" in sent_text adapter.handle_message.assert_not_awaited() @pytest.mark.asyncio async def test_wrong_actor_and_stale_reply_never_advance_or_fall_through() -> None: adapter = _adapter() start = _message("/nutritionpreview") await adapter._handle_command(_update(start, 1), SimpleNamespace()) wrong_actor = _reply("1990-01-01", actor_id=42) stale = _reply("1990-01-01", prompt_id=899) await adapter._handle_text_message( _update(wrong_actor, 2), SimpleNamespace(), ) await adapter._handle_text_message(_update(stale, 3), SimpleNamespace()) assert adapter._bot.send_message.await_count == 1 adapter.handle_message.assert_not_awaited() @pytest.mark.asyncio async def test_invalid_answer_keeps_original_prompt_correlation() -> None: adapter = _adapter() start = _message("/nutritionpreview") await adapter._handle_command(_update(start, 1), SimpleNamespace()) prompt_id = 901 for index, answer in enumerate(VALID_ANSWERS[:20]): receipt_id = 902 + index reply = _reply( answer, prompt_id=prompt_id, receipt_id=receipt_id, ) await adapter._handle_text_message( _update(reply, 2 + index), SimpleNamespace(), ) prompt_id = receipt_id invalid = _reply( "not-a-number", prompt_id=prompt_id, receipt_id=922, ) valid = _reply( "3", prompt_id=prompt_id, receipt_id=923, ) await adapter._handle_text_message( _update(invalid, 22), SimpleNamespace(), ) await adapter._handle_text_message(_update(valid, 23), SimpleNamespace()) sent_texts = [ call.kwargs["text"] for call in adapter._bot.send_message.await_args_list ] assert "입력 형식" in sent_texts[-2] assert "식사 시간 제약을 적어 주세요" in sent_texts[-1] @pytest.mark.asyncio async def test_wrong_user_preview_command_is_consumed_without_state() -> None: adapter = _adapter() message = _message("/nutritionpreview") message.from_user = SimpleNamespace(id=42) await adapter._handle_command(_update(message, 1), SimpleNamespace()) adapter._bot.send_message.assert_not_awaited() adapter.handle_message.assert_not_awaited() @pytest.mark.asyncio async def test_preview_topic_callback_never_reaches_production_onboarding() -> None: adapter = _adapter() adapter._handle_nutrition_onboarding_callback = AsyncMock() message = _message("preview callback") query = SimpleNamespace( data="non1:opaque-production-callback", message=message, from_user=SimpleNamespace(id=OWNER_ID), answer=AsyncMock(), ) update = SimpleNamespace(callback_query=query) await adapter._handle_callback_query(update, SimpleNamespace()) query.answer.assert_awaited_once() adapter._handle_nutrition_onboarding_callback.assert_not_awaited() @pytest.mark.asyncio async def test_preview_confirm_callback_routes_before_preview_guard() -> None: adapter = _adapter() adapter._handle_nutrition_onboarding_callback = AsyncMock() preview = adapter._get_nutrition_onboarding_preview() assert preview is not None preview._session = PreviewSession( cursor=22, expected_message_id="900", answers=(), phase="confirming", ) message = _message("summary") query = SimpleNamespace( data="npv:confirm", message=message, from_user=SimpleNamespace(id=OWNER_ID), answer=AsyncMock(), edit_message_reply_markup=AsyncMock(), ) await adapter._handle_callback_query( SimpleNamespace(callback_query=query), SimpleNamespace(), ) assert preview._session is None query.edit_message_reply_markup.assert_awaited_once_with( reply_markup=None ) assert "운영자 검토" in adapter._bot.send_message.await_args.kwargs["text"] adapter._handle_nutrition_onboarding_callback.assert_not_awaited() @pytest.mark.asyncio async def test_restart_lost_preview_confirmation_fails_closed() -> None: before_restart = _adapter() preview = before_restart._get_nutrition_onboarding_preview() assert preview is not None preview._session = PreviewSession( cursor=22, expected_message_id="900", answers=(), phase="confirming", ) restarted = _adapter() restarted._handle_nutrition_onboarding_callback = AsyncMock() query = SimpleNamespace( data="npv:confirm", message=_message("summary"), from_user=SimpleNamespace(id=OWNER_ID), answer=AsyncMock(), edit_message_reply_markup=AsyncMock(), ) await restarted._handle_callback_query( SimpleNamespace(callback_query=query), SimpleNamespace(), ) query.answer.assert_awaited_once_with( text="이 카드는 이미 처리됐거나 사용할 수 없습니다." ) query.edit_message_reply_markup.assert_not_awaited() restarted._bot.send_message.assert_not_awaited() restarted._handle_nutrition_onboarding_callback.assert_not_awaited()