Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加静态路径准备和自动构建功能的单元测试

Browse Source
This commit is contained in:
晴猫
2026-03-15 10:06:59 +09:00
parent 9c577840cc
commit 34a8de56c3
2 changed files with 379 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
pytests/webui/test_app.py Normal file
View File

@@ -0,0 +1,122 @@
from unittest.mock import patch
from src.webui import app as webui_app
def test_ensure_static_path_ready_uses_existing_static_path(tmp_path) -> None:
static_path = tmp_path / "dist"
static_path.mkdir()
(static_path / "index.html").write_text("<html></html>", encoding="utf-8")
with (
patch.object(webui_app, "_resolve_static_path", return_value=static_path),
patch.object(webui_app, "_try_build_dashboard") as build_mock,
):
result = webui_app._ensure_static_path_ready()
assert result == static_path
build_mock.assert_not_called()
def test_ensure_static_path_ready_retries_after_auto_build(tmp_path) -> None:
static_path = tmp_path / "dist"
static_path.mkdir()
(static_path / "index.html").write_text("<html></html>", encoding="utf-8")
with (
patch.object(webui_app, "_resolve_static_path", side_effect=[None, static_path]),
patch.object(
webui_app,
"_try_build_dashboard",
return_value=webui_app.DashboardAutoRecoveryResult(succeeded=True),
) as build_mock,
):
result = webui_app._ensure_static_path_ready()
assert result == static_path
build_mock.assert_called_once_with()
def test_ensure_static_path_ready_logs_manual_hint_when_auto_build_fails() -> None:
with (
patch.object(webui_app, "_resolve_static_path", return_value=None),
patch.object(
webui_app,
"_try_build_dashboard",
return_value=webui_app.DashboardAutoRecoveryResult(
succeeded=False,
manual_recovery_command=webui_app._MANUAL_BUILD_COMMAND,
),
),
patch.object(webui_app.logger, "warning") as warning_mock,
):
result = webui_app._ensure_static_path_ready()
assert result is None
warning_mock.assert_any_call(webui_app.t("startup.webui_auto_recovery_failed"))
warning_mock.assert_any_call(
webui_app.t("startup.webui_manual_build_hint", command=webui_app._MANUAL_BUILD_COMMAND)
)
def test_setup_static_files_does_not_duplicate_warning_when_static_path_is_unavailable() -> None:
app = webui_app.FastAPI()
with (
patch.object(webui_app, "_ensure_static_path_ready", return_value=None),
patch.object(webui_app.logger, "warning") as warning_mock,
):
webui_app._setup_static_files(app)
warning_mock.assert_not_called()
def test_get_dashboard_build_command_defaults_to_npm(tmp_path) -> None:
(tmp_path / "package.json").write_text("{}", encoding="utf-8")
with patch.object(
webui_app.shutil,
"which",
side_effect=lambda tool_name: "/usr/bin/npm" if tool_name == "npm" else None,
):
command = webui_app._get_dashboard_build_command(tmp_path)
assert command == ["npm", "run", "build"]
def test_try_build_dashboard_installs_missing_dependencies_before_build(monkeypatch, tmp_path) -> None:
(tmp_path / "package.json").write_text("{}", encoding="utf-8")
run_results = [
webui_app.CompletedProcess(args=["npm", "install", "--no-package-lock"], returncode=0, stdout="", stderr=""),
webui_app.CompletedProcess(args=["npm", "run", "build"], returncode=0, stdout="", stderr=""),
]
monkeypatch.setattr(webui_app, "_get_dashboard_root", lambda: tmp_path)
monkeypatch.setattr(webui_app, "_should_auto_install_dashboard_dependencies", lambda dashboard_root: True)
with (
patch.object(webui_app, "_get_dashboard_build_command", return_value=["npm", "run", "build"]),
patch.object(webui_app, "run", side_effect=run_results) as run_mock,
):
result = webui_app._try_build_dashboard()
assert result.succeeded is True
assert run_mock.call_count == 2
install_call = run_mock.call_args_list[0]
build_call = run_mock.call_args_list[1]
assert install_call.args[0] == ["npm", "install", "--no-package-lock"]
assert install_call.kwargs["cwd"] == tmp_path
assert build_call.args[0] == ["npm", "run", "build"]
assert build_call.kwargs["cwd"] == tmp_path
def test_resolve_static_path_uses_dashboard_dist(monkeypatch, tmp_path) -> None:
dashboard_dist = tmp_path / "dashboard" / "dist"
dashboard_dist.mkdir(parents=True)
monkeypatch.setattr(webui_app, "_get_project_root", lambda: tmp_path)
with patch.object(webui_app, "import_module", side_effect=ImportError):
resolved_path = webui_app._resolve_static_path()
assert resolved_path == dashboard_dist

284
src/webui/app.py
View File

@@ -1,15 +1,44 @@
"""FastAPI 应用工厂 - 创建和配置 WebUI 应用实例"""
from dataclasses import dataclass
from importlib import import_module
from pathlib import Path
from subprocess import CompletedProcess, TimeoutExpired, run
import mimetypes
import shutil
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from src.common.i18n import t
from src.common.logger import get_logger
logger = get_logger("webui.app")
_DASHBOARD_BUILD_TIMEOUT_SECONDS = 300
_DASHBOARD_INSTALL_TIMEOUT_SECONDS = 600
_MANUAL_BUILD_COMMAND = "cd dashboard && npm install && npm run build"
_DASHBOARD_BUILD_COMMANDS = {
"bun": ["bun", "run", "build"],
"npm": ["npm", "run", "build"],
"pnpm": ["pnpm", "build"],
"yarn": ["yarn", "build"],
}
_DASHBOARD_INSTALL_COMMANDS = {
"bun": ["bun", "install"],
"npm": ["npm", "install", "--no-package-lock"],
"pnpm": ["pnpm", "install"],
"yarn": ["yarn", "install"],
}
@dataclass(frozen=True)
class DashboardAutoRecoveryResult:
succeeded: bool
manual_recovery_command: str | None = None
def _resolve_safe_static_file_path(static_path: Path, full_path: str) -> Path | None:
static_root = static_path.resolve()
@@ -18,12 +47,215 @@ def _resolve_safe_static_file_path(static_path: Path, full_path: str) -> Path |
candidate_path = (static_root / full_path).resolve()
candidate_path.relative_to(static_root)
except (OSError, RuntimeError, ValueError):
logger.warning(f"🚫 检测到疑似路径穿越请求: {full_path}")
logger.warning(t("startup.webui_path_traversal_detected", full_path=full_path))
return None
return candidate_path
def _get_project_root() -> Path:
return Path(__file__).resolve().parents[2]
def _get_dashboard_root() -> Path:
return _get_project_root() / "dashboard"
def _format_dashboard_shell_commands(*commands: list[str]) -> str:
formatted_commands = " && ".join(" ".join(command) for command in commands)
return f"cd dashboard && {formatted_commands}"
def _validate_static_path(static_path: Path | None) -> tuple[str, dict[str, object]] | None:
if static_path is None:
return "startup.webui_static_dir_missing", {}
if not static_path.exists():
return "startup.webui_static_dir_missing_with_path", {"static_path": static_path}
index_path = static_path / "index.html"
if not index_path.exists():
return "startup.webui_index_missing", {"index_path": index_path}
return None
def _summarize_command_output(command_result: CompletedProcess[str] | TimeoutExpired) -> str:
output_chunks: list[str] = []
stdout = command_result.stdout
stderr = command_result.stderr
if isinstance(stdout, str) and stdout.strip():
output_chunks.append(stdout.strip())
if isinstance(stderr, str) and stderr.strip():
output_chunks.append(stderr.strip())
if not output_chunks:
return ""
combined_output = "\n".join(output_chunks)
max_output_length = 2000
if len(combined_output) <= max_output_length:
return combined_output
return combined_output[-max_output_length:]
def _get_preferred_dashboard_package_manager(dashboard_root: Path) -> str:
if (dashboard_root / "pnpm-lock.yaml").exists():
return "pnpm"
if (dashboard_root / "yarn.lock").exists():
return "yarn"
if (dashboard_root / "bun.lock").exists() or (dashboard_root / "bun.lockb").exists():
return "bun"
return "npm"
def _get_dashboard_build_command(dashboard_root: Path) -> list[str] | None:
if not (dashboard_root / "package.json").exists():
return None
preferred_package_manager = _get_preferred_dashboard_package_manager(dashboard_root)
package_managers = [
preferred_package_manager,
*[package_manager for package_manager in _DASHBOARD_BUILD_COMMANDS if package_manager != preferred_package_manager],
]
for package_manager in package_managers:
if shutil.which(package_manager):
return _DASHBOARD_BUILD_COMMANDS[package_manager]
return None
def _get_dashboard_manual_recovery_command(dashboard_root: Path, build_command: list[str] | None = None) -> str:
package_manager = build_command[0] if build_command is not None else _get_preferred_dashboard_package_manager(dashboard_root)
install_command = _DASHBOARD_INSTALL_COMMANDS.get(package_manager)
selected_build_command = _DASHBOARD_BUILD_COMMANDS.get(package_manager)
if install_command is None or selected_build_command is None:
return _MANUAL_BUILD_COMMAND
return _format_dashboard_shell_commands(install_command, selected_build_command)
def _should_auto_install_dashboard_dependencies(dashboard_root: Path) -> bool:
return not (dashboard_root / "node_modules").exists()
def _try_build_dashboard() -> DashboardAutoRecoveryResult:
dashboard_root = _get_dashboard_root()
if not dashboard_root.exists():
logger.warning(t("startup.webui_dashboard_source_missing", dashboard_root=dashboard_root))
return DashboardAutoRecoveryResult(succeeded=False)
build_command = _get_dashboard_build_command(dashboard_root)
if build_command is None:
logger.warning(t("startup.webui_auto_build_tool_missing"))
manual_recovery_command = _get_dashboard_manual_recovery_command(dashboard_root)
return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command)
manual_recovery_command = _get_dashboard_manual_recovery_command(dashboard_root, build_command)
if _should_auto_install_dashboard_dependencies(dashboard_root):
install_command = _DASHBOARD_INSTALL_COMMANDS[build_command[0]]
logger.info(t("startup.webui_auto_install_started", command=" ".join(install_command)))
try:
install_result = run(
install_command,
capture_output=True,
check=False,
cwd=dashboard_root,
text=True,
timeout=_DASHBOARD_INSTALL_TIMEOUT_SECONDS,
)
except TimeoutExpired as exc:
logger.warning(
t("startup.webui_auto_install_timeout", timeout_seconds=_DASHBOARD_INSTALL_TIMEOUT_SECONDS),
)
install_output = _summarize_command_output(exc)
if install_output:
logger.warning(t("startup.webui_auto_install_failed_output", output=install_output))
return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command)
except OSError as exc:
logger.warning(t("startup.webui_auto_install_exec_failed", error=exc))
return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command)
if install_result.returncode != 0:
logger.warning(t("startup.webui_auto_install_failed", return_code=install_result.returncode))
install_output = _summarize_command_output(install_result)
if install_output:
logger.warning(t("startup.webui_auto_install_failed_output", output=install_output))
return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command)
logger.info(t("startup.webui_auto_install_succeeded"))
logger.info(t("startup.webui_auto_build_started", command=" ".join(build_command)))
try:
build_result = run(
build_command,
capture_output=True,
check=False,
cwd=dashboard_root,
text=True,
timeout=_DASHBOARD_BUILD_TIMEOUT_SECONDS,
)
except TimeoutExpired as exc:
logger.warning(
t("startup.webui_auto_build_timeout", timeout_seconds=_DASHBOARD_BUILD_TIMEOUT_SECONDS),
)
build_output = _summarize_command_output(exc)
if build_output:
logger.warning(t("startup.webui_auto_build_failed_output", output=build_output))
return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command)
except OSError as exc:
logger.warning(t("startup.webui_auto_build_exec_failed", error=exc))
return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command)
if build_result.returncode != 0:
logger.warning(t("startup.webui_auto_build_failed", return_code=build_result.returncode))
build_output = _summarize_command_output(build_result)
if build_output:
logger.warning(t("startup.webui_auto_build_failed_output", output=build_output))
return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command)
logger.info(t("startup.webui_auto_build_succeeded"))
return DashboardAutoRecoveryResult(succeeded=True, manual_recovery_command=manual_recovery_command)
def _ensure_static_path_ready() -> Path | None:
static_path = _resolve_static_path()
validation_error = _validate_static_path(static_path)
if validation_error is None:
return static_path
logger.info(t("startup.webui_static_assets_try_auto_build"))
auto_recovery_result = _try_build_dashboard()
if auto_recovery_result.succeeded:
static_path = _resolve_static_path()
validation_error = _validate_static_path(static_path)
if validation_error is None:
return static_path
logger.warning(t("startup.webui_auto_build_artifacts_invalid"))
error_key, error_kwargs = validation_error
logger.warning(t(error_key, **error_kwargs))
logger.warning(
t(
"startup.webui_manual_build_hint",
command=auto_recovery_result.manual_recovery_command or _MANUAL_BUILD_COMMAND,
)
)
return None
if auto_recovery_result.manual_recovery_command is not None:
logger.warning(t("startup.webui_auto_recovery_failed"))
logger.warning(t("startup.webui_manual_build_hint", command=auto_recovery_result.manual_recovery_command))
return None
def create_app(
host: str = "0.0.0.0",
port: int = 8001,
@@ -71,7 +303,7 @@ def _setup_cors(app: FastAPI, port: int):
],
expose_headers=["Content-Length", "Content-Type"],
)
logger.debug("✅ CORS 中间件已配置")
logger.debug(t("startup.webui_cors_configured"))
def _setup_anti_crawler(app: FastAPI):
@@ -83,15 +315,15 @@ def _setup_anti_crawler(app: FastAPI):
app.add_middleware(AntiCrawlerMiddleware, mode=anti_crawler_mode)
mode_descriptions = {
"false": "已禁用",
"strict": "严格模式",
"loose": "宽松模式",
"basic": "基础模式",
"false": t("startup.webui_anti_crawler_mode_disabled"),
"strict": t("startup.webui_anti_crawler_mode_strict"),
"loose": t("startup.webui_anti_crawler_mode_loose"),
"basic": t("startup.webui_anti_crawler_mode_basic"),
}
mode_desc = mode_descriptions.get(anti_crawler_mode, "基础模式")
logger.info(f"🛡️ 防爬虫中间件已配置: {mode_desc}")
mode_desc = mode_descriptions.get(anti_crawler_mode, t("startup.webui_anti_crawler_mode_basic"))
logger.info(t("startup.webui_anti_crawler_configured", mode_desc=mode_desc))
except Exception as e:
logger.error(f"❌ 配置防爬虫中间件失败: {e}", exc_info=True)
logger.error(t("startup.webui_anti_crawler_config_failed", error=e), exc_info=True)
def _setup_robots_txt(app: FastAPI):
@@ -102,9 +334,9 @@ def _setup_robots_txt(app: FastAPI):
async def robots_txt():
return create_robots_txt_response()
logger.debug("✅ robots.txt 路由已注册")
logger.debug(t("startup.webui_robots_route_registered"))
except Exception as e:
logger.error(f"❌ 注册robots.txt路由失败: {e}", exc_info=True)
logger.error(t("startup.webui_robots_route_register_failed", error=e), exc_info=True)
def _register_api_routes(app: FastAPI):
@@ -114,9 +346,9 @@ def _register_api_routes(app: FastAPI):
for router in get_all_routers():
app.include_router(router)
logger.info("✅ WebUI API 路由已注册")
logger.info(t("startup.webui_api_routes_registered"))
except Exception as e:
logger.error(f"❌ 注册 WebUI API 路由失败: {e}", exc_info=True)
logger.error(t("startup.webui_api_routes_register_failed", error=e), exc_info=True)
def _setup_static_files(app: FastAPI):
@@ -126,20 +358,18 @@ def _setup_static_files(app: FastAPI):
mimetypes.add_type("text/css", ".css")
mimetypes.add_type("application/json", ".json")
static_path = _resolve_static_path()
static_path = _ensure_static_path_ready()
if static_path is None:
logger.warning("❌ WebUI 静态文件目录不存在")
logger.warning("💡 请先构建前端: cd dashboard && npm run build")
return
if not static_path.exists():
logger.warning(f"❌ WebUI 静态文件目录不存在: {static_path}")
logger.warning("💡 请先构建前端: cd dashboard && npm run build")
logger.warning(t("startup.webui_static_dir_missing_with_path", static_path=static_path))
logger.warning(t("startup.webui_manual_build_hint", command=_MANUAL_BUILD_COMMAND))
return
if not (static_path / "index.html").exists():
logger.warning(f"❌ 未找到 index.html: {static_path / 'index.html'}")
logger.warning("💡 请确认前端已正确构建")
logger.warning(t("startup.webui_index_missing", index_path=static_path / "index.html"))
logger.warning(t("startup.webui_manual_build_hint", command=_MANUAL_BUILD_COMMAND))
return
@app.get("/{full_path:path}", include_in_schema=False)
@@ -151,7 +381,7 @@ def _setup_static_files(app: FastAPI):
file_path = _resolve_safe_static_file_path(static_path, full_path)
if file_path is None:
raise HTTPException(status_code=404, detail="Not Found")
raise HTTPException(status_code=404, detail=t("core.not_found"))
if file_path.exists() and file_path.is_file():
media_type = mimetypes.guess_type(str(file_path))[0]
@@ -164,7 +394,7 @@ def _setup_static_files(app: FastAPI):
response.headers["X-Robots-Tag"] = "noindex, nofollow, noarchive"
return response
logger.info(f"✅ WebUI 静态文件服务已配置: {static_path}")
logger.info(t("startup.webui_static_files_configured", static_path=static_path))
def _resolve_static_path() -> Path | None:
@@ -178,8 +408,8 @@ def _resolve_static_path() -> Path | None:
except Exception:
pass
base_dir = Path(__file__).parent.parent.parent
static_path = base_dir / "webui" / "dist"
base_dir = _get_project_root()
static_path = base_dir / "dashboard" / "dist"
if static_path.exists():
return static_path
return None
@@ -192,7 +422,7 @@ def show_access_token():
token_manager = get_token_manager()
current_token = token_manager.get_token()
logger.info(f"🔑 WebUI Access Token: {current_token}")
logger.info("💡 请使用此 Token 登录 WebUI")
logger.info(t("startup.webui_access_token", token=current_token))
logger.info(t("startup.webui_access_token_login_hint"))
except Exception as e:
logger.error(f"❌ 获取 Access Token 失败: {e}")
logger.error(t("startup.webui_access_token_failed", error=e))