from __future__ import annotations import json import pytest from aiohttp import CookieJar from aiohttp.test_utils import TestClient, TestServer def test_multiple_bot_profiles_are_atomic_and_redacted(app_modules, tmp_path): config = app_modules.load("wbb.admin.bot_config") path = tmp_path / "bots.json" first_token = "123456:" + "A" * 30 second_token = "654321:" + "B" * 30 config.update_telegram_config(path, {"api_id": 12345, "api_hash": "a" * 32}) first = config.create_bot_profile( path, { "label": "Primary", "bot_token": first_token, "sudo_users_id": "100 200 100", "message_dump_chat": -100123, "business_assistant": { "base_url": "https://models.example.com/v1", "api_key": "secret-assistant-key", "model": "support-model", }, }, ) second = config.create_bot_profile( path, {"label": "Secondary", "bot_token": second_token, "enabled": False}, ) status = config.telegram_config_status(path) assert status["api_ready"] is True assert len(status["bots"]) == 2 assert status["bots"][0]["sudo_users_id"] == ["100", "200"] assert status["bots"][0]["role_ids"] == ["full_access"] assert status["bots"][0]["permissions"] assert "teacher_directory.manage" in status["bots"][0]["permissions"] assert status["bots"][0]["business_assistant"] == { "base_url": "https://models.example.com/v1", "api_key_configured": True, "model": "support-model", "timeout_seconds": 30, "max_output_tokens": 600, "configured": True, } serialized = json.dumps(status) assert first_token not in serialized assert second_token not in serialized assert "a" * 32 not in serialized assert "secret-assistant-key" not in serialized assert path.stat().st_mode & 0o777 == 0o600 config.update_bot_profile(path, first["bot_id"], {"label": "Renamed", "bot_token": ""}) stored = config.get_bot_profile_secrets(path, first["bot_id"]) assert stored["bot_token"] == first_token assert stored["label"] == "Renamed" config.delete_bot_profile(path, second["bot_id"]) assert len(config.telegram_config_status(path)["bots"]) == 1 def test_custom_roles_are_assignable_and_permissions_are_merged(app_modules, tmp_path): config = app_modules.load("wbb.admin.bot_config") path = tmp_path / "bots.json" config.update_telegram_config(path, {"api_id": 12345, "api_hash": "a" * 32}) role = config.create_bot_role( path, { "name": "积分与抽奖", "description": "社区运营", "permissions": ["points.manage", "giveaways.manage"], }, ) profile = config.create_bot_profile( path, { "label": "Operator", "bot_token": "123456:" + "A" * 30, "role_ids": [role["role_id"], "karma_manager"], }, ) assert profile["role_ids"] == [role["role_id"], "karma_manager"] assert set(profile["permissions"]) == { "points.manage", "giveaways.manage", "karma.manage", } config.update_bot_role( path, role["role_id"], {"permissions": ["automation.manage"]}, ) updated = config.telegram_config_status(path)["bots"][0] assert set(updated["permissions"]) == {"automation.manage", "karma.manage"} with pytest.raises(config.BotConfigError) as error: config.delete_bot_role(path, role["role_id"]) assert error.value.code == "role_in_use" config.update_bot_profile(path, profile["bot_id"], {"role_ids": []}) config.delete_bot_role(path, role["role_id"]) assert all( item["role_id"] != role["role_id"] for item in config.telegram_config_status(path)["roles"] ) def test_bot_profile_validation(app_modules, tmp_path): config = app_modules.load("wbb.admin.bot_config") path = tmp_path / "bots.json" with pytest.raises(config.BotConfigError) as error: config.create_bot_profile(path, {"label": "Bad", "bot_token": "not-a-token"}) assert error.value.code == "invalid_bot_token" with pytest.raises(config.BotConfigError) as error: config.update_telegram_config( path, {"api_id": 12345, "api_hash": "123456:" + "A" * 30}, ) assert error.value.code == "invalid_api_hash" assert "不能使用机器人令牌" in str(error.value) def test_role_permissions_gate_telegram_modules(app_modules): permissions = app_modules.load("wbb.services.bot_permissions") assert permissions.module_allowed("points", {"points.manage"}) is True assert permissions.module_allowed("points", {"giveaways.manage"}) is False assert permissions.module_allowed("admin", {"chat.members"}) is True assert permissions.module_allowed("admin", {"chat.invites"}) is True assert permissions.module_allowed("admin", set()) is False assert permissions.module_allowed("chat_watcher", set()) is True assert permissions.TELEGRAM_COMMAND_PERMISSIONS["ban"] == "chat.members" assert permissions.TELEGRAM_COMMAND_PERMISSIONS["pin"] == "chat.announcements" assert permissions.TELEGRAM_COMMAND_PERMISSIONS["invite"] == "chat.invites" assert ( permissions.api_permission("DELETE", "/api/admin/v1/chats/-100123") == "chat.profile" ) def test_supervisor_assigns_each_worker_an_isolated_database(app_modules, tmp_path): supervisor_module = app_modules.load("wbb.admin.supervisor") supervisor = supervisor_module.BotSupervisor(tmp_path / "bots.json", project_root=tmp_path) profile = { "bot_id": "bot.one/secondary", "bot_token": "123456:" + "A" * 30, "api_id": 12345, "api_hash": "a" * 32, "sudo_users_id": [], "log_group_id": 0, "gban_log_group_id": 0, "message_dump_chat": 0, "permissions": ["points.manage", "giveaways.manage"], "business_assistant": { "base_url": "https://models.example.com/v1", "api_key": "worker-only-key", "model": "support-model", }, } environment = supervisor._worker_environment(profile, 18088) assert environment["WBB_BOT_PROFILE_ID"] == "bot.one/secondary" assert environment["WBB_BOT_DATABASE"] == "wbb_bot_bot_one_secondary" assert environment["WBB_BOT_PERMISSIONS"] == "points.manage,giveaways.manage" assert environment["BUSINESS_ASSISTANT_OPENAI_API_KEY"] == "worker-only-key" assert environment["BUSINESS_ASSISTANT_OPENAI_MODEL"] == "support-model" async def test_bot_admin_api_crud_and_token_test_are_redacted(app_modules, monkeypatch): admin_api = app_modules.load("wbb.admin.api") application = admin_api.build_admin_application() await application["admin_api"].initialize() client = TestClient(TestServer(application), cookie_jar=CookieJar(unsafe=True)) await client.start_server() token = "123456:" + "C" * 30 try: login = await client.post( "/api/admin/v1/auth/login", json={"username": "admin", "password": "qwe0.123456"}, ) login_data = (await login.json())["data"] changed = await client.put( "/api/admin/v1/auth/password", headers={"X-CSRF-Token": login_data["csrf_token"]}, json={ "current_password": "qwe0.123456", "new_password": "changed-pass-123", }, ) csrf = (await changed.json())["data"]["csrf_token"] headers = {"X-CSRF-Token": csrf} telegram = await client.put( "/api/admin/v1/settings/telegram", headers=headers, json={"api_id": 12345, "api_hash": "d" * 32, "confirm": True}, ) assert telegram.status == 200 assert (await telegram.json())["data"]["api_hash_configured"] is True created_role = await client.post( "/api/admin/v1/roles", headers=headers, json={ "name": "积分专员", "description": "只负责积分", "permissions": ["points.manage"], "confirm": True, }, ) assert created_role.status == 201 role = (await created_role.json())["data"] created = await client.post( "/api/admin/v1/bots", headers=headers, json={ "label": "Test Bot", "bot_token": token, "enabled": False, "role_ids": [role["role_id"]], "confirm": True, }, ) assert created.status == 201 created_payload = await created.json() profile = created_payload["data"] assert profile["bot_token_configured"] is True assert profile["permissions"] == ["points.manage"] assert token not in json.dumps(created_payload) async def fake_test_bot_token(_token: str): assert _token == token return { "id": "123456", "username": "test_bot", "name": "Test Bot", "can_connect_to_business": True, } monkeypatch.setattr(admin_api, "test_bot_token", fake_test_bot_token) tested = await client.post( f"/api/admin/v1/bots/{profile['bot_id']}/test", headers=headers, json={"confirm": True}, ) assert tested.status == 200 assert (await tested.json())["data"]["username"] == "test_bot" assert (await tested.json())["data"]["can_connect_to_business"] is True listed = await client.get("/api/admin/v1/bots") listed_payload = await listed.json() assert listed_payload["data"]["items"][0]["identity"]["username"] == "test_bot" assert token not in json.dumps(listed_payload) deleted = await client.delete( f"/api/admin/v1/bots/{profile['bot_id']}", headers=headers, json={"confirm": True}, ) assert deleted.status == 200 deleted_role = await client.delete( f"/api/admin/v1/roles/{role['role_id']}", headers=headers, json={"confirm": True}, ) assert deleted_role.status == 200 finally: await client.close() async def test_bot_role_permission_is_enforced_by_worker_api(app_modules): admin_api = app_modules.load("wbb.admin.api") app_modules.wbb.BOT_PERMISSIONS = set() application = admin_api.build_admin_application() await application["admin_api"].initialize() client = TestClient(TestServer(application), cookie_jar=CookieJar(unsafe=True)) await client.start_server() try: login = await client.post( "/api/admin/v1/auth/login", json={"username": "admin", "password": "qwe0.123456"}, ) login_data = (await login.json())["data"] changed = await client.put( "/api/admin/v1/auth/password", headers={"X-CSRF-Token": login_data["csrf_token"]}, json={ "current_password": "qwe0.123456", "new_password": "changed-pass-123", }, ) assert changed.status == 200 denied = await client.get( "/api/admin/v1/points/accounts?chat_id=-100" ) assert denied.status == 403 payload = await denied.json() assert payload["error"]["code"] == "bot_role_permission_denied" assert payload["error"]["details"]["required_permission"] == "points.manage" app_modules.wbb.BOT_PERMISSIONS = {"points.manage"} allowed = await client.get( "/api/admin/v1/points/accounts?chat_id=-100" ) assert allowed.status == 200 finally: await client.close()