TelegramBotAdmin/tests/test_points.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta

import pytest


def test_leaderboard_labels_never_mention_members(app_modules):
    identity = app_modules.load("wbb.services.member_identity")

    assert (
        identity.non_mention_account_name(
            {
                "user_id": 10,
                "username": "alice",
                "first_name": "小明",
                "display_name": "小明 同学",
            }
        )
        == "小明 同学"
    )
    fallback = identity.non_mention_account_name(
        {
            "user_id": 11,
            "username": "alice",
            "first_name": "",
            "display_name": "",
        }
    )
    assert fallback == "用户 11"
    assert "@" not in fallback
    assert "tg://user" not in fallback


async def test_adjustment_is_idempotent_and_balance_cannot_be_negative(app_modules):
    points = app_modules.load("wbb.utils.dbpoints")

    account, created = await points.adjust_points(
        chat_id=-100,
        user_id=10,
        delta=20,
        source=points.SOURCE_ADMIN,
        idempotency_key="adjust:1",
        actor_id="admin",
        reason="seed",
    )
    assert created is True
    assert account["balance"] == 20

    account, created = await points.adjust_points(
        chat_id=-100,
        user_id=10,
        delta=20,
        source=points.SOURCE_ADMIN,
        idempotency_key="adjust:1",
        actor_id="admin",
        reason="duplicate",
    )
    assert created is False
    assert account["balance"] == 20

    with pytest.raises(points.InsufficientPoints):
        await points.adjust_points(
            chat_id=-100,
            user_id=10,
            delta=-21,
            source=points.SOURCE_ADMIN,
            idempotency_key="adjust:2",
            reason="too much",
        )


async def test_noop_set_is_recorded_and_retry_does_not_overwrite_new_points(app_modules):
    points = app_modules.load("wbb.utils.dbpoints")
    await points.adjust_points(
        chat_id=-100,
        user_id=10,
        delta=20,
        source=points.SOURCE_ADMIN,
        idempotency_key="set-seed",
        reason="seed",
    )

    account, created = await points.set_points(
        chat_id=-100,
        user_id=10,
        balance=20,
        actor_id="admin",
        reason="confirm balance",
        idempotency_key="set-request",
    )
    assert created is True
    assert account["balance"] == 20

    await points.adjust_points(
        chat_id=-100,
        user_id=10,
        delta=5,
        source=points.SOURCE_ADMIN,
        idempotency_key="set-later-award",
        reason="later award",
    )
    account, created = await points.set_points(
        chat_id=-100,
        user_id=10,
        balance=20,
        actor_id="admin",
        reason="retried request",
        idempotency_key="set-request",
    )
    assert created is False
    assert account["balance"] == 25

    transaction = await points.get_point_transaction_by_key("set-request")
    assert transaction["delta"] == 0


async def test_checkin_uses_chat_timezone_day_boundary(app_modules):
    points = app_modules.load("wbb.utils.dbpoints")
    await points.set_point_rules(-100, {"enabled": True, "timezone": "Asia/Shanghai"})
    before_midnight = datetime(2026, 1, 1, 15, 59, tzinfo=UTC)

    account, created = await points.award_checkin(
        chat_id=-100,
        user_id=10,
        username="alice",
        first_name="Alice",
        now=before_midnight,
    )
    assert created is True
    assert account["balance"] == 10

    _, created = await points.award_checkin(
        chat_id=-100,
        user_id=10,
        username="alice",
        first_name="Alice",
        now=before_midnight + timedelta(seconds=30),
    )
    assert created is False

    account, created = await points.award_checkin(
        chat_id=-100,
        user_id=10,
        username="alice",
        first_name="Alice",
        now=before_midnight + timedelta(minutes=2),
    )
    assert created is True
    assert account["balance"] == 20


async def test_checkin_messages_are_normalized_and_persisted(app_modules):
    points = app_modules.load("wbb.utils.dbpoints")

    saved = await points.set_point_rules(
        -100,
        {
            "checkin_success_message": "  获得 {points}，余额 {balance}  ",
            "checkin_already_message": "",
        },
    )

    assert saved["checkin_success_message"] == "获得 {points}，余额 {balance}"
    assert (
        saved["checkin_already_message"]
        == points.DEFAULT_CHECKIN_ALREADY_MESSAGE
    )
    loaded = await points.get_point_rules(-100)
    assert loaded["checkin_success_message"] == "获得 {points}，余额 {balance}"


async def test_activity_cooldown_duplicate_content_and_daily_cap(app_modules):
    points = app_modules.load("wbb.utils.dbpoints")
    await points.set_point_rules(
        -100,
        {
            "enabled": True,
            "activity_points": 1,
            "activity_cooldown_seconds": 300,
            "activity_daily_cap": 2,
        },
    )
    now = datetime(2099, 1, 1, 4, 0, tzinfo=UTC)

    _, first = await points.award_activity(
        chat_id=-100,
        user_id=10,
        message_id=1,
        content="hello",
        username=None,
        first_name="Alice",
        now=now,
    )
    _, cooldown = await points.award_activity(
        chat_id=-100,
        user_id=10,
        message_id=2,
        content="another",
        username=None,
        first_name="Alice",
        now=now + timedelta(seconds=299),
    )
    _, repeated = await points.award_activity(
        chat_id=-100,
        user_id=10,
        message_id=3,
        content="hello",
        username=None,
        first_name="Alice",
        now=now + timedelta(seconds=301),
    )
    _, second = await points.award_activity(
        chat_id=-100,
        user_id=10,
        message_id=4,
        content="different",
        username=None,
        first_name="Alice",
        now=now + timedelta(seconds=301),
    )
    _, capped = await points.award_activity(
        chat_id=-100,
        user_id=10,
        message_id=5,
        content="third",
        username=None,
        first_name="Alice",
        now=now + timedelta(seconds=602),
    )

    assert (first, cooldown, repeated, second, capped) == (True, False, False, True, False)
    assert (await points.get_point_account(-100, 10))["balance"] == 2


async def test_upvote_pair_cooldown_and_reconciliation(app_modules):
    points = app_modules.load("wbb.utils.dbpoints")
    await points.set_point_rules(-100, {"enabled": True})
    now = datetime.now(UTC)

    _, first = await points.award_upvote(
        chat_id=-100,
        voter_id=20,
        target_id=10,
        message_id=1,
        username=None,
        first_name="Alice",
        now=now,
    )
    _, blocked = await points.award_upvote(
        chat_id=-100,
        voter_id=20,
        target_id=10,
        message_id=2,
        username=None,
        first_name="Alice",
        now=now + timedelta(hours=23),
    )
    _, next_day = await points.award_upvote(
        chat_id=-100,
        voter_id=20,
        target_id=10,
        message_id=3,
        username=None,
        first_name="Alice",
        now=now + timedelta(hours=25),
    )
    assert (first, blocked, next_day) == (True, False, True)

    await points.accountsdb.update_one(
        {"chat_id": -100, "user_id": 10}, {"$set": {"balance": 999}}
    )
    repaired = await points.reconcile_account(-100, 10)
    assert repaired["balance"] == 4