"""Tests for duty_teller.handlers.commands (start, set_phone, calendar_link, help_cmd).""" from unittest.mock import AsyncMock, MagicMock, patch import pytest from duty_teller.handlers.commands import ( start, set_phone, calendar_link, help_cmd, set_role, ) def _make_update(message=None, effective_user=None, effective_chat=None): """Build a minimal Update with message and effective_user/chat.""" update = MagicMock() update.message = message update.effective_user = effective_user update.effective_chat = effective_chat return update def _make_user( user_id=1, first_name="Test", last_name="User", username="testuser", ): """Build a minimal Telegram user.""" u = MagicMock() u.id = user_id u.first_name = first_name u.last_name = last_name u.username = username return u def _make_chat(chat_type="private"): """Build a minimal chat.""" ch = MagicMock() ch.type = chat_type return ch @pytest.mark.asyncio async def test_start_calls_get_or_create_user_and_reply_text(): """start: calls get_or_create_user with expected fields and reply_text with start.greeting.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user(42, first_name="Alice", last_name="B") update = _make_update(message=message, effective_user=user) with patch( "duty_teller.handlers.commands.session_scope", ) as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch( "duty_teller.handlers.commands.get_or_create_user", ) as mock_get_user: with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Hi! Use /help." with patch("duty_teller.handlers.commands.get_lang", return_value="en"): await start(update, MagicMock()) mock_get_user.assert_called_once() call_kw = mock_get_user.call_args[1] assert call_kw["telegram_user_id"] == 42 assert call_kw["full_name"] == "Alice B" assert call_kw["username"] == "testuser" message.reply_text.assert_called_once_with("Hi! Use /help.") mock_t.assert_called_with("en", "start.greeting") @pytest.mark.asyncio async def test_start_no_message_returns_early(): """start: if update.message is None, does not call get_or_create_user or reply.""" update = _make_update(message=None, effective_user=_make_user()) with patch("duty_teller.handlers.commands.session_scope") as mock_scope: with patch("duty_teller.handlers.commands.get_or_create_user") as mock_get_user: await start(update, MagicMock()) mock_get_user.assert_not_called() mock_scope.assert_not_called() @pytest.mark.asyncio async def test_start_no_effective_user_returns_early(): """start: if update.effective_user is None, does not call executor or reply.""" message = MagicMock() message.reply_text = AsyncMock() update = _make_update(message=message, effective_user=None) with patch("duty_teller.handlers.commands.session_scope") as mock_scope: with patch("duty_teller.handlers.commands.get_or_create_user") as mock_get_user: await start(update, MagicMock()) mock_get_user.assert_not_called() mock_scope.assert_not_called() message.reply_text.assert_not_called() @pytest.mark.asyncio async def test_set_phone_private_with_number_replies_saved(): """set_phone in private chat with number -> reply 'saved'.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() chat = _make_chat("private") update = _make_update(message=message, effective_user=user, effective_chat=chat) context = MagicMock() context.args = ["+79001234567"] with patch("duty_teller.handlers.commands.session_scope") as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch("duty_teller.handlers.commands.get_or_create_user"): with patch( "duty_teller.handlers.commands.set_user_phone", return_value=MagicMock(), ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Phone saved: +79001234567" await set_phone(update, context) message.reply_text.assert_called_once_with("Phone saved: +79001234567") @pytest.mark.asyncio async def test_set_phone_private_no_number_replies_cleared(): """set_phone in private chat without number -> reply 'cleared'.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() chat = _make_chat("private") update = _make_update(message=message, effective_user=user, effective_chat=chat) context = MagicMock() context.args = [] with patch("duty_teller.handlers.commands.session_scope") as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch("duty_teller.handlers.commands.get_or_create_user"): with patch( "duty_teller.handlers.commands.set_user_phone", return_value=MagicMock(), ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Phone cleared" await set_phone(update, context) message.reply_text.assert_called_once_with("Phone cleared") @pytest.mark.asyncio async def test_set_phone_group_replies_private_only(): """set_phone in group chat -> reply private_only.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() chat = _make_chat("group") update = _make_update(message=message, effective_user=user, effective_chat=chat) context = MagicMock() context.args = ["+79001234567"] with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Private only" await set_phone(update, context) message.reply_text.assert_called_once_with("Private only") mock_t.assert_called_with("en", "set_phone.private_only") @pytest.mark.asyncio async def test_set_phone_result_error_replies_error(): """set_phone: set_user_phone returns None (error) -> reply set_phone.error.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() chat = _make_chat("private") update = _make_update(message=message, effective_user=user, effective_chat=chat) context = MagicMock() context.args = ["+79001234567"] with patch("duty_teller.handlers.commands.session_scope") as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch("duty_teller.handlers.commands.get_or_create_user"): with patch( "duty_teller.handlers.commands.set_user_phone", return_value=None, ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Error saving phone" await set_phone(update, context) message.reply_text.assert_called_once_with("Error saving phone") mock_t.assert_called_with("en", "set_phone.error") @pytest.mark.asyncio async def test_calendar_link_with_user_and_token_replies_with_url(): """calendar_link with allowed user and token -> reply with personal and team URLs.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() chat = _make_chat("private") update = _make_update(message=message, effective_user=user, effective_chat=chat) with patch("duty_teller.handlers.commands.session_scope") as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch("duty_teller.handlers.commands.get_or_create_user") as mock_get_user: mock_user = MagicMock() mock_user.id = 10 mock_user.phone = None mock_get_user.return_value = mock_user with patch( "duty_teller.handlers.commands.can_access_miniapp_for_telegram_user", return_value=True, ): with patch("duty_teller.handlers.commands.config") as mock_cfg: mock_cfg.MINI_APP_BASE_URL = "https://example.com" with patch( "duty_teller.handlers.commands.create_calendar_token", return_value="abc43token", ): with patch( "duty_teller.handlers.commands.get_lang", return_value="en" ): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.side_effect = lambda lang, key, **kw: ( f"Personal: {kw.get('url_personal', '')} Team: {kw.get('url_team', '')}" if "success" in key else "Hint" ) await calendar_link(update, MagicMock()) message.reply_text.assert_called_once() call_args = message.reply_text.call_args[0][0] assert "/api/calendar/ical/abc43token.ics" in call_args assert "/api/calendar/ical/team/abc43token.ics" in call_args @pytest.mark.asyncio async def test_calendar_link_in_group_replies_private_only(): """calendar_link in group chat -> reply calendar_link.private_only.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() chat = _make_chat("group") update = _make_update(message=message, effective_user=user, effective_chat=chat) with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Private chat only" await calendar_link(update, MagicMock()) message.reply_text.assert_called_once_with("Private chat only") mock_t.assert_called_with("en", "calendar_link.private_only") @pytest.mark.asyncio async def test_calendar_link_no_url_replies_error(): """calendar_link: can_access True but MINI_APP_BASE_URL empty -> reply calendar_link.error.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() chat = _make_chat("private") update = _make_update(message=message, effective_user=user, effective_chat=chat) with patch("duty_teller.handlers.commands.session_scope") as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch("duty_teller.handlers.commands.get_or_create_user") as mock_get_user: mock_user = MagicMock() mock_user.id = 10 mock_get_user.return_value = mock_user with patch( "duty_teller.handlers.commands.can_access_miniapp_for_telegram_user", return_value=True, ): with patch("duty_teller.handlers.commands.config") as mock_cfg: mock_cfg.MINI_APP_BASE_URL = "" with patch( "duty_teller.handlers.commands.create_calendar_token", return_value="token", ): with patch( "duty_teller.handlers.commands.get_lang", return_value="en" ): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Calendar error" await calendar_link(update, MagicMock()) message.reply_text.assert_called_once_with("Calendar error") mock_t.assert_called_with("en", "calendar_link.error") @pytest.mark.asyncio async def test_calendar_link_denied_replies_access_denied(): """calendar_link when user not in allowlist -> reply access_denied.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() chat = _make_chat("private") update = _make_update(message=message, effective_user=user, effective_chat=chat) with patch("duty_teller.handlers.commands.session_scope") as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch("duty_teller.handlers.commands.get_or_create_user") as mock_get_user: mock_user = MagicMock() mock_user.id = 10 mock_user.phone = None mock_get_user.return_value = mock_user with patch( "duty_teller.handlers.commands.can_access_miniapp_for_telegram_user", return_value=False, ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Access denied" await calendar_link(update, MagicMock()) message.reply_text.assert_called_once_with("Access denied") mock_t.assert_called_with("en", "calendar_link.access_denied") @pytest.mark.asyncio async def test_help_cmd_replies_with_help_text(): """help_cmd: reply_text called with help content (title, start, help, etc.).""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() update = _make_update(message=message, effective_user=user) with patch( "duty_teller.handlers.commands.is_admin_async", new_callable=AsyncMock, return_value=False, ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.side_effect = lambda lang, key: f"[{key}]" await help_cmd(update, MagicMock()) message.reply_text.assert_called_once() text = message.reply_text.call_args[0][0] assert "help.title" in text or "[help.title]" in text assert "help.start" in text or "[help.start]" in text @pytest.mark.asyncio async def test_help_cmd_admin_includes_import_schedule_and_set_role(): """help_cmd: when is_admin_async True, reply includes help.import_schedule and help.set_role.""" message = MagicMock() message.reply_text = AsyncMock() user = _make_user() update = _make_update(message=message, effective_user=user) with patch( "duty_teller.handlers.commands.is_admin_async", new_callable=AsyncMock, return_value=True, ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.side_effect = lambda lang, key: f"[{key}]" await help_cmd(update, MagicMock()) message.reply_text.assert_called_once() text = message.reply_text.call_args[0][0] assert "help.import_schedule" in text or "[help.import_schedule]" in text assert "help.set_role" in text or "[help.set_role]" in text # --- set_role --- @pytest.mark.asyncio async def test_set_role_no_effective_user_returns_early(): """set_role: no update.effective_user -> no reply, no executor.""" message = MagicMock() message.reply_text = AsyncMock() update = _make_update(message=message, effective_user=None) context = MagicMock() context.args = [] with patch("duty_teller.handlers.commands.session_scope") as mock_scope: await set_role(update, context) message.reply_text.assert_not_called() mock_scope.assert_not_called() @pytest.mark.asyncio async def test_set_role_not_admin_replies_admin_only(): """set_role: non-admin -> reply import.admin_only.""" message = MagicMock() message.reply_text = AsyncMock() message.reply_to_message = None user = _make_user() update = _make_update(message=message, effective_user=user) context = MagicMock() context.args = [] with patch( "duty_teller.handlers.commands.is_admin_async", new_callable=AsyncMock, return_value=False, ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Admin only" await set_role(update, context) message.reply_text.assert_called_once_with("Admin only") mock_t.assert_called_with("en", "import.admin_only") @pytest.mark.asyncio async def test_set_role_invalid_usage_replies_usage(): """set_role: invalid role_name (not user|admin) -> reply set_role.usage.""" message = MagicMock() message.reply_text = AsyncMock() message.reply_to_message = MagicMock() message.reply_to_message.from_user = MagicMock() message.reply_to_message.from_user.id = 111 user = _make_user() update = _make_update(message=message, effective_user=user) context = MagicMock() context.args = ["invalid_role"] with patch( "duty_teller.handlers.commands.is_admin_async", new_callable=AsyncMock, return_value=True, ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.session_scope") as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch( "duty_teller.handlers.commands.get_user_by_telegram_id", return_value=MagicMock(id=1, full_name="X"), ): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Usage: ..." await set_role(update, context) message.reply_text.assert_called_once_with("Usage: ...") mock_t.assert_called_with("en", "set_role.usage") @pytest.mark.asyncio async def test_set_role_user_not_found_replies_user_not_found(): """set_role: target user not found -> reply set_role.user_not_found.""" message = MagicMock() message.reply_text = AsyncMock() message.reply_to_message = MagicMock() message.reply_to_message.from_user = MagicMock() message.reply_to_message.from_user.id = 999 user = _make_user() update = _make_update(message=message, effective_user=user) context = MagicMock() context.args = ["admin"] with patch( "duty_teller.handlers.commands.is_admin_async", new_callable=AsyncMock, return_value=True, ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch( "duty_teller.handlers.commands.get_user_by_telegram_id", return_value=None, ): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "User not found" await set_role(update, context) message.reply_text.assert_called_once_with("User not found") mock_t.assert_called_with("en", "set_role.user_not_found") @pytest.mark.asyncio async def test_set_role_success_reply_replies_done(): """set_role: reply to user + role -> set_user_role success -> reply set_role.done.""" message = MagicMock() message.reply_text = AsyncMock() message.reply_to_message = MagicMock() message.reply_to_message.from_user = MagicMock() message.reply_to_message.from_user.id = 888 user = _make_user() update = _make_update(message=message, effective_user=user) context = MagicMock() context.args = ["admin"] mock_target = MagicMock() mock_target.id = 10 mock_target.full_name = "Target User" with patch( "duty_teller.handlers.commands.is_admin_async", new_callable=AsyncMock, return_value=True, ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch( "duty_teller.handlers.commands.session_scope", ) as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch( "duty_teller.handlers.commands.get_user_by_telegram_id", return_value=mock_target, ): with patch( "duty_teller.handlers.commands.set_user_role", return_value=mock_target, ): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Role set" await set_role(update, context) message.reply_text.assert_called_once_with("Role set") mock_t.assert_called_with("en", "set_role.done", name="Target User", role="admin") @pytest.mark.asyncio async def test_set_role_db_error_replies_error(): """set_role: set_user_role returns None -> reply set_role.error.""" message = MagicMock() message.reply_text = AsyncMock() message.reply_to_message = MagicMock() message.reply_to_message.from_user = MagicMock() message.reply_to_message.from_user.id = 777 user = _make_user() update = _make_update(message=message, effective_user=user) context = MagicMock() context.args = ["user"] mock_target = MagicMock() mock_target.id = 5 mock_target.full_name = "Someone" with patch( "duty_teller.handlers.commands.is_admin_async", new_callable=AsyncMock, return_value=True, ): with patch("duty_teller.handlers.commands.get_lang", return_value="en"): with patch("duty_teller.handlers.commands.session_scope") as mock_scope: mock_session = MagicMock() mock_scope.return_value.__enter__.return_value = mock_session mock_scope.return_value.__exit__.return_value = None with patch( "duty_teller.handlers.commands.get_user_by_telegram_id", return_value=mock_target, ): with patch( "duty_teller.handlers.commands.set_user_role", return_value=None, ): with patch("duty_teller.handlers.commands.t") as mock_t: mock_t.return_value = "Error" await set_role(update, context) message.reply_text.assert_called_once_with("Error") mock_t.assert_called_with("en", "set_role.error")