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

    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"


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 = await client.post(
            "/api/admin/v1/bots",
            headers=headers,
            json={
                "label": "Test Bot",
                "bot_token": token,
                "enabled": False,
                "confirm": True,
            },
        )
        assert created.status == 201
        created_payload = await created.json()
        profile = created_payload["data"]
        assert profile["bot_token_configured"] is True
        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
    finally:
        await client.close()