TelegramBotAdmin/tests/test_chat_management.py

from __future__ import annotations

from unittest.mock import AsyncMock

import pytest
from pyrogram.enums import ChatMemberStatus
from pyrogram.errors import ChatNotModified


def chat_overview() -> dict[str, object]:
    return {
        "chat_id": -100,
        "title": "旧标题",
        "description": "旧描述",
        "type": "supergroup",
        "accessible": True,
        "bot_status": "administrator",
        "bot_privileges": ["can_change_info"],
        "permissions": {},
    }


async def test_profile_update_does_not_refresh_after_success(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.chat_management")
    ensure_permission = AsyncMock()
    sync_managed_chat = AsyncMock(return_value=chat_overview())
    set_chat_title = AsyncMock()
    set_chat_description = AsyncMock()
    monkeypatch.setattr(service, "ensure_permission", ensure_permission)
    monkeypatch.setattr(service, "sync_managed_chat", sync_managed_chat)
    monkeypatch.setattr(service.app, "set_chat_title", set_chat_title, raising=False)
    monkeypatch.setattr(
        service.app, "set_chat_description", set_chat_description, raising=False
    )

    result = await service.update_chat_profile(
        -100, title="新标题", description="新描述"
    )

    ensure_permission.assert_awaited_once_with(
        -100, "can_change_info", actor_id=None
    )
    sync_managed_chat.assert_awaited_once_with(-100)
    set_chat_title.assert_awaited_once_with(-100, "新标题")
    set_chat_description.assert_awaited_once_with(-100, "新描述")
    assert result["title"] == "新标题"
    assert result["description"] == "新描述"


async def test_profile_update_skips_values_already_applied(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.chat_management")
    monkeypatch.setattr(service, "ensure_permission", AsyncMock())
    monkeypatch.setattr(
        service, "sync_managed_chat", AsyncMock(return_value=chat_overview())
    )
    set_chat_title = AsyncMock()
    set_chat_description = AsyncMock()
    monkeypatch.setattr(service.app, "set_chat_title", set_chat_title, raising=False)
    monkeypatch.setattr(
        service.app, "set_chat_description", set_chat_description, raising=False
    )

    result = await service.update_chat_profile(
        -100, title=" 旧标题 ", description="旧描述"
    )

    set_chat_title.assert_not_awaited()
    set_chat_description.assert_not_awaited()
    assert result["title"] == "旧标题"
    assert result["description"] == "旧描述"


async def test_profile_update_accepts_chat_not_modified(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.chat_management")
    monkeypatch.setattr(service, "ensure_permission", AsyncMock())
    monkeypatch.setattr(
        service, "sync_managed_chat", AsyncMock(return_value=chat_overview())
    )
    set_chat_title = AsyncMock(side_effect=ChatNotModified())
    monkeypatch.setattr(service.app, "set_chat_title", set_chat_title, raising=False)

    result = await service.update_chat_profile(-100, title="新标题")

    assert result["title"] == "新标题"


async def test_profile_update_wraps_real_telegram_failure(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.chat_management")
    monkeypatch.setattr(service, "ensure_permission", AsyncMock())
    monkeypatch.setattr(
        service, "sync_managed_chat", AsyncMock(return_value=chat_overview())
    )
    monkeypatch.setattr(
        service.app,
        "set_chat_title",
        AsyncMock(side_effect=RuntimeError("telegram unavailable")),
        raising=False,
    )

    with pytest.raises(service.ChatManagementError) as caught:
        await service.update_chat_profile(-100, title="新标题")

    assert caught.value.code == "chat_profile_update_failed"
    assert caught.value.status == 502
    assert str(caught.value) == "Telegram 未能更新群标题，请稍后重试。"


async def test_automation_settings_store_configurable_risk_rules(
    app_modules, monkeypatch
):
    service = app_modules.load("wbb.services.chat_management")
    monkeypatch.setattr(service, "ensure_permission", AsyncMock())

    result = await service.apply_automation_settings(
        -100,
        {
            "risk_rules": [
                {
                    "rule_id": "ads",
                    "name": "广告内容",
                    "enabled": True,
                    "keywords": ["广告", "做爱"],
                    "match_images": True,
                    "match_links": True,
                    "actions": ["delete", "kick"],
                    "duration_seconds": 7200,
                },
                {
                    "rule_id": "disabled-links",
                    "name": "停用链接规则",
                    "enabled": False,
                    "keywords": ["未生效"],
                    "match_links": True,
                    "actions": ["delete"],
                    "duration_seconds": 3600,
                },
            ],
            "identity_monitor": {"enabled": True, "notify_in_chat": True},
        },
    )

    assert result["risk_rules"][0] == {
        "rule_id": "ads",
        "name": "广告内容",
        "enabled": True,
        "keywords": ["做爱", "广告"],
        "match_images": True,
        "match_links": True,
        "actions": ["delete", "kick"],
        "duration_seconds": 7200,
    }
    assert len(result["risk_rules"]) == 2
    assert result["identity_monitor"] == {"enabled": True, "notify_in_chat": True}
    assert await service.get_blacklisted_words(-100) == ["做爱", "广告"]


async def test_automation_settings_reject_conflicting_member_actions(
    app_modules, monkeypatch
):
    service = app_modules.load("wbb.services.chat_management")
    monkeypatch.setattr(service, "ensure_permission", AsyncMock())

    with pytest.raises(service.ChatManagementError) as caught:
        await service.apply_automation_settings(
            -100,
            {
                "risk_rules": [
                    {
                        "rule_id": "ads",
                        "name": "广告内容",
                        "enabled": True,
                        "keywords": ["广告"],
                        "actions": ["warn", "ban"],
                    }
                ]
            },
        )

    assert caught.value.code == "invalid_risk_rules"


async def test_automation_settings_wrap_legacy_policy_and_blacklist(app_modules):
    service = app_modules.load("wbb.services.chat_management")
    dbadmin = app_modules.load("wbb.utils.dbadmin")
    dbfunctions = app_modules.load("wbb.utils.dbfunctions")
    await dbadmin.update_managed_chat_settings(
        -100,
        {
            "risk_control": {
                "enabled": True,
                "keywords": ["广告"],
                "match_links": True,
                "actions": ["delete", "ban"],
            }
        },
    )
    await dbfunctions.save_blacklist_filter(-100, "推广")

    result = await service.get_automation_settings(-100)

    assert "risk_control" not in result
    assert result["risk_rules"] == [
        {
            "rule_id": "legacy-risk-control",
            "name": "原风控策略",
            "enabled": True,
            "keywords": ["广告", "推广"],
            "match_images": False,
            "match_links": True,
            "actions": ["delete", "ban"],
            "duration_seconds": 3600,
        }
    ]


async def test_member_search_accepts_display_name_from_recent_activity(
    app_modules, monkeypatch
):
    service = app_modules.load("wbb.services.chat_management")
    dbadmin = app_modules.load("wbb.utils.dbadmin")
    await dbadmin.upsert_recent_chat_member(
        chat_id=-100,
        user_id=42,
        username=None,
        first_name="小明",
        last_name="同学",
        is_bot=False,
    )

    async def no_telegram_results():
        if False:
            yield None

    monkeypatch.setattr(
        service.app,
        "get_chat_members",
        lambda *_args, **_kwargs: no_telegram_results(),
        raising=False,
    )
    app_modules.app.members[(-100, 42)] = type(
        "Member",
        (),
        {
            "status": ChatMemberStatus.MEMBER,
            "privileges": None,
            "user": type(
                "User",
                (),
                {
                    "id": 42,
                    "username": None,
                    "first_name": "小明",
                    "last_name": "同学",
                    "is_bot": False,
                    "is_deleted": False,
                },
            )(),
            "custom_title": None,
            "until_date": None,
        },
    )()

    results = await service.search_chat_members(-100, "小明")

    assert [item["user"]["id"] for item in results] == ["42"]