Nikolay Tatarinov
e3240d0981
- Added `bot_username` to settings for dynamic retrieval of the bot's username. - Implemented `_resolve_bot_username` function to fetch the bot's username if not set, improving user experience in group chats. - Updated `DutyWithUser` schema to include optional `phone` and `username` fields for enhanced duty information. - Enhanced API responses to include contact details for users, ensuring better communication. - Introduced a new current duty view in the web app, displaying active duty information along with contact options. - Updated CSS styles for better presentation of contact information in duty cards. - Added unit tests to verify the inclusion of contact details in API responses and the functionality of the current duty view.
1057 lines
47 KiB
Python
1057 lines
47 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, Forbidden
|
|
from telegram.constants import ChatMemberStatus
|
|
|
|
import duty_teller.config as config
|
|
from duty_teller.handlers import group_duty_pin as mod
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def no_mini_app_url():
|
|
"""Ensure BOT_USERNAME is empty so duty messages are sent without contact button (reply_markup=None)."""
|
|
with patch.object(config, "BOT_USERNAME", ""):
|
|
yield
|
|
|
|
|
|
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 ---
|
|
|
|
|
|
def test_get_contact_button_markup_empty_username_returns_none():
|
|
"""_get_contact_button_markup: BOT_USERNAME empty -> returns None."""
|
|
with patch.object(config, "BOT_USERNAME", ""):
|
|
assert mod._get_contact_button_markup("en") is None
|
|
|
|
|
|
def test_get_contact_button_markup_returns_markup_when_username_set():
|
|
"""_get_contact_button_markup: BOT_USERNAME set -> returns InlineKeyboardMarkup with t.me deep link (startapp=duty)."""
|
|
from telegram import InlineKeyboardMarkup
|
|
|
|
with patch.object(config, "BOT_USERNAME", "MyDutyBot"):
|
|
with patch.object(mod, "t", return_value="View contacts"):
|
|
result = mod._get_contact_button_markup("en")
|
|
assert result is not None
|
|
assert isinstance(result, InlineKeyboardMarkup)
|
|
assert len(result.inline_keyboard) == 1
|
|
assert len(result.inline_keyboard[0]) == 1
|
|
btn = result.inline_keyboard[0][0]
|
|
assert btn.text == "View contacts"
|
|
assert btn.url.startswith("https://t.me/")
|
|
assert "startapp=duty" in btn.url
|
|
assert btn.url == "https://t.me/MyDutyBot?startapp=duty"
|
|
|
|
|
|
@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_sends_new_unpins_pins_saves_schedules_next():
|
|
"""update_group_pin: with message_id and text, sends new message, unpins, pins new, saves id, deletes old, schedules next."""
|
|
new_msg = MagicMock()
|
|
new_msg.message_id = 999
|
|
context = MagicMock()
|
|
context.job = MagicMock()
|
|
context.job.data = {"chat_id": 123}
|
|
context.bot = MagicMock()
|
|
context.bot.send_message = AsyncMock(return_value=new_msg)
|
|
context.bot.unpin_chat_message = AsyncMock()
|
|
context.bot.pin_chat_message = AsyncMock()
|
|
context.bot.delete_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.object(config, "DUTY_PIN_NOTIFY", True):
|
|
with patch.object(mod, "_sync_is_trusted", return_value=True):
|
|
with patch.object(
|
|
mod, "_sync_get_pin_refresh_data", return_value=(1, "Current duty", None)
|
|
):
|
|
with patch.object(mod, "_schedule_next_update", AsyncMock()):
|
|
with patch.object(mod, "_sync_save_pin") as mock_save:
|
|
await mod.update_group_pin(context)
|
|
context.bot.send_message.assert_called_once_with(
|
|
chat_id=123, text="Current duty", reply_markup=None
|
|
)
|
|
context.bot.unpin_chat_message.assert_called_once_with(chat_id=123)
|
|
context.bot.pin_chat_message.assert_called_once_with(
|
|
chat_id=123, message_id=999, disable_notification=False
|
|
)
|
|
mock_save.assert_called_once_with(123, 999)
|
|
context.bot.delete_message.assert_called_once_with(chat_id=123, message_id=1)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_group_pin_delete_message_raises_bad_request_still_schedules():
|
|
"""update_group_pin: delete_message raises BadRequest -> save and schedule still done, log warning."""
|
|
new_msg = MagicMock()
|
|
new_msg.message_id = 999
|
|
context = MagicMock()
|
|
context.job = MagicMock()
|
|
context.job.data = {"chat_id": 123}
|
|
context.bot = MagicMock()
|
|
context.bot.send_message = AsyncMock(return_value=new_msg)
|
|
context.bot.unpin_chat_message = AsyncMock()
|
|
context.bot.pin_chat_message = AsyncMock()
|
|
context.bot.delete_message = AsyncMock(side_effect=BadRequest("Message to delete not found"))
|
|
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(config, "DUTY_PIN_NOTIFY", True):
|
|
with patch.object(mod, "_sync_is_trusted", return_value=True):
|
|
with patch.object(
|
|
mod, "_sync_get_pin_refresh_data", return_value=(1, "Current duty", None)
|
|
):
|
|
with patch.object(mod, "_schedule_next_update", AsyncMock()) as mock_schedule:
|
|
with patch.object(mod, "_sync_save_pin") as mock_save:
|
|
with patch.object(mod, "logger") as mock_logger:
|
|
await mod.update_group_pin(context)
|
|
mock_save.assert_called_once_with(123, 999)
|
|
mock_schedule.assert_called_once_with(context.application, 123, None)
|
|
mock_logger.warning.assert_called_once()
|
|
assert "Could not delete old pinned message" in mock_logger.warning.call_args[0][0]
|
|
|
|
|
|
@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 send_message."""
|
|
context = MagicMock()
|
|
context.job = MagicMock()
|
|
context.job.data = {"chat_id": 456}
|
|
context.bot = MagicMock()
|
|
context.bot.send_message = AsyncMock()
|
|
|
|
with patch.object(mod, "_sync_is_trusted", return_value=True):
|
|
with patch.object(
|
|
mod, "_sync_get_pin_refresh_data", return_value=(None, "No duty", None)
|
|
):
|
|
await mod.update_group_pin(context)
|
|
context.bot.send_message.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_group_pin_send_raises_no_unpin_pin_schedule_still_called():
|
|
"""update_group_pin: send_message raises BadRequest -> no unpin/pin/save, _schedule_next_update still called."""
|
|
context = MagicMock()
|
|
context.job = MagicMock()
|
|
context.job.data = {"chat_id": 111}
|
|
context.bot = MagicMock()
|
|
context.bot.send_message = AsyncMock(side_effect=BadRequest("Chat not found"))
|
|
context.bot.unpin_chat_message = AsyncMock()
|
|
context.bot.pin_chat_message = AsyncMock()
|
|
context.application = MagicMock()
|
|
|
|
with patch.object(mod, "_sync_is_trusted", return_value=True):
|
|
with patch.object(
|
|
mod, "_sync_get_pin_refresh_data", return_value=(2, "Text", None)
|
|
):
|
|
with patch.object(mod, "_schedule_next_update", AsyncMock()) as mock_schedule:
|
|
await mod.update_group_pin(context)
|
|
context.bot.unpin_chat_message.assert_not_called()
|
|
context.bot.pin_chat_message.assert_not_called()
|
|
mock_schedule.assert_called_once_with(context.application, 111, None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_group_pin_repin_raises_still_schedules_next():
|
|
"""update_group_pin: send_message ok, unpin or pin raises -> no _sync_save_pin, schedule still called, log."""
|
|
new_msg = MagicMock()
|
|
new_msg.message_id = 888
|
|
context = MagicMock()
|
|
context.job = MagicMock()
|
|
context.job.data = {"chat_id": 222}
|
|
context.bot = MagicMock()
|
|
context.bot.send_message = AsyncMock(return_value=new_msg)
|
|
context.bot.unpin_chat_message = AsyncMock(
|
|
side_effect=Forbidden("Not enough rights")
|
|
)
|
|
context.bot.pin_chat_message = AsyncMock()
|
|
context.application = MagicMock()
|
|
|
|
with patch.object(config, "DUTY_PIN_NOTIFY", True):
|
|
with patch.object(mod, "_sync_is_trusted", return_value=True):
|
|
with patch.object(
|
|
mod, "_sync_get_pin_refresh_data", return_value=(3, "Text", None)
|
|
):
|
|
with patch.object(
|
|
mod, "_schedule_next_update", AsyncMock()
|
|
) as mock_schedule:
|
|
with patch.object(mod, "_sync_save_pin") as mock_save:
|
|
with patch.object(mod, "logger") as mock_logger:
|
|
await mod.update_group_pin(context)
|
|
context.bot.send_message.assert_called_once_with(
|
|
chat_id=222, text="Text", reply_markup=None
|
|
)
|
|
mock_save.assert_not_called()
|
|
mock_logger.warning.assert_called_once()
|
|
assert "Unpin or pin" in mock_logger.warning.call_args[0][0]
|
|
mock_schedule.assert_called_once_with(context.application, 222, None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_group_pin_duty_pin_notify_false_pins_silent():
|
|
"""update_group_pin: DUTY_PIN_NOTIFY False -> send new, unpin, pin new with disable_notification=True, save, delete old, schedule."""
|
|
new_msg = MagicMock()
|
|
new_msg.message_id = 777
|
|
context = MagicMock()
|
|
context.job = MagicMock()
|
|
context.job.data = {"chat_id": 333}
|
|
context.bot = MagicMock()
|
|
context.bot.send_message = AsyncMock(return_value=new_msg)
|
|
context.bot.unpin_chat_message = AsyncMock()
|
|
context.bot.pin_chat_message = AsyncMock()
|
|
context.bot.delete_message = AsyncMock()
|
|
context.application = MagicMock()
|
|
|
|
with patch.object(config, "DUTY_PIN_NOTIFY", False):
|
|
with patch.object(mod, "_sync_is_trusted", return_value=True):
|
|
with patch.object(
|
|
mod, "_sync_get_pin_refresh_data", return_value=(4, "Text", None)
|
|
):
|
|
with patch.object(
|
|
mod, "_schedule_next_update", AsyncMock()
|
|
) as mock_schedule:
|
|
with patch.object(mod, "_sync_save_pin") as mock_save:
|
|
await mod.update_group_pin(context)
|
|
context.bot.send_message.assert_called_once_with(
|
|
chat_id=333, text="Text", reply_markup=None
|
|
)
|
|
context.bot.unpin_chat_message.assert_called_once_with(chat_id=333)
|
|
context.bot.pin_chat_message.assert_called_once_with(
|
|
chat_id=333, message_id=777, disable_notification=True
|
|
)
|
|
mock_save.assert_called_once_with(333, 777)
|
|
context.bot.delete_message.assert_called_once_with(chat_id=333, message_id=4)
|
|
mock_schedule.assert_called_once_with(context.application, 333, 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_is_trusted", return_value=True):
|
|
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_untrusted_group_rejects():
|
|
"""pin_duty_cmd in untrusted group -> reply group.not_trusted, no send/pin."""
|
|
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.send_message = AsyncMock()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "_sync_is_trusted", return_value=False):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Not authorized"
|
|
await mod.pin_duty_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Not authorized")
|
|
mock_t.assert_called_with("en", "group.not_trusted")
|
|
context.bot.send_message.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pin_duty_cmd_no_message_id_creates_sends_pins_saves_schedules_replies_pinned():
|
|
"""pin_duty_cmd: no pin record -> send_message, pin, save_pin, schedule, 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()
|
|
new_msg = MagicMock()
|
|
new_msg.message_id = 42
|
|
context.bot.send_message = AsyncMock(return_value=new_msg)
|
|
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, "_sync_is_trusted", return_value=True):
|
|
with patch.object(mod, "_sync_get_message_id", return_value=None):
|
|
with patch.object(
|
|
mod, "_get_duty_message_text_sync", return_value="Duty text"
|
|
):
|
|
with patch.object(mod, "_sync_save_pin") as mock_save:
|
|
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 = "Pinned"
|
|
await mod.pin_duty_cmd(update, context)
|
|
context.bot.send_message.assert_called_once_with(
|
|
chat_id=100, text="Duty text", reply_markup=None
|
|
)
|
|
context.bot.pin_chat_message.assert_called_once_with(
|
|
chat_id=100, message_id=42, disable_notification=True
|
|
)
|
|
mock_save.assert_called_once_with(100, 42)
|
|
update.message.reply_text.assert_called_once_with("Pinned")
|
|
mock_t.assert_called_with("en", "pin_duty.pinned")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pin_duty_cmd_no_message_id_send_message_raises_replies_failed():
|
|
"""pin_duty_cmd: no pin record, send_message raises BadRequest -> 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.send_message = AsyncMock(side_effect=BadRequest("Chat not found"))
|
|
context.application = MagicMock()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "_sync_is_trusted", return_value=True):
|
|
with patch.object(mod, "_sync_get_message_id", return_value=None):
|
|
with patch.object(mod, "_get_duty_message_text_sync", return_value="Duty"):
|
|
with patch.object(mod, "_sync_save_pin") as mock_save:
|
|
with patch.object(
|
|
mod, "_schedule_next_update", AsyncMock()
|
|
) as mock_schedule:
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Failed"
|
|
await mod.pin_duty_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Failed")
|
|
mock_t.assert_called_with("en", "pin_duty.failed")
|
|
mock_save.assert_not_called()
|
|
mock_schedule.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pin_duty_cmd_no_message_id_pin_raises_saves_and_replies_could_not_pin():
|
|
"""pin_duty_cmd: no pin record, pin_chat_message raises -> save pin, reply could_not_pin_make_admin."""
|
|
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()
|
|
new_msg = MagicMock()
|
|
new_msg.message_id = 43
|
|
context.bot.send_message = AsyncMock(return_value=new_msg)
|
|
context.bot.pin_chat_message = AsyncMock(side_effect=Forbidden("Not enough rights"))
|
|
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, "_sync_is_trusted", return_value=True):
|
|
with patch.object(mod, "_sync_get_message_id", return_value=None):
|
|
with patch.object(mod, "_get_duty_message_text_sync", return_value="Duty"):
|
|
with patch.object(mod, "_sync_save_pin") as mock_save:
|
|
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.pin_duty_cmd(update, context)
|
|
context.bot.send_message.assert_called_once_with(
|
|
chat_id=100, text="Duty", reply_markup=None
|
|
)
|
|
mock_save.assert_called_once_with(100, 43)
|
|
update.message.reply_text.assert_called_once_with("Make me admin to pin")
|
|
mock_t.assert_called_with("en", "pin_duty.could_not_pin_make_admin")
|
|
|
|
|
|
@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_is_trusted", return_value=True):
|
|
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")
|
|
|
|
|
|
# --- refresh_pin_cmd ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_pin_cmd_private_replies_group_only():
|
|
"""refresh_pin_cmd in private chat -> reply refresh_pin.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.refresh_pin_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Group only")
|
|
mock_t.assert_called_with("en", "refresh_pin.group_only")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_pin_cmd_group_updated_replies_updated():
|
|
"""refresh_pin_cmd in group with pin record -> _refresh_pin_for_chat returns updated -> reply refresh_pin.updated."""
|
|
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_is_trusted", return_value=True):
|
|
with patch.object(
|
|
mod, "_refresh_pin_for_chat", AsyncMock(return_value="updated")
|
|
):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Updated"
|
|
await mod.refresh_pin_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Updated")
|
|
mock_t.assert_called_with("en", "refresh_pin.updated")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_pin_cmd_group_no_message_replies_no_message():
|
|
"""refresh_pin_cmd: no pin record -> _refresh_pin_for_chat returns no_message -> reply refresh_pin.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_is_trusted", return_value=True):
|
|
with patch.object(
|
|
mod, "_refresh_pin_for_chat", AsyncMock(return_value="no_message")
|
|
):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "No message"
|
|
await mod.refresh_pin_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("No message")
|
|
mock_t.assert_called_with("en", "refresh_pin.no_message")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_pin_cmd_group_edit_raises_replies_failed():
|
|
"""refresh_pin_cmd: _refresh_pin_for_chat returns failed (e.g. BadRequest) -> reply refresh_pin.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()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "_sync_is_trusted", return_value=True):
|
|
with patch.object(
|
|
mod, "_refresh_pin_for_chat", AsyncMock(return_value="failed")
|
|
):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Failed"
|
|
await mod.refresh_pin_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Failed")
|
|
mock_t.assert_called_with("en", "refresh_pin.failed")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_pin_cmd_untrusted_group_rejects():
|
|
"""refresh_pin_cmd in untrusted group -> reply group.not_trusted, _refresh_pin_for_chat not called."""
|
|
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_is_trusted", return_value=False):
|
|
with patch.object(mod, "_refresh_pin_for_chat", AsyncMock()) as mock_refresh:
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Not authorized"
|
|
await mod.refresh_pin_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Not authorized")
|
|
mock_t.assert_called_with("en", "group.not_trusted")
|
|
mock_refresh.assert_not_called()
|
|
|
|
|
|
# --- 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, "_sync_is_trusted", return_value=True):
|
|
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", reply_markup=None
|
|
)
|
|
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_untrusted_group_does_not_send_duty():
|
|
"""my_chat_member_handler: bot added to untrusted group -> send group.not_trusted only, no duty message/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()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "_sync_is_trusted", return_value=False):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Not authorized"
|
|
await mod.my_chat_member_handler(update, context)
|
|
context.bot.send_message.assert_called_once_with(chat_id=200, text="Not authorized")
|
|
mock_t.assert_called_with("en", "group.not_trusted")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_my_chat_member_handler_trusted_group_sends_duty():
|
|
"""my_chat_member_handler: bot added to trusted group -> send duty, pin, schedule (same as test_my_chat_member_handler_bot_added_sends_pins_and_schedules)."""
|
|
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, "_sync_is_trusted", return_value=True):
|
|
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", reply_markup=None
|
|
)
|
|
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, "_sync_is_trusted", return_value=True):
|
|
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)
|
|
|
|
|
|
# --- _refresh_pin_for_chat untrusted ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_pin_for_chat_untrusted_removes_pin():
|
|
"""_refresh_pin_for_chat: when group not trusted -> delete_pin, remove job, unpin/delete message, return untrusted."""
|
|
context = MagicMock()
|
|
context.bot = MagicMock()
|
|
context.bot.unpin_chat_message = AsyncMock()
|
|
context.bot.delete_message = AsyncMock()
|
|
context.application = MagicMock()
|
|
context.application.job_queue = MagicMock()
|
|
mock_job = MagicMock()
|
|
context.application.job_queue.get_jobs_by_name = MagicMock(return_value=[mock_job])
|
|
|
|
with patch.object(mod, "_sync_is_trusted", return_value=False):
|
|
with patch.object(mod, "_sync_get_message_id", return_value=11):
|
|
with patch.object(mod, "_sync_delete_pin") as mock_delete_pin:
|
|
result = await mod._refresh_pin_for_chat(context, 100)
|
|
assert result == "untrusted"
|
|
mock_delete_pin.assert_called_once_with(100)
|
|
context.application.job_queue.get_jobs_by_name.assert_called_once_with("duty_pin_100")
|
|
mock_job.schedule_removal.assert_called_once()
|
|
context.bot.unpin_chat_message.assert_called_once_with(chat_id=100)
|
|
context.bot.delete_message.assert_called_once_with(chat_id=100, message_id=11)
|
|
|
|
|
|
# --- trust_group_cmd / untrust_group_cmd ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trust_group_cmd_non_admin_rejects():
|
|
"""trust_group_cmd: non-admin -> reply import.admin_only."""
|
|
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()
|
|
update.effective_user.id = 111
|
|
context = MagicMock()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "is_admin_async", AsyncMock(return_value=False)):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Admin only"
|
|
await mod.trust_group_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Admin only")
|
|
mock_t.assert_called_with("en", "import.admin_only")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trust_group_cmd_admin_adds_group():
|
|
"""trust_group_cmd: admin in group, group not yet trusted -> _sync_trust_group, reply added, then send+pin if no pin."""
|
|
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()
|
|
update.effective_user.id = 111
|
|
context = MagicMock()
|
|
context.bot = MagicMock()
|
|
new_msg = MagicMock()
|
|
new_msg.message_id = 50
|
|
context.bot.send_message = AsyncMock(return_value=new_msg)
|
|
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, "is_admin_async", AsyncMock(return_value=True)):
|
|
with patch.object(mod, "_sync_trust_group", return_value=False):
|
|
with patch.object(mod, "_sync_get_message_id", return_value=None):
|
|
with patch.object(mod, "_get_duty_message_text_sync", return_value="Duty text"):
|
|
with patch.object(mod, "_sync_save_pin") as mock_save:
|
|
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 = "Added"
|
|
with patch.object(
|
|
config, "DUTY_PIN_NOTIFY", False
|
|
):
|
|
await mod.trust_group_cmd(update, context)
|
|
update.message.reply_text.assert_any_call("Added")
|
|
mock_t.assert_any_call("en", "trust_group.added")
|
|
context.bot.send_message.assert_called_once_with(
|
|
chat_id=100, text="Duty text", reply_markup=None
|
|
)
|
|
context.bot.pin_chat_message.assert_called_once_with(
|
|
chat_id=100, message_id=50, disable_notification=True
|
|
)
|
|
mock_save.assert_called_once_with(100, 50)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trust_group_cmd_admin_already_trusted_replies_already_trusted():
|
|
"""trust_group_cmd: admin, group already trusted -> reply already_trusted, no send/pin."""
|
|
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()
|
|
update.effective_user.id = 111
|
|
context = MagicMock()
|
|
context.bot = MagicMock()
|
|
context.bot.send_message = AsyncMock()
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "is_admin_async", AsyncMock(return_value=True)):
|
|
with patch.object(mod, "_sync_trust_group", return_value=True):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Already trusted"
|
|
await mod.trust_group_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Already trusted")
|
|
mock_t.assert_called_with("en", "trust_group.already_trusted")
|
|
context.bot.send_message.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_untrust_group_cmd_removes_group():
|
|
"""untrust_group_cmd: admin, trusted group with pin -> remove from trusted, delete pin, remove job, unpin/delete message, reply removed."""
|
|
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()
|
|
update.effective_user.id = 111
|
|
context = MagicMock()
|
|
context.bot = MagicMock()
|
|
context.bot.unpin_chat_message = AsyncMock()
|
|
context.bot.delete_message = AsyncMock()
|
|
context.application = MagicMock()
|
|
mock_job = MagicMock()
|
|
context.application.job_queue = MagicMock()
|
|
context.application.job_queue.get_jobs_by_name = MagicMock(return_value=[mock_job])
|
|
|
|
with patch("duty_teller.handlers.group_duty_pin.get_lang", return_value="en"):
|
|
with patch.object(mod, "is_admin_async", AsyncMock(return_value=True)):
|
|
with patch.object(mod, "_sync_untrust_group", return_value=(True, 99)):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Removed"
|
|
await mod.untrust_group_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Removed")
|
|
mock_t.assert_called_with("en", "untrust_group.removed")
|
|
context.application.job_queue.get_jobs_by_name.assert_called_once_with("duty_pin_100")
|
|
mock_job.schedule_removal.assert_called_once()
|
|
context.bot.unpin_chat_message.assert_called_once_with(chat_id=100)
|
|
context.bot.delete_message.assert_called_once_with(chat_id=100, message_id=99)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_untrust_group_cmd_not_trusted_replies_not_trusted():
|
|
"""untrust_group_cmd: group not in trusted list -> reply not_trusted."""
|
|
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, "is_admin_async", AsyncMock(return_value=True)):
|
|
with patch.object(mod, "_sync_untrust_group", return_value=(False, None)):
|
|
with patch("duty_teller.handlers.group_duty_pin.t") as mock_t:
|
|
mock_t.return_value = "Not trusted"
|
|
await mod.untrust_group_cmd(update, context)
|
|
update.message.reply_text.assert_called_once_with("Not trusted")
|
|
mock_t.assert_called_with("en", "untrust_group.not_trusted")
|