Nikolay Tatarinov
15f80ee46b
All checks were successful
CI / lint-and-test (push) Successful in 23s
- Introduced new tests for the `import_duty_schedule_cmd` to verify behavior when no message or effective user is present, ensuring proper early returns. - Added tests for admin checks to confirm that only authorized users can initiate duty schedule imports, enhancing security. - Implemented error handling tests for the `handle_handover_time_text` function to ensure appropriate responses for invalid time formats and non-admin users. - Enhanced overall test coverage for the duty schedule import functionality, contributing to improved reliability and maintainability of the codebase. - Updated the `.coverage` file to reflect the latest coverage metrics.
415 lines
17 KiB
Python
415 lines
17 KiB
Python
"""Tests for duty_teller.handlers.group_duty_pin (sync wrappers, update_group_pin, pin_duty_cmd)."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from telegram.error import BadRequest
|
|
from telegram.constants import ChatMemberStatus
|
|
|
|
from duty_teller.handlers import group_duty_pin as mod
|
|
|
|
|
|
class TestSyncWrappers:
|
|
"""Tests for _get_duty_message_text_sync, _sync_save_pin, _sync_delete_pin, _sync_get_message_id, _get_all_pin_chat_ids_sync."""
|
|
|
|
def test_get_duty_message_text_sync(self):
|
|
with patch.object(mod, "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.object(mod, "get_duty_message_text", return_value="Duty text"):
|
|
result = mod._get_duty_message_text_sync("en")
|
|
assert result == "Duty text"
|
|
mock_scope.assert_called_once()
|
|
|
|
def test_sync_save_pin(self):
|
|
with patch.object(mod, "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.object(mod, "save_pin") as mock_save:
|
|
mod._sync_save_pin(100, 42)
|
|
mock_save.assert_called_once_with(mock_session, 100, 42)
|
|
|
|
def test_sync_delete_pin(self):
|
|
with patch.object(mod, "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.object(mod, "delete_pin") as mock_delete:
|
|
mod._sync_delete_pin(200)
|
|
mock_delete.assert_called_once_with(mock_session, 200)
|
|
|
|
def test_sync_get_message_id(self):
|
|
with patch.object(mod, "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.object(mod, "get_message_id", return_value=99):
|
|
result = mod._sync_get_message_id(300)
|
|
assert result == 99
|
|
|
|
def test_get_all_pin_chat_ids_sync(self):
|
|
with patch.object(mod, "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.object(mod, "get_all_pin_chat_ids", return_value=[10, 20]):
|
|
result = mod._get_all_pin_chat_ids_sync()
|
|
assert result == [10, 20]
|
|
|
|
def test_get_next_shift_end_sync(self):
|
|
"""_get_next_shift_end_sync: uses session_scope and get_next_shift_end_utc."""
|
|
with patch.object(mod, "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
|
|
when = datetime(2026, 2, 22, 9, 0, tzinfo=timezone.utc)
|
|
with patch.object(mod, "get_next_shift_end_utc", return_value=when):
|
|
result = mod._get_next_shift_end_sync()
|
|
assert result == when
|
|
mock_scope.assert_called_once()
|
|
|
|
|
|
# --- _schedule_next_update ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_next_update_job_queue_none_returns_early():
|
|
"""_schedule_next_update: job_queue is None -> log and return, no run_once."""
|
|
application = MagicMock()
|
|
application.job_queue = None
|
|
with patch.object(mod, "logger") as mock_logger:
|
|
await mod._schedule_next_update(application, 123, datetime.now(timezone.utc))
|
|
mock_logger.warning.assert_called_once()
|
|
assert "Job queue not available" in mock_logger.warning.call_args[0][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_next_update_when_utc_not_none_runs_once_with_delay():
|
|
"""_schedule_next_update: when_utc set -> remove old jobs, run_once with delay."""
|
|
application = MagicMock()
|
|
job_queue = MagicMock()
|
|
job_queue.get_jobs_by_name = MagicMock(return_value=[])
|
|
job_queue.run_once = MagicMock()
|
|
application.job_queue = job_queue
|
|
when_utc = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(hours=1)
|
|
|
|
await mod._schedule_next_update(application, 456, when_utc)
|
|
|
|
job_queue.get_jobs_by_name.assert_called_once_with("duty_pin_456")
|
|
job_queue.run_once.assert_called_once()
|
|
call_kw = job_queue.run_once.call_args[1]
|
|
assert call_kw["data"] == {"chat_id": 456}
|
|
assert call_kw["name"] == "duty_pin_456"
|
|
assert call_kw["when"].total_seconds() >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_next_update_when_utc_none_runs_once_with_retry_delay():
|
|
"""_schedule_next_update: when_utc None -> run_once with RETRY_WHEN_NO_DUTY_MINUTES."""
|
|
application = MagicMock()
|
|
job_queue = MagicMock()
|
|
job_queue.get_jobs_by_name = MagicMock(return_value=[])
|
|
job_queue.run_once = MagicMock()
|
|
application.job_queue = job_queue
|
|
|
|
await mod._schedule_next_update(application, 789, None)
|
|
|
|
job_queue.run_once.assert_called_once()
|
|
call_kw = job_queue.run_once.call_args[1]
|
|
assert call_kw["when"] == timedelta(minutes=mod.RETRY_WHEN_NO_DUTY_MINUTES)
|
|
assert call_kw["data"] == {"chat_id": 789}
|
|
assert call_kw["name"] == "duty_pin_789"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_group_pin_edits_message_and_schedules_next():
|
|
"""update_group_pin: with message_id and text, edits message and schedules next update."""
|
|
context = MagicMock()
|
|
context.job = MagicMock()
|
|
context.job.data = {"chat_id": 123}
|
|
context.bot = MagicMock()
|
|
context.bot.edit_message_text = AsyncMock()
|
|
context.application = MagicMock()
|
|
context.application.job_queue = MagicMock()
|
|
context.application.job_queue.get_jobs_by_name = MagicMock(return_value=[])
|
|
context.application.job_queue.run_once = MagicMock()
|
|
|
|
with patch.object(mod, "_sync_get_message_id", return_value=1):
|
|
with patch.object(
|
|
mod, "_get_duty_message_text_sync", return_value="Current duty"
|
|
):
|
|
with patch.object(mod, "_get_next_shift_end_sync", return_value=None):
|
|
with patch.object(mod, "_schedule_next_update", AsyncMock()):
|
|
await mod.update_group_pin(context)
|
|
context.bot.edit_message_text.assert_called_once_with(
|
|
chat_id=123, message_id=1, text="Current duty"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_group_pin_no_message_id_skips():
|
|
"""update_group_pin: when no pin record (message_id None), does not edit."""
|
|
context = MagicMock()
|
|
context.job = MagicMock()
|
|
context.job.data = {"chat_id": 456}
|
|
context.bot = MagicMock()
|
|
context.bot.edit_message_text = AsyncMock()
|
|
|
|
with patch.object(mod, "_sync_get_message_id", return_value=None):
|
|
await mod.update_group_pin(context)
|
|
context.bot.edit_message_text.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_group_pin_edit_raises_bad_request_still_schedules_next():
|
|
"""update_group_pin: edit_message_text BadRequest -> log, _schedule_next_update still called."""
|
|
context = MagicMock()
|
|
context.job = MagicMock()
|
|
context.job.data = {"chat_id": 111}
|
|
context.bot = MagicMock()
|
|
context.bot.edit_message_text = AsyncMock(
|
|
side_effect=BadRequest("Message not modified")
|
|
)
|
|
context.application = MagicMock()
|
|
|
|
with patch.object(mod, "_sync_get_message_id", return_value=2):
|
|
with patch.object(mod, "_get_duty_message_text_sync", return_value="Text"):
|
|
with patch.object(mod, "_get_next_shift_end_sync", return_value=None):
|
|
with patch.object(
|
|
mod, "_schedule_next_update", AsyncMock()
|
|
) as mock_schedule:
|
|
await mod.update_group_pin(context)
|
|
mock_schedule.assert_called_once_with(context.application, 111, None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pin_duty_cmd_group_only_reply():
|
|
"""pin_duty_cmd in private chat -> reply group_only."""
|
|
update = MagicMock()
|
|
update.message = MagicMock()
|
|
update.message.reply_text = AsyncMock()
|
|
update.effective_chat = MagicMock()
|
|
update.effective_chat.type = "private"
|
|
update.effective_chat.id = 1
|
|
update.effective_user = MagicMock()
|
|
context = MagicMock()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Group only"
|
|
await mod.pin_duty_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Group only")
|
|
mock_t.assert_called_with("en", "pin_duty.group_only")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pin_duty_cmd_group_pins_and_replies_pinned():
|
|
"""pin_duty_cmd in group with existing pin record -> pin and reply pinned."""
|
|
update = MagicMock()
|
|
update.message = MagicMock()
|
|
update.message.reply_text = AsyncMock()
|
|
update.effective_chat = MagicMock()
|
|
update.effective_chat.type = "group"
|
|
update.effective_chat.id = 100
|
|
update.effective_user = MagicMock()
|
|
context = MagicMock()
|
|
context.bot = MagicMock()
|
|
context.bot.pin_chat_message = AsyncMock()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "_sync_get_message_id", return_value=5):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Pinned"
|
|
await mod.pin_duty_cmd(update, context)
|
|
context.bot.pin_chat_message.assert_called_once_with(
|
|
chat_id=100, message_id=5, disable_notification=True
|
|
)
|
|
update.message.reply_text.assert_called_once_with("Pinned")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pin_duty_cmd_no_message_id_replies_no_message():
|
|
"""pin_duty_cmd: no pin record (_sync_get_message_id -> None) -> reply pin_duty.no_message."""
|
|
update = MagicMock()
|
|
update.message = MagicMock()
|
|
update.message.reply_text = AsyncMock()
|
|
update.effective_chat = MagicMock()
|
|
update.effective_chat.type = "group"
|
|
update.effective_chat.id = 100
|
|
update.effective_user = MagicMock()
|
|
context = MagicMock()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "_sync_get_message_id", return_value=None):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "No message to pin"
|
|
await mod.pin_duty_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("No message to pin")
|
|
mock_t.assert_called_with("en", "pin_duty.no_message")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pin_duty_cmd_pin_raises_replies_failed():
|
|
"""pin_duty_cmd: pin_chat_message raises -> reply pin_duty.failed."""
|
|
update = MagicMock()
|
|
update.message = MagicMock()
|
|
update.message.reply_text = AsyncMock()
|
|
update.effective_chat = MagicMock()
|
|
update.effective_chat.type = "group"
|
|
update.effective_chat.id = 100
|
|
update.effective_user = MagicMock()
|
|
context = MagicMock()
|
|
context.bot = MagicMock()
|
|
context.bot.pin_chat_message = AsyncMock(
|
|
side_effect=BadRequest("Not enough rights")
|
|
)
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "_sync_get_message_id", return_value=5):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Failed to pin"
|
|
await mod.pin_duty_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Failed to pin")
|
|
mock_t.assert_called_with("en", "pin_duty.failed")
|
|
|
|
|
|
# --- my_chat_member_handler ---
|
|
|
|
|
|
def _make_my_chat_member_update(
|
|
old_status,
|
|
new_status,
|
|
chat_id=100,
|
|
chat_type="group",
|
|
bot_id=999,
|
|
from_user_id=1,
|
|
):
|
|
"""Build Update with my_chat_member for group add/remove."""
|
|
update = MagicMock()
|
|
update.effective_user = MagicMock()
|
|
update.effective_user.id = from_user_id
|
|
update.effective_chat = MagicMock()
|
|
update.effective_chat.id = chat_id
|
|
update.effective_chat.type = chat_type
|
|
update.my_chat_member = MagicMock()
|
|
update.my_chat_member.old_chat_member = MagicMock()
|
|
update.my_chat_member.old_chat_member.status = old_status
|
|
update.my_chat_member.new_chat_member = MagicMock()
|
|
update.my_chat_member.new_chat_member.status = new_status
|
|
update.my_chat_member.new_chat_member.user = MagicMock()
|
|
update.my_chat_member.new_chat_member.user.id = bot_id
|
|
return update
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_my_chat_member_handler_bot_added_sends_pins_and_schedules():
|
|
"""my_chat_member_handler: bot added to group -> send_message, pin, save_pin, schedule."""
|
|
update = _make_my_chat_member_update(
|
|
old_status=ChatMemberStatus.LEFT,
|
|
new_status=ChatMemberStatus.ADMINISTRATOR,
|
|
chat_id=200,
|
|
bot_id=999,
|
|
)
|
|
context = MagicMock()
|
|
context.bot = MagicMock()
|
|
context.bot.id = 999
|
|
context.bot.send_message = AsyncMock(return_value=MagicMock(message_id=42))
|
|
context.bot.pin_chat_message = AsyncMock()
|
|
context.application = MagicMock()
|
|
context.application.job_queue = MagicMock()
|
|
context.application.job_queue.get_jobs_by_name = MagicMock(return_value=[])
|
|
context.application.job_queue.run_once = MagicMock()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "_get_duty_message_text_sync", return_value="Duty text"):
|
|
with patch.object(mod, "_sync_save_pin"):
|
|
with patch.object(mod, "_get_next_shift_end_sync", return_value=None):
|
|
with patch.object(mod, "_schedule_next_update", AsyncMock()):
|
|
await mod.my_chat_member_handler(update, context)
|
|
context.bot.send_message.assert_called_once_with(chat_id=200, text="Duty text")
|
|
context.bot.pin_chat_message.assert_called_once_with(
|
|
chat_id=200, message_id=42, disable_notification=True
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_my_chat_member_handler_pin_raises_sends_could_not_pin():
|
|
"""my_chat_member_handler: pin_chat_message raises -> send_message could_not_pin_make_admin."""
|
|
update = _make_my_chat_member_update(
|
|
old_status=ChatMemberStatus.LEFT,
|
|
new_status=ChatMemberStatus.MEMBER,
|
|
chat_id=201,
|
|
bot_id=999,
|
|
)
|
|
context = MagicMock()
|
|
context.bot = MagicMock()
|
|
context.bot.id = 999
|
|
context.bot.send_message = AsyncMock(return_value=MagicMock(message_id=43))
|
|
context.bot.pin_chat_message = AsyncMock(side_effect=BadRequest("Cannot pin"))
|
|
context.application = MagicMock()
|
|
context.application.job_queue = MagicMock()
|
|
context.application.job_queue.get_jobs_by_name = MagicMock(return_value=[])
|
|
context.application.job_queue.run_once = MagicMock()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "_get_duty_message_text_sync", return_value="Duty"):
|
|
with patch.object(mod, "_sync_save_pin"):
|
|
with patch.object(mod, "_get_next_shift_end_sync", return_value=None):
|
|
with patch.object(mod, "_schedule_next_update", AsyncMock()):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Make me admin to pin"
|
|
await mod.my_chat_member_handler(update, context)
|
|
assert context.bot.send_message.call_count >= 2
|
|
pin_hint_calls = [
|
|
c
|
|
for c in context.bot.send_message.call_args_list
|
|
if c[1].get("text") == "Make me admin to pin"
|
|
]
|
|
assert len(pin_hint_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_my_chat_member_handler_bot_removed_deletes_pin_and_jobs():
|
|
"""my_chat_member_handler: bot removed -> _sync_delete_pin, remove jobs by name."""
|
|
update = _make_my_chat_member_update(
|
|
old_status=ChatMemberStatus.ADMINISTRATOR,
|
|
new_status=ChatMemberStatus.LEFT,
|
|
chat_id=202,
|
|
bot_id=999,
|
|
)
|
|
context = MagicMock()
|
|
context.bot = MagicMock()
|
|
context.bot.id = 999
|
|
context.application = MagicMock()
|
|
job_queue = MagicMock()
|
|
mock_job = MagicMock()
|
|
job_queue.get_jobs_by_name = MagicMock(return_value=[mock_job])
|
|
context.application.job_queue = job_queue
|
|
|
|
with patch.object(mod, "_sync_delete_pin"):
|
|
await mod.my_chat_member_handler(update, context)
|
|
job_queue.get_jobs_by_name.assert_called_once_with("duty_pin_202")
|
|
mock_job.schedule_removal.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_restore_group_pin_jobs_calls_schedule_for_each_chat():
|
|
"""restore_group_pin_jobs: for each chat_id from _get_all_pin_chat_ids_sync, calls _schedule_next_update."""
|
|
application = MagicMock()
|
|
application.job_queue = MagicMock()
|
|
application.job_queue.get_jobs_by_name = MagicMock(return_value=[])
|
|
application.job_queue.run_once = MagicMock()
|
|
|
|
with patch.object(mod, "_get_all_pin_chat_ids_sync", return_value=[10, 20]):
|
|
with patch.object(mod, "_get_next_shift_end_sync", return_value=None):
|
|
with patch.object(
|
|
mod, "_schedule_next_update", AsyncMock()
|
|
) as mock_schedule:
|
|
await mod.restore_group_pin_jobs(application)
|
|
assert mock_schedule.call_count == 2
|
|
mock_schedule.assert_any_call(application, 10, None)
|
|
mock_schedule.assert_any_call(application, 20, None)
|