mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-16 01:40:15 +08:00
0d8e8682db
* refactor: migrate to fastapi * structure refactor * fix: pyright fix * refactor: improve error handling and public messages in plugin services * feat(api): refactor API client integration and enhance request handling - Updated API client configuration to use a dedicated HTTP client. - Introduced utility functions for generating options, queries, and form data for API requests. - Refactored multiple API methods to utilize the new utility functions for improved consistency and readability. - Renamed types for clarity and updated import statements accordingly. feat(docs): add script to update OpenAPI JSON from YAML spec - Created a Python script to convert OpenAPI YAML specification to JSON format. - The script supports customizable input and output paths. - Ensured the script handles directory creation for output paths and validates the YAML structure. * fix * feat(auth): implement rate limiting for v1 login endpoint and enhance request handling * Refactor dashboard API routers to use legacy_router for backward compatibility - Changed all instances of dashboard_router to legacy_router across multiple API modules including platform, plugins, providers, sessions, skills, stats, subagents, t2i, tools, updates, and asgi_runtime. - Updated route definitions to ensure existing endpoints remain functional under the new router structure. - Introduced support for Quart request context in asgi_runtime to enhance compatibility with existing Quart-based plugins. - Added a test case to validate the functionality of the new Quart request context handling in plugin extensions. * chore: remove cli test * fix: update dashboard tests for fastapi migration * chore: satisfy ruff checks * fix: update openapi api key scopes * fix: sync config scope chip selection * fix: restore quart dependency * docs: clarify quart plugin api compatibility * docs: update openapi scope documentation * fix: use singular skill openapi scope * fix: hide update service exception details * fix: address fastapi review comments * fix: address dashboard review findings * docs: revert unrelated package deployment changes * docs: update agent api generation guidance * feat: add plugin page web api helpers * docs: add plugin page bridge demo * fix: type plugin upload files * fix: stabilize plugin page uploads * fix: type plugin web request proxy * docs: remove plugin page docs example * fix: authenticate plugin page SSE bridge
260 lines
9.5 KiB
Python
260 lines
9.5 KiB
Python
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 将项目根目录添加到 sys.path
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from astrbot.core.utils.io import should_use_bundled_dashboard_dist
|
|
from main import check_dashboard_files, check_env
|
|
|
|
|
|
class _version_info:
|
|
def __init__(self, major, minor):
|
|
self.major = major
|
|
self.minor = minor
|
|
|
|
def __eq__(self, other):
|
|
if isinstance(other, tuple):
|
|
return (self.major, self.minor) == other[:2]
|
|
return (self.major, self.minor) == (other.major, other.minor)
|
|
|
|
def __ge__(self, other):
|
|
if isinstance(other, tuple):
|
|
return (self.major, self.minor) >= other[:2]
|
|
return (self.major, self.minor) >= (other.major, other.minor)
|
|
|
|
def __le__(self, other):
|
|
if isinstance(other, tuple):
|
|
return (self.major, self.minor) <= other[:2]
|
|
return (self.major, self.minor) <= (other.major, other.minor)
|
|
|
|
def __gt__(self, other):
|
|
if isinstance(other, tuple):
|
|
return (self.major, self.minor) > other[:2]
|
|
return (self.major, self.minor) > (other.major, other.minor)
|
|
|
|
def __lt__(self, other):
|
|
if isinstance(other, tuple):
|
|
return (self.major, self.minor) < other[:2]
|
|
return (self.major, self.minor) < (other.major, other.minor)
|
|
|
|
|
|
def test_check_env(monkeypatch):
|
|
version_info_correct = _version_info(3, 10)
|
|
version_info_wrong = _version_info(3, 9)
|
|
monkeypatch.setattr(sys, "version_info", version_info_correct)
|
|
with mock.patch("os.makedirs") as mock_makedirs:
|
|
check_env()
|
|
# check_env uses get_astrbot_*_path() which returns absolute paths,
|
|
# so just verify makedirs was called the expected number of times
|
|
assert mock_makedirs.call_count >= 4
|
|
# Verify all calls used exist_ok=True
|
|
for call_args in mock_makedirs.call_args_list:
|
|
assert call_args[1].get("exist_ok") is True
|
|
|
|
monkeypatch.setattr(sys, "version_info", version_info_wrong)
|
|
with pytest.raises(SystemExit):
|
|
check_env()
|
|
|
|
|
|
def test_check_env_appends_user_site_packages_after_runtime_paths(monkeypatch):
|
|
astrbot_root = "/tmp/astrbot-root"
|
|
site_packages_path = "/tmp/astrbot-site-packages"
|
|
original_sys_path = list(sys.path)
|
|
|
|
monkeypatch.setattr(sys, "version_info", _version_info(3, 12))
|
|
monkeypatch.setattr("main.get_astrbot_root", lambda: astrbot_root)
|
|
monkeypatch.setattr(
|
|
"main.get_astrbot_site_packages_path", lambda: site_packages_path
|
|
)
|
|
monkeypatch.setattr("main.get_astrbot_config_path", lambda: "/tmp/config")
|
|
monkeypatch.setattr("main.get_astrbot_plugin_path", lambda: "/tmp/plugins")
|
|
monkeypatch.setattr("main.get_astrbot_temp_path", lambda: "/tmp/temp")
|
|
monkeypatch.setattr("main.get_astrbot_knowledge_base_path", lambda: "/tmp/kb")
|
|
monkeypatch.setattr(sys, "path", ["/runtime/lib", *original_sys_path])
|
|
|
|
with mock.patch("os.makedirs"):
|
|
check_env()
|
|
|
|
assert sys.path[0] == astrbot_root
|
|
assert sys.path[-1] == site_packages_path
|
|
assert sys.path.index(site_packages_path) > sys.path.index("/runtime/lib")
|
|
|
|
|
|
def test_check_env_does_not_append_duplicate_user_site_packages(monkeypatch):
|
|
astrbot_root = "/tmp/astrbot-root"
|
|
site_packages_path = "/tmp/astrbot-site-packages"
|
|
original_sys_path = list(sys.path)
|
|
|
|
monkeypatch.setattr(sys, "version_info", _version_info(3, 12))
|
|
monkeypatch.setattr("main.get_astrbot_root", lambda: astrbot_root)
|
|
monkeypatch.setattr(
|
|
"main.get_astrbot_site_packages_path", lambda: site_packages_path
|
|
)
|
|
monkeypatch.setattr("main.get_astrbot_config_path", lambda: "/tmp/config")
|
|
monkeypatch.setattr("main.get_astrbot_plugin_path", lambda: "/tmp/plugins")
|
|
monkeypatch.setattr("main.get_astrbot_temp_path", lambda: "/tmp/temp")
|
|
monkeypatch.setattr("main.get_astrbot_knowledge_base_path", lambda: "/tmp/kb")
|
|
monkeypatch.setattr(
|
|
sys, "path", [astrbot_root, *original_sys_path, site_packages_path]
|
|
)
|
|
|
|
with mock.patch("os.makedirs"):
|
|
check_env()
|
|
|
|
assert sys.path.count(site_packages_path) == 1
|
|
|
|
|
|
def test_version_info_comparisons():
|
|
"""Test _version_info comparison operators with tuples and other instances."""
|
|
v3_10 = _version_info(3, 10)
|
|
v3_9 = _version_info(3, 9)
|
|
v3_11 = _version_info(3, 11)
|
|
|
|
# Test __eq__ with tuples
|
|
assert v3_10 == (3, 10)
|
|
assert v3_10 != (3, 9)
|
|
assert v3_9 == (3, 9)
|
|
|
|
# Test __ge__ with tuples
|
|
assert v3_10 >= (3, 10)
|
|
assert v3_10 >= (3, 9)
|
|
assert not (v3_9 >= (3, 10))
|
|
assert v3_11 >= (3, 10)
|
|
|
|
# Test __eq__ with other _version_info instances
|
|
assert v3_10 == _version_info(3, 10)
|
|
assert v3_10 != v3_9
|
|
assert v3_10 == v3_10 # Same instance
|
|
|
|
assert v3_10 != v3_11
|
|
|
|
# Test __ge__ with other _version_info instances
|
|
assert v3_10 >= v3_10
|
|
assert v3_10 >= v3_9
|
|
assert not (v3_9 >= v3_10)
|
|
assert v3_11 >= v3_10
|
|
|
|
assert v3_11 >= v3_11 # Same instance
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_dashboard_files_not_exists(monkeypatch):
|
|
"""Tests dashboard download when files do not exist."""
|
|
monkeypatch.setattr(os.path, "exists", lambda x: False)
|
|
|
|
with mock.patch("main.download_dashboard") as mock_download:
|
|
await check_dashboard_files()
|
|
mock_download.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_dashboard_files_exists_and_version_match(monkeypatch):
|
|
"""Tests that dashboard is not downloaded when it exists and version matches."""
|
|
# Mock os.path.exists to return True
|
|
monkeypatch.setattr(os.path, "exists", lambda x: True)
|
|
|
|
# Mock get_dashboard_version to return the current version
|
|
with mock.patch("main.get_dashboard_version") as mock_get_version:
|
|
# We need to import VERSION from main's context
|
|
from main import VERSION
|
|
|
|
mock_get_version.return_value = f"v{VERSION}"
|
|
|
|
with mock.patch("main.download_dashboard") as mock_download:
|
|
await check_dashboard_files()
|
|
# Assert that download_dashboard was NOT called
|
|
mock_download.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_dashboard_files_exists_but_version_mismatch(monkeypatch):
|
|
"""Tests that a warning is logged when dashboard version mismatches."""
|
|
monkeypatch.setattr(os.path, "exists", lambda x: True)
|
|
|
|
with mock.patch(
|
|
"main.get_dashboard_version", mock.AsyncMock(return_value="v0.0.1")
|
|
):
|
|
with mock.patch("main.logger.warning") as mock_logger_warning:
|
|
await check_dashboard_files()
|
|
mock_logger_warning.assert_called_once()
|
|
call_args, _ = mock_logger_warning.call_args
|
|
assert "WebUI version mismatch" in call_args[0]
|
|
|
|
|
|
def test_should_use_bundled_dashboard_dist_when_data_dist_is_stale(tmp_path):
|
|
user_dist = tmp_path / "user-dist"
|
|
bundled_dist = tmp_path / "bundled-dist"
|
|
(user_dist / "assets").mkdir(parents=True)
|
|
(bundled_dist / "assets").mkdir(parents=True)
|
|
(user_dist / "assets" / "version").write_text("v4.24.2", encoding="utf-8")
|
|
(bundled_dist / "assets" / "version").write_text("v4.24.4", encoding="utf-8")
|
|
|
|
with mock.patch(
|
|
"astrbot.core.utils.io.get_bundled_dashboard_dist_path",
|
|
return_value=bundled_dist,
|
|
):
|
|
assert should_use_bundled_dashboard_dist(user_dist, "v4.24.4") is True
|
|
|
|
|
|
def test_should_keep_data_dist_when_version_file_is_malformed(tmp_path):
|
|
user_dist = tmp_path / "user-dist"
|
|
bundled_dist = tmp_path / "bundled-dist"
|
|
(user_dist / "assets").mkdir(parents=True)
|
|
(bundled_dist / "assets").mkdir(parents=True)
|
|
(user_dist / "assets" / "version").write_text("not-a-version", encoding="utf-8")
|
|
|
|
with mock.patch(
|
|
"astrbot.core.utils.io.get_bundled_dashboard_dist_path",
|
|
return_value=bundled_dist,
|
|
):
|
|
assert should_use_bundled_dashboard_dist(user_dist, "4.24.4") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_dashboard_files_uses_bundled_dist_when_data_dist_is_stale(
|
|
tmp_path,
|
|
):
|
|
"""Tests that a stale data/dist does not override bundled dashboard assets."""
|
|
data_dir = tmp_path / "data"
|
|
data_dist = data_dir / "dist"
|
|
bundled_dist = tmp_path / "bundled-dist"
|
|
data_dist.mkdir(parents=True)
|
|
bundled_dist.mkdir()
|
|
|
|
with mock.patch("main.get_astrbot_data_path", return_value=str(data_dir)):
|
|
with mock.patch(
|
|
"main.get_dashboard_version", mock.AsyncMock(return_value="v0.0.1")
|
|
):
|
|
with mock.patch(
|
|
"main.should_use_bundled_dashboard_dist", return_value=True
|
|
):
|
|
with mock.patch(
|
|
"main.get_bundled_dashboard_dist_path",
|
|
return_value=Path(bundled_dist),
|
|
):
|
|
with mock.patch("main.download_dashboard") as mock_download:
|
|
result = await check_dashboard_files()
|
|
|
|
assert result == str(bundled_dist)
|
|
mock_download.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_dashboard_files_with_webui_dir_arg(monkeypatch):
|
|
"""Tests that providing a valid webui_dir skips all checks."""
|
|
valid_dir = "/tmp/my-custom-webui"
|
|
monkeypatch.setattr(os.path, "exists", lambda path: path == valid_dir)
|
|
|
|
with mock.patch("main.download_dashboard") as mock_download:
|
|
with mock.patch("main.get_dashboard_version") as mock_get_version:
|
|
result = await check_dashboard_files(webui_dir=valid_dir)
|
|
assert result == valid_dir
|
|
mock_download.assert_not_called()
|
|
mock_get_version.assert_not_called()
|