AstrBot/tests/test_dashboard.py

import pytest
import os
from quart import Quart
from astrbot.dashboard.server import AstrBotDashboard
from astrbot.core.db.sqlite import SQLiteDatabase
from astrbot.core.core_lifecycle import AstrBotCoreLifecycle
from astrbot.core import LogBroker
from astrbot.core.star.star_handler import star_handlers_registry
from astrbot.core.star.star import star_registry


@pytest.fixture(scope="module")
def core_lifecycle_td():
    db = SQLiteDatabase("data/data_v3.db")
    log_broker = LogBroker()
    core_lifecycle_td = AstrBotCoreLifecycle(log_broker, db)
    return core_lifecycle_td


@pytest.fixture(scope="module")
def app(core_lifecycle_td):
    db = SQLiteDatabase("data/data_v3.db")
    server = AstrBotDashboard(core_lifecycle_td, db)
    return server.app


@pytest.fixture(scope="module")
def header():
    return {}


@pytest.mark.asyncio
async def test_init_core_lifecycle_td(core_lifecycle_td):
    await core_lifecycle_td.initialize()
    assert core_lifecycle_td is not None


@pytest.mark.asyncio
async def test_auth_login(
    app: Quart, core_lifecycle_td: AstrBotCoreLifecycle, header: dict
):
    test_client = app.test_client()
    response = await test_client.post(
        "/api/auth/login", json={"username": "wrong", "password": "password"}
    )
    data = await response.get_json()
    assert data["status"] == "error"

    response = await test_client.post(
        "/api/auth/login",
        json={
            "username": core_lifecycle_td.astrbot_config["dashboard"]["username"],
            "password": core_lifecycle_td.astrbot_config["dashboard"]["password"],
        },
    )
    data = await response.get_json()
    assert data["status"] == "ok" and "token" in data["data"]
    header["Authorization"] = f"Bearer {data['data']['token']}"


@pytest.mark.asyncio
async def test_get_stat(app: Quart, header: dict):
    test_client = app.test_client()
    response = await test_client.get("/api/stat/get")
    assert response.status_code == 401
    response = await test_client.get("/api/stat/get", headers=header)
    assert response.status_code == 200
    data = await response.get_json()
    assert data["status"] == "ok" and "platform" in data["data"]


@pytest.mark.asyncio
async def test_plugins(app: Quart, header: dict):
    test_client = app.test_client()
    # 已经安装的插件
    response = await test_client.get("/api/plugin/get", headers=header)
    assert response.status_code == 200
    data = await response.get_json()
    assert data["status"] == "ok"

    # 插件市场
    response = await test_client.get("/api/plugin/market_list", headers=header)
    assert response.status_code == 200
    data = await response.get_json()
    assert data["status"] == "ok"

    # 插件安装
    response = await test_client.post(
        "/api/plugin/install",
        json={"url": "https://github.com/Soulter/astrbot_plugin_essential"},
        headers=header,
    )
    assert response.status_code == 200
    data = await response.get_json()
    assert data["status"] == "ok"
    exists = False
    for md in star_registry:
        if md.name == "astrbot_plugin_essential":
            exists = True
            break
    assert exists is True, "插件 astrbot_plugin_essential 未成功载入"

    # 插件更新
    response = await test_client.post(
        "/api/plugin/update", json={"name": "astrbot_plugin_essential"}, headers=header
    )
    assert response.status_code == 200
    data = await response.get_json()
    assert data["status"] == "ok"

    # 插件卸载
    response = await test_client.post(
        "/api/plugin/uninstall",
        json={"name": "astrbot_plugin_essential"},
        headers=header,
    )
    assert response.status_code == 200
    data = await response.get_json()
    assert data["status"] == "ok"
    exists = False
    for md in star_registry:
        if md.name == "astrbot_plugin_essential":
            exists = True
            break
    assert exists is False, "插件 astrbot_plugin_essential 未成功卸载"
    exists = False
    for md in star_handlers_registry:
        if "astrbot_plugin_essential" in md.handler_module_path:
            exists = True
            break
    assert exists is False, "插件 astrbot_plugin_essential 未成功卸载"


@pytest.mark.asyncio
async def test_check_update(app: Quart, header: dict):
    test_client = app.test_client()
    response = await test_client.get("/api/update/check", headers=header)
    assert response.status_code == 200
    data = await response.get_json()
    assert data["status"] == "success"


@pytest.mark.asyncio
async def test_do_update(
    app: Quart, header: dict, core_lifecycle_td: AstrBotCoreLifecycle
):
    global VERSION
    test_client = app.test_client()
    os.makedirs("data/astrbot_release", exist_ok=True)
    core_lifecycle_td.astrbot_updator.MAIN_PATH = "data/astrbot_release"
    VERSION = "114.514.1919810"
    response = await test_client.post(
        "/api/update/do", headers=header, json={"version": "latest"}
    )
    assert response.status_code == 200
    data = await response.get_json()
    assert data["status"] == "error"  # 已经是最新版本

    response = await test_client.post(
        "/api/update/do", headers=header, json={"version": "v3.4.0", "reboot": False}
    )
    assert response.status_code == 200
    data = await response.get_json()
    assert data["status"] == "ok"
    assert os.path.exists("data/astrbot_release/astrbot")