TelegramBotAdmin/tests/test_admin_shortcuts.py

from __future__ import annotations

from datetime import datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest


@pytest.mark.parametrize(
    ("text", "expected"),
    [
        ("封禁", ("/ban", ["ban"])),
        ("解封", ("/unban", ["unban"])),
        ("禁言", ("/mute", ["mute"])),
        ("解除禁言", ("/unmute", ["unmute"])),
        ("警告", ("/warn", ["warn"])),
        ("删除消息", ("/del", ["del"])),
        ("删除", ("/del", ["del"])),
        ("删", ("/del", ["del"])),
        ("禁言10分钟", ("/tmute 10m", ["tmute", "10m"])),
        ("  禁言1分钟  ", ("/tmute 1m", ["tmute", "1m"])),
        ("禁言１分钟", ("/tmute 1m", ["tmute", "1m"])),
        ("禁言\u200b1分钟", ("/tmute 1m", ["tmute", "1m"])),
        ("禁言 2 小时", ("/tmute 2h", ["tmute", "2h"])),
        ("禁言1天", ("/tmute 1d", ["tmute", "1d"])),
        ("踢出", ("/kick", ["kick"])),
        ("踢", ("/kick", ["kick"])),
        ("飞", ("/kick", ["kick"])),
    ],
)
def test_member_action_shortcut_parser(app_modules, text, expected):
    shortcuts = app_modules.load("wbb.services.member_shortcuts")
    assert shortcuts.member_action_shortcut_command(text) == expected


@pytest.mark.parametrize("text", ["禁言0分钟", "禁言100分钟", "封禁 原因", "踢他"])
def test_member_action_shortcut_rejects_ambiguous_text(app_modules, text):
    shortcuts = app_modules.load("wbb.services.member_shortcuts")
    assert shortcuts.member_action_shortcut_command(text) is None


@pytest.mark.asyncio
async def test_member_action_shortcut_executes_temporary_mute(
    app_modules, monkeypatch
):
    admin = app_modules.load("wbb.modules.admin")
    chat_id = -100123
    actor_id = 10
    target_id = 20
    privileges = SimpleNamespace(
        can_post_messages=False,
        can_edit_messages=False,
        can_delete_messages=True,
        can_restrict_members=True,
        can_promote_members=False,
        can_change_info=False,
        can_invite_users=False,
        can_pin_messages=False,
        can_manage_video_chats=False,
    )
    app_modules.app.members[(chat_id, actor_id)] = SimpleNamespace(
        privileges=privileges
    )
    app_modules.app.get_users = AsyncMock(
        return_value=SimpleNamespace(mention="@target")
    )
    monkeypatch.setattr(admin, "list_admins", AsyncMock(return_value=[]))

    restrict_member = AsyncMock()
    response = SimpleNamespace(chat=SimpleNamespace(id=chat_id), id=30)
    target_reply = AsyncMock(return_value=response)
    scheduled = []
    monkeypatch.setattr(
        admin,
        "schedule_group_cleanup",
        lambda *messages, **kwargs: scheduled.append((messages, kwargs)),
    )
    replied_message = SimpleNamespace(
        from_user=SimpleNamespace(id=target_id),
        sender_chat=None,
        reply_text=target_reply,
    )
    message = SimpleNamespace(
        text="  禁言１分钟\u200b  ",
        command=None,
        chat=SimpleNamespace(id=chat_id, restrict_member=restrict_member),
        from_user=SimpleNamespace(id=actor_id, mention="@admin"),
        sender_chat=None,
        reply_to_message=replied_message,
        reply_text=AsyncMock(),
    )

    assert await admin._member_action_shortcut_filter(None, None, message)
    await admin.member_action_shortcut(None, message)

    restrict_member.assert_awaited_once()
    args, kwargs = restrict_member.await_args
    assert args == (target_id,)
    assert kwargs["until_date"] > datetime.now()
    assert message.text == "  禁言１分钟\u200b  "
    assert message.command is None
    assert "禁言时长：** 1m" in target_reply.await_args.args[0]
    assert "自动删除" not in target_reply.await_args.args[0]
    assert scheduled == [
        ((message, replied_message, response), {"delay_seconds": 10})
    ]


@pytest.mark.asyncio
async def test_delete_message_shortcut_waits_before_cleanup(
    app_modules, monkeypatch
):
    admin = app_modules.load("wbb.modules.admin")
    chat_id = -100123
    actor_id = 10
    privileges = SimpleNamespace(
        can_post_messages=False,
        can_edit_messages=False,
        can_delete_messages=True,
        can_restrict_members=True,
        can_promote_members=False,
        can_change_info=False,
        can_invite_users=False,
        can_pin_messages=False,
        can_manage_video_chats=False,
    )
    app_modules.app.members[(chat_id, actor_id)] = SimpleNamespace(
        privileges=privileges
    )
    response = SimpleNamespace(chat=SimpleNamespace(id=chat_id), id=31)
    target_reply = AsyncMock(return_value=response)
    replied_message = SimpleNamespace(
        from_user=SimpleNamespace(id=20),
        sender_chat=None,
        reply_text=target_reply,
    )
    message = SimpleNamespace(
        text="删除消息",
        command=None,
        chat=SimpleNamespace(id=chat_id),
        from_user=SimpleNamespace(id=actor_id, mention="@admin"),
        sender_chat=None,
        reply_to_message=replied_message,
        reply_text=AsyncMock(return_value=response),
    )
    scheduled = []
    monkeypatch.setattr(
        admin,
        "schedule_group_cleanup",
        lambda *messages, **kwargs: scheduled.append((messages, kwargs)),
    )

    await admin.member_action_shortcut(None, message)

    message.reply_text.assert_awaited_once()
    assert "自动删除" not in message.reply_text.await_args.args[0]
    target_reply.assert_not_awaited()
    assert scheduled == [
        ((message, replied_message, response), {"delay_seconds": 10})
    ]