TelegramBotAdmin/tests/test_points_commands.py

from __future__ import annotations

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

import pytest


@pytest.fixture
def points_module(app_modules, monkeypatch):
    permissions = types.ModuleType("wbb.core.decorators.permissions")

    def admins_only(*_args, **_kwargs):
        return lambda function: function

    permissions.adminsOnly = admins_only
    monkeypatch.setitem(
        sys.modules, "wbb.core.decorators.permissions", permissions
    )
    module = app_modules.load("wbb.modules.points")
    return module


def _checkin_message() -> SimpleNamespace:
    response = SimpleNamespace(
        chat=SimpleNamespace(id=-100, type=SimpleNamespace(value="supergroup")),
        id=11,
    )
    return SimpleNamespace(
        chat=SimpleNamespace(id=-100, type=SimpleNamespace(value="supergroup")),
        id=10,
        from_user=SimpleNamespace(
            id=20,
            username="customer",
            first_name="顾客",
            last_name=None,
        ),
        reply_text=AsyncMock(return_value=response),
    )


@pytest.mark.parametrize("created", [True, False])
async def test_checkin_schedules_user_and_bot_messages_for_cleanup(
    points_module, monkeypatch, created
):
    message = _checkin_message()
    account = {"balance": 10}
    cleanup = []
    monkeypatch.setattr(
        points_module,
        "award_checkin",
        AsyncMock(return_value=(account, created)),
    )
    monkeypatch.setattr(
        points_module,
        "get_point_rules",
        AsyncMock(
            return_value={
                "checkin_points": 10,
                "checkin_success_message": (
                    "领取成功 +{points}，余额 {balance}，今日第 {rank} 个，"
                    "连续 {streak} 天，本月 {month_days} 天"
                ),
                "checkin_already_message": (
                    "今天领过了，余额 {balance}，今日第 {rank} 个，"
                    "连续 {streak} 天，本月 {month_days} 天"
                ),
            }
        ),
    )
    monkeypatch.setattr(
        points_module,
        "get_checkin_metrics",
        AsyncMock(return_value={"rank": 3, "streak": 5, "month_days": 12}),
    )
    monkeypatch.setattr(
        points_module,
        "schedule_group_cleanup",
        lambda *messages: cleanup.extend(messages),
    )

    response = await points_module.points_checkin_command(None, message)

    expected_message = (
        "领取成功 +10，余额 10，今日第 3 个，连续 5 天，本月 12 天"
        if created
        else "今天领过了，余额 10，今日第 3 个，连续 5 天，本月 12 天"
    )
    assert message.reply_text.await_args.args == (expected_message,)
    assert (
        message.reply_text.await_args.kwargs["parse_mode"]
        == points_module.ParseMode.DISABLED
    )
    assert cleanup == [message, response]


async def test_checkin_error_schedules_user_and_bot_messages_for_cleanup(
    points_module, monkeypatch
):
    message = _checkin_message()
    cleanup = []
    monkeypatch.setattr(
        points_module,
        "award_checkin",
        AsyncMock(side_effect=points_module.PointsError("签到暂不可用")),
    )
    monkeypatch.setattr(
        points_module,
        "schedule_group_cleanup",
        lambda *messages: cleanup.extend(messages),
    )

    response = await points_module.points_checkin_command(None, message)

    assert message.reply_text.await_args.args == ("签到暂不可用",)
    assert (
        message.reply_text.await_args.kwargs["parse_mode"]
        == points_module.ParseMode.DISABLED
    )
    assert cleanup == [message, response]