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_remove_unavailable_chat_only_deletes_management_index(app_modules): service = app_modules.load("wbb.services.chat_management") await app_modules.wbb.db.managed_chats.insert_one( { "bot_id": "primary", "chat_id": -100, "title": "机器人已退出的群", "accessible": False, } ) await app_modules.wbb.db.managed_chat_settings.insert_one( {"bot_id": "primary", "chat_id": -100, "welcome_enabled": True} ) assert await service.remove_unavailable_chat(-100) == { "chat_id": "-100", "removed": True, } assert await app_modules.wbb.db.managed_chats.find_one({"chat_id": -100}) is None assert ( await app_modules.wbb.db.managed_chat_settings.find_one({"chat_id": -100}) is not None ) async def test_remove_unavailable_chat_rejects_accessible_chat(app_modules): service = app_modules.load("wbb.services.chat_management") await app_modules.wbb.db.managed_chats.insert_one( { "bot_id": "primary", "chat_id": -100, "title": "仍可管理的群", "accessible": True, } ) with pytest.raises(service.ChatManagementError) as caught: await service.remove_unavailable_chat(-100) assert caught.value.code == "chat_still_accessible" assert caught.value.status == 409 assert await app_modules.wbb.db.managed_chats.find_one({"chat_id": -100}) async def test_remove_unavailable_chat_reports_missing_record(app_modules): service = app_modules.load("wbb.services.chat_management") with pytest.raises(service.ChatManagementError) as caught: await service.remove_unavailable_chat(-100) assert caught.value.code == "chat_not_found" assert caught.value.status == 404 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"]