TelegramBotAdmin/tests/test_member_identity.py

from types import SimpleNamespace


def user(
    user_id: int,
    *,
    username: str | None,
    first_name: str,
    last_name: str | None = None,
):
    return SimpleNamespace(
        id=user_id,
        username=username,
        first_name=first_name,
        last_name=last_name,
        is_bot=False,
    )


async def test_identity_monitor_records_only_changes_after_the_first_observation(
    app_modules,
):
    identity = app_modules.load("wbb.services.member_identity")
    admin_db = app_modules.load("wbb.utils.dbadmin")
    points_db = app_modules.load("wbb.utils.dbpoints")
    chat_id = -100123

    await points_db.adjust_points(
        chat_id=chat_id,
        user_id=42,
        delta=10,
        source=points_db.SOURCE_ADMIN,
        idempotency_key="identity-account",
        actor_id="admin",
        reason="建立测试账户",
        username="old_name",
        first_name="小明",
        display_name="小明 同学",
    )
    await admin_db.update_managed_chat_settings(
        chat_id,
        {"identity_monitor": {"enabled": True, "notify_in_chat": False}},
    )

    first = await identity.observe_member_identity(
        chat_id=chat_id,
        user=user(42, username="old_name", first_name="小明", last_name="同学"),
    )
    changed = await identity.observe_member_identity(
        chat_id=chat_id,
        user=user(42, username="new_name", first_name="大明", last_name="同学"),
    )

    assert first is None
    assert changed["changed_fields"] == ["username", "display_name"]
    assert changed["before"]["display_name"] == "小明 同学"
    assert changed["after"]["display_name"] == "大明 同学"
    history, total = await admin_db.list_member_identity_changes(chat_id)
    assert total == 1
    assert history[0]["change_id"] == changed["change_id"]
    account = await points_db.get_point_account(chat_id, 42)
    assert account["display_name"] == "大明 同学"
    assert account["username"] == "new_name"
    assert app_modules.app.sent_messages == []


async def test_identity_monitor_can_notify_in_chat_and_ignores_disabled_periods(
    app_modules,
):
    identity = app_modules.load("wbb.services.member_identity")
    admin_db = app_modules.load("wbb.utils.dbadmin")
    chat_id = -100456

    await identity.observe_member_identity(
        chat_id=chat_id,
        user=user(7, username="before", first_name="旧昵称"),
    )
    await identity.observe_member_identity(
        chat_id=chat_id,
        user=user(7, username="while_off", first_name="关闭期间"),
    )
    await admin_db.update_managed_chat_settings(
        chat_id,
        {"identity_monitor": {"enabled": True, "notify_in_chat": True}},
    )

    event = await identity.observe_member_identity(
        chat_id=chat_id,
        user=user(7, username=None, first_name="新昵称"),
    )

    assert event["before"]["username"] == "while_off"
    assert event["after"]["username"] is None
    history, total = await admin_db.list_member_identity_changes(chat_id)
    assert total == 1
    assert history[0]["after"]["display_name"] == "新昵称"
    assert len(app_modules.app.sent_messages) == 1
    notified_chat_id, text = app_modules.app.sent_messages[0]
    assert notified_chat_id == chat_id
    assert "昵称：关闭期间 -> 新昵称" in text
    assert "用户名：@while_off -> 未设置" in text


async def test_joined_members_are_observed_even_when_the_sender_is_the_same_user(
    app_modules,
):
    identity = app_modules.load("wbb.services.member_identity")
    admin_db = app_modules.load("wbb.utils.dbadmin")
    joined = user(9, username="joined", first_name="新成员")
    message = SimpleNamespace(from_user=joined, new_chat_members=[joined, user(10, username=None, first_name="第二位")])

    changes = await identity.observe_message_member_identities(
        chat_id=-100789,
        message=message,
    )

    assert changes == []
    members = await admin_db.list_recent_chat_members(-100789, limit=10)
    assert {item["user_id"] for item in members} == {9, 10}
    first = next(item for item in members if item["user_id"] == 9)
    assert first["message_count"] == 1