TelegramBotAdmin/tests/test_bot_config.py

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,
        },
    )
    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"]
    serialized = json.dumps(status)
    assert first_token not in serialized
    assert second_token not in serialized
    assert "a" * 32 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"


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"],
    }

    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"


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"}

        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"

        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()