Files
duty-teller/tests/test_handlers_commands.py
Nikolay Tatarinov 77a94fa91b
All checks were successful
CI / lint-and-test (push) Successful in 23s
feat: add team calendar ICS endpoint and related functionality
- Implemented a new API endpoint to generate an ICS calendar for team duty shifts, accessible via a valid token.
- Enhanced the `calendar_link` command to return both personal and team calendar URLs.
- Added a new function to build the team ICS file, ensuring each event includes the duty holder's name in the description.
- Updated tests to cover the new team calendar functionality, including validation for token formats and response content.
- Revised internationalization messages to reflect the new team calendar links.
2026-02-21 23:41:00 +03:00

566 lines
23 KiB
Python

"""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")