nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-07-05 11:28:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(deps): add uv-backed package installer (#5987)

Browse Source 
* feat(deps): add uv-backed package installer

* feat(deps): support package cache root
This commit is contained in:
InfinityPacer
2026-06-23 13:36:15 +08:00
committed by GitHub
parent 126279c63b
commit 0c53fb86fd
21 changed files with 1654 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -16,7 +16,7 @@ app/helper/*.pyd
app/helper/*.bin
app/plugins/**
!app/plugins/__init__.py
config/cookies/**
config/cookies/
config/app.env
config/user.db*
config/sites/**
@@ -25,6 +25,7 @@ config/logs/
config/plugins/
config/temp/
config/cache/
config/.cache/
.runtime/
public/
.moviepilot.env

13
app/cli.py
View File

@@ -325,11 +325,16 @@ def _best_effort_auto_update() -> None:
]
update_env = os.environ.copy()
package_cache_root = Path(update_env.get("PACKAGE_CACHE_ROOT", "").strip() or settings.PACKAGE_CACHE_PATH)
update_env.setdefault("PACKAGE_CACHE_ROOT", str(package_cache_root))
update_env.setdefault("PIP_CACHE_DIR", str(package_cache_root / "pip"))
update_env.setdefault("UV_CACHE_DIR", str(package_cache_root / "uv"))
if settings.PIP_PROXY:
update_env["PIP_PROXY"] = settings.PIP_PROXY
if settings.PROXY_HOST:
update_env.setdefault("http_proxy", settings.PROXY_HOST)
update_env.setdefault("https_proxy", settings.PROXY_HOST)
update_env.setdefault("HTTP_PROXY", settings.PROXY_HOST)
update_env.setdefault("HTTPS_PROXY", settings.PROXY_HOST)
update_env["PROXY_HOST"] = settings.PROXY_HOST
for key in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"):
update_env[key] = settings.PROXY_HOST
if settings.GITHUB_TOKEN:
update_env.setdefault("GITHUB_TOKEN", settings.GITHUB_TOKEN)

10
app/core/config.py
View File

@@ -170,6 +170,10 @@ class ConfigModel(BaseModel):
GLOBAL_IMAGE_CACHE_DAYS: int = 7
# 临时文件保留天数
TEMP_FILE_DAYS: int = 3
# pip/uv 包下载缓存保留天数
PACKAGE_CACHE_DAYS: int = 90
# pip/uv 包下载缓存根目录，留空时使用配置目录下的 .cache
PACKAGE_CACHE_ROOT: Optional[str] = None
# 元数据识别缓存过期时间（小时），0为自动
META_CACHE_EXPIRE: int = 0
@@ -942,6 +946,12 @@ class Settings(BaseSettings, ConfigModel, LogConfigModel):
def CACHE_PATH(self):
return self.CONFIG_PATH / "cache"
@property
def PACKAGE_CACHE_PATH(self):
if self.PACKAGE_CACHE_ROOT and self.PACKAGE_CACHE_ROOT.strip():
return Path(self.PACKAGE_CACHE_ROOT).expanduser()
return self.CONFIG_PATH / ".cache"
@property
def ROOT_PATH(self):
return Path(__file__).parents[2]

169
app/helper/package_installer.py Normal file
View File

@@ -0,0 +1,169 @@
from __future__ import annotations
import os
import shutil
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal
from urllib.parse import urlsplit, urlunsplit
PackageBackend = Literal["uv", "pip"]
@dataclass(frozen=True)
class PackageInstallRequest:
"""
Python 包安装请求，集中描述依赖文件、工具缓存、代理和本地 wheels 候选源。
"""
requirements_file: Path
python_bin: Path
find_links_dirs: list[Path] = field(default_factory=list)
constraints_file: Path | None = None
config_dir: Path = Path("/config")
package_cache_root: Path | None = None
pip_index_url: str | None = None
proxy_url: str | None = None
purpose: str = "plugin"
@dataclass(frozen=True)
class PackageInstallStrategy:
"""
单次安装尝试的完整执行信息，命令和日志展示命令分离以避免泄露凭据。
"""
strategy_name: str
backend: PackageBackend
command: list[str]
env: dict[str, str]
safe_log_command: list[str]
def redact_url(value: str) -> str:
"""
脱敏 URL 中的 userinfo，保留 scheme、host、path、query 便于定位镜像源。
"""
parsed = urlsplit(value)
if "@" not in parsed.netloc:
return value
host = parsed.netloc.rsplit("@", 1)[-1]
return urlunsplit((parsed.scheme, host, parsed.path, parsed.query, parsed.fragment))
def redact_command(command: list[str]) -> list[str]:
"""
脱敏命令参数中的 URL 凭据，用于日志展示。
"""
return [redact_url(item) if "://" in item else item for item in command]
def build_package_install_env(request: PackageInstallRequest, include_moviepilot_proxy: bool = True) -> dict[str, str]:
"""
构造 pip/uv 安装子进程环境，默认把包下载缓存放到持久化配置目录。
"""
env = os.environ.copy()
config_dir = Path(request.config_dir)
if request.package_cache_root:
package_cache_root = Path(request.package_cache_root)
env["PACKAGE_CACHE_ROOT"] = str(package_cache_root)
else:
package_cache_root = Path(env.get("PACKAGE_CACHE_ROOT") or config_dir / ".cache")
env.setdefault("PACKAGE_CACHE_ROOT", str(package_cache_root))
env.setdefault("PIP_CACHE_DIR", str(package_cache_root / "pip"))
env.setdefault("UV_CACHE_DIR", str(package_cache_root / "uv"))
proxy = (request.proxy_url or "").strip()
if proxy and include_moviepilot_proxy:
for key in ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"):
env[key] = proxy
return env
def _find_uv(python_bin: Path) -> Path | None:
"""
优先使用解释器同目录 uv，保证虚拟环境内 wrapper 与真实安装环境一致。
"""
uv_name = "uv.exe" if os.name == "nt" else "uv"
sibling = python_bin.with_name(uv_name)
if sibling.exists():
return sibling
found = shutil.which("uv")
return Path(found) if found else None
def _base_install_args(request: PackageInstallRequest) -> list[str]:
args: list[str] = []
for directory in request.find_links_dirs:
args.extend(["--find-links", str(directory)])
if request.constraints_file:
args.extend(["-c", str(request.constraints_file)])
args.extend(["-r", str(request.requirements_file)])
return args
def _network_variants(request: PackageInstallRequest) -> list[tuple[str, bool, bool]]:
has_index = bool((request.pip_index_url or "").strip())
has_proxy = bool((request.proxy_url or "").strip())
variants: list[tuple[str, bool, bool]] = []
if has_index and has_proxy:
variants.append(("镜像+代理", True, True))
if has_index:
variants.append(("镜像", True, False))
if has_proxy:
variants.append(("代理", False, True))
variants.append(("直连", False, False))
return variants
def _build_uv_command(uv_bin: Path, request: PackageInstallRequest, use_index: bool) -> list[str]:
command = [str(uv_bin), "pip", "install", "--python", str(request.python_bin)]
if use_index and request.pip_index_url:
command.extend(["--default-index", request.pip_index_url])
command.extend(_base_install_args(request))
return command
def _build_pip_command(request: PackageInstallRequest, use_index: bool) -> list[str]:
command = [str(request.python_bin), "-m", "pip", "install"]
if use_index and request.pip_index_url:
command.extend(["-i", request.pip_index_url])
command.extend(_base_install_args(request))
return command
def build_package_install_strategies(request: PackageInstallRequest) -> list[PackageInstallStrategy]:
"""
按 uv 优先、pip 兜底顺序构造网络降级策略。
"""
strategies: list[PackageInstallStrategy] = []
variants = _network_variants(request)
uv_bin = _find_uv(Path(request.python_bin))
if uv_bin:
for variant_name, use_index, use_proxy in variants:
command = _build_uv_command(uv_bin, request, use_index)
env = build_package_install_env(request, include_moviepilot_proxy=use_proxy)
strategies.append(
PackageInstallStrategy(
strategy_name=f"uv:{variant_name}",
backend="uv",
command=command,
env=env,
safe_log_command=redact_command(command),
)
)
for variant_name, use_index, use_proxy in variants:
command = _build_pip_command(request, use_index)
env = build_package_install_env(request, include_moviepilot_proxy=use_proxy)
strategies.append(
PackageInstallStrategy(
strategy_name=f"pip:{variant_name}",
backend="pip",
command=command,
env=env,
safe_log_command=redact_command(command),
)
)
return strategies

111
app/helper/plugin.py
View File

@@ -29,6 +29,7 @@ from requests import Response
from app.core.cache import cached, is_fresh
from app.core.config import settings
from app.db.systemconfig_oper import SystemConfigOper
from app.helper.package_installer import PackageInstallRequest, build_package_install_strategies
from app.log import logger
from app.schemas.types import SystemConfigKey
from app.utils.http import RequestUtils, AsyncRequestUtils
@@ -1013,19 +1014,6 @@ class PluginHelper(metaclass=WeakSingleton):
# 去重并保持稳定顺序，避免重复传递相同目录
return list(dict.fromkeys(wheels_dirs))
@staticmethod
def __build_pip_install_strategies(base_cmd: List[str]) -> List[Tuple[str, List[str]]]:
"""
为 pip 命令构建统一的网络降级策略，避免不同安装路径各自拼接参数。
"""
strategies = []
if settings.PIP_PROXY:
strategies.append(("镜像站", base_cmd + ["-i", settings.PIP_PROXY]))
if settings.PROXY_HOST:
strategies.append(("代理", base_cmd + ["--proxy", settings.PROXY_HOST]))
strategies.append(("直连", base_cmd))
return strategies
@staticmethod
def __build_runtime_pip_command(*args: str) -> List[str]:
"""
@@ -1388,6 +1376,45 @@ class PluginHelper(metaclass=WeakSingleton):
importlib.reload(site)
importlib.invalidate_caches()
@classmethod
def __build_package_install_request(
cls,
requirements_file: Path,
find_links_dirs: Optional[List[Path]] = None,
constraints_file: Optional[Path] = None,
purpose: str = "plugin",
) -> PackageInstallRequest:
"""
将 MoviePilot 运行配置转换为 pip/uv 安装请求，统一缓存、镜像和代理语义。
"""
return PackageInstallRequest(
requirements_file=requirements_file,
python_bin=Path(sys.executable),
find_links_dirs=find_links_dirs or [],
constraints_file=constraints_file,
config_dir=settings.CONFIG_PATH,
package_cache_root=settings.PACKAGE_CACHE_PATH,
pip_index_url=settings.PIP_PROXY or None,
proxy_url=settings.PROXY_HOST or None,
purpose=purpose,
)
@classmethod
def __repair_if_runtime_broken(cls, snapshot_file: Optional[Path] = None) -> Tuple[bool, str]:
"""
安装失败后检查主运行环境；若已异常，先恢复主程序依赖再继续向上返回安装失败。
"""
health_ok, health_message = cls.__run_runtime_healthcheck()
if health_ok:
return True, ""
repair_ok, repair_message = cls.__repair_main_runtime_dependencies(snapshot_file)
if not repair_ok:
return False, f"插件依赖安装失败后主运行环境异常，且恢复失败：{health_message}; {repair_message}"
restored, restored_message = cls.__run_runtime_healthcheck()
if not restored:
return False, f"插件依赖安装失败后主运行环境异常，恢复后仍异常：{restored_message}"
return True, "主运行环境已恢复"
@classmethod
def __run_runtime_healthcheck(cls) -> Tuple[bool, str]:
"""
@@ -1420,15 +1447,19 @@ class PluginHelper(metaclass=WeakSingleton):
return False, f"恢复依赖文件不存在：{repair_target}"
last_error = ""
base_cmd = [sys.executable, "-m", "pip", "install", "-r", str(repair_target)]
for strategy_name, pip_command in cls.__build_pip_install_strategies(base_cmd):
logger.warning(f"[PIP] 运行环境异常，尝试使用策略：{strategy_name} 恢复{repair_desc}")
success, message = SystemUtils.execute_with_subprocess(pip_command)
request = cls.__build_package_install_request(repair_target, purpose="runtime-repair")
for strategy in build_package_install_strategies(request):
logger.warning(f"[PIP] 运行环境异常，尝试使用策略：{strategy.strategy_name} 恢复{repair_desc}")
success, message = SystemUtils.execute_with_subprocess(
strategy.command,
env=strategy.env,
safe_command=strategy.safe_log_command,
)
if success:
cls.__refresh_import_system()
return True, message
last_error = message
logger.error(f"[PIP] 使用策略：{strategy_name} 恢复{repair_desc}失败：{message}")
logger.error(f"[PIP] 使用策略：{strategy.strategy_name} 恢复{repair_desc}失败：{message}")
return False, last_error or f"恢复{repair_desc}失败"
@classmethod
@@ -1461,11 +1492,9 @@ class PluginHelper(metaclass=WeakSingleton):
seen_dirs.add(candidate_key)
resolved_dirs.append(candidate_path)
find_links_option = []
if resolved_dirs:
for local_wheels_dir in resolved_dirs:
logger.debug(f"[PIP] 发现可用的 wheels 目录: {local_wheels_dir}，将优先从本地安装。")
find_links_option.extend(["--find-links", str(local_wheels_dir)])
else:
logger.debug(f"[PIP] 未发现可用的 wheels 目录，将仅使用在线源。")
@@ -1484,23 +1513,32 @@ class PluginHelper(metaclass=WeakSingleton):
logger.error(f"[PIP] 创建运行环境约束文件失败：{e}")
return False, f"创建运行环境约束文件失败：{e}"
base_cmd = [sys.executable, "-m", "pip", "install"] + find_links_option
if constraints_file:
# 这里固定约束到主程序依赖的当前版本，避免共享 venv 被插件改写核心运行环境。
base_cmd.extend(["-c", str(constraints_file)])
base_cmd.extend(["-r", str(requirements_file)])
strategies = cls.__build_pip_install_strategies(base_cmd)
request = cls.__build_package_install_request(
requirements_file,
find_links_dirs=resolved_dirs,
constraints_file=constraints_file,
purpose="plugin",
)
strategies = build_package_install_strategies(request)
try:
# pip 会修改当前解释器的 site-packages，安装与缓存刷新必须串行，避免运行态模块被并发安装窗口污染。
with cls._pip_install_lock:
loaded_modules_before_install = set(sys.modules.keys())
# 遍历策略进行安装
for strategy_name, pip_command in strategies:
logger.debug(f"[PIP] 尝试使用策略：{strategy_name} 安装依赖，命令：{' '.join(pip_command)}")
success, message = SystemUtils.execute_with_subprocess(pip_command)
last_error = ""
for strategy in strategies:
logger.debug(
f"[PIP] 尝试使用策略：{strategy.strategy_name} 安装依赖，"
f"命令：{' '.join(strategy.safe_log_command)}"
)
success, message = SystemUtils.execute_with_subprocess(
strategy.command,
env=strategy.env,
safe_command=strategy.safe_log_command,
)
if success:
logger.debug(f"[PIP] 策略：{strategy_name} 安装依赖成功，输出：{message}")
logger.debug(f"[PIP] 策略：{strategy.strategy_name} 安装依赖成功，输出：{message}")
health_ok, health_message = cls.__run_runtime_healthcheck()
if not health_ok:
logger.error(f"[PIP] 依赖安装后运行环境自检失败：{health_message}")
@@ -1532,11 +1570,22 @@ class PluginHelper(metaclass=WeakSingleton):
logger.debug(f"[PIP] 已刷新导入系统，新加载的模块: {loaded_modules_during_install}")
return True, message
logger.error(f"[PIP] 策略：{strategy_name} 安装依赖失败，错误信息：{message}")
last_error = message
repair_ok, repair_message = cls.__repair_if_runtime_broken(
constraints_file if protected_packages else None
)
logger.error(f"[PIP] 策略：{strategy.strategy_name} 安装依赖失败，错误信息：{message}")
if not repair_ok or repair_message:
return False, (
f"策略 {strategy.strategy_name} 安装依赖失败：{message}；"
f"{repair_message}"
)
finally:
if constraints_file:
constraints_file.unlink(missing_ok=True)
if last_error:
return False, f"[PIP] 所有策略均安装依赖失败：{last_error}"
return False, "[PIP] 所有策略均安装依赖失败，请检查网络连接、PIP 配置或插件依赖约束"
@staticmethod

18
app/startup/modules_initializer.py
View File

@@ -73,6 +73,24 @@ def clear_temp():
SystemUtils.clear(settings.TEMP_PATH, days=settings.TEMP_FILE_DAYS)
# 清理图片缓存目录中7天前的文件
SystemUtils.clear(settings.CACHE_PATH / "images", days=settings.GLOBAL_IMAGE_CACHE_DAYS)
# 清理 pip/uv 包下载缓存，不接管整个 .cache 目录。
clear_package_tool_cache()
def clear_package_tool_cache():
"""
清理 pip/uv 包下载缓存，只处理 MoviePilot 管理的工具子目录。
"""
days = settings.PACKAGE_CACHE_DAYS
if days <= 0:
return
tool_cache_root = settings.PACKAGE_CACHE_PATH
for child in ("pip", "uv"):
cache_path = tool_cache_root / child
try:
SystemUtils.clear(cache_path, days=days)
except Exception as err:
logger.warning("清理包下载缓存失败：%s - %s", cache_path, err)
def user_auth():

67
app/utils/system.py
View File

@@ -6,6 +6,7 @@ import re
import shutil
import subprocess
import sys
import urllib.parse
import uuid
from pathlib import Path
from typing import List, Optional, Tuple, Union
@@ -20,6 +21,8 @@ class SystemUtils:
系统工具类，提供系统相关的操作和信息获取方法。
"""
_URL_WITH_USERINFO_PATTERN = re.compile(r"([A-Za-z][A-Za-z0-9+.-]*://[^\s]+)")
@staticmethod
def execute(cmd: str) -> str:
"""
@@ -33,22 +36,69 @@ class SystemUtils:
return ""
@staticmethod
def execute_with_subprocess(pip_command: list) -> Tuple[bool, str]:
def redact_url_userinfo(value: str) -> str:
"""
脱敏 URL 中的 userinfo，避免命令输出泄露镜像源或代理凭据。
"""
def replace(match: re.Match[str]) -> str:
candidate = match.group(1)
trailing = ""
while candidate and candidate[-1] in ".,;:)":
trailing = candidate[-1] + trailing
candidate = candidate[:-1]
parsed = urllib.parse.urlsplit(candidate)
if not parsed.username and not parsed.password:
return match.group(1)
host = parsed.netloc.rsplit("@", 1)[-1]
redacted = urllib.parse.urlunsplit((
parsed.scheme,
host,
parsed.path,
parsed.query,
parsed.fragment,
))
return f"{redacted}{trailing}"
return SystemUtils._URL_WITH_USERINFO_PATTERN.sub(replace, value or "")
@staticmethod
def redact_command_url_userinfo(command: list[str]) -> List[str]:
"""
脱敏命令参数中的 URL userinfo，供错误信息展示。
"""
return [SystemUtils.redact_url_userinfo(str(item)) for item in command]
@staticmethod
def execute_with_subprocess(
pip_command: list,
env: Optional[dict[str, str]] = None,
safe_command: Optional[list[str]] = None,
) -> Tuple[bool, str]:
"""
执行命令并捕获标准输出和错误输出，记录日志。
:param pip_command: 要执行的命令，以列表形式提供
:param env: 传递给子进程的环境变量
:param safe_command: 用于错误信息展示的脱敏命令
:return: (命令是否成功, 输出信息或错误信息)
"""
display_command = safe_command or pip_command
try:
# 使用 subprocess.run 捕获标准输出和标准错误
result = subprocess.run(pip_command, check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
result = subprocess.run(
pip_command,
check=True,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
)
# 合并 stdout 和 stderr
output = result.stdout + result.stderr
output = SystemUtils.redact_url_userinfo(result.stdout + result.stderr)
return True, output
except subprocess.CalledProcessError as e:
stdout = (e.stdout or "").strip()
stderr = (e.stderr or "").strip()
stdout = SystemUtils.redact_url_userinfo((e.stdout or "").strip())
stderr = SystemUtils.redact_url_userinfo((e.stderr or "").strip())
# 不同命令/兼容层可能把失败原因写入 stdout，失败时需要同时保留两路输出。
output_parts = []
if stdout:
@@ -58,12 +108,15 @@ class SystemUtils:
if not output_parts:
output_parts.append("无标准输出或错误输出")
error_message = (
f"命令：{' '.join(pip_command)}，执行失败，"
f"命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，执行失败，"
f"返回码：{e.returncode}，{'; '.join(output_parts)}"
)
return False, error_message
except Exception as e:
error_message = f"未知错误，命令：{' '.join(pip_command)}，错误：{str(e)}"
error_message = (
f"未知错误，命令：{' '.join(SystemUtils.redact_command_url_userinfo(display_command))}，"
f"错误：{SystemUtils.redact_url_userinfo(str(e))}"
)
return False, error_message
@staticmethod

22
docker/entrypoint.sh
View File

@@ -27,6 +27,23 @@ export PATH="${VENV_PATH}/bin:$PATH"
# 校正设置目录
CONFIG_DIR="${CONFIG_DIR:-/config}"
function apply_package_cache_env() {
PACKAGE_CACHE_ROOT="${PACKAGE_CACHE_ROOT:-${CONFIG_DIR}/.cache}"
export PACKAGE_CACHE_ROOT
export PIP_CACHE_DIR="${PIP_CACHE_DIR:-${PACKAGE_CACHE_ROOT}/pip}"
export UV_CACHE_DIR="${UV_CACHE_DIR:-${PACKAGE_CACHE_ROOT}/uv}"
mkdir -p "${PIP_CACHE_DIR}" "${UV_CACHE_DIR}"
}
function apply_package_proxy_env() {
if [ -n "${PROXY_HOST}" ]; then
export HTTP_PROXY="${PROXY_HOST}"
export HTTPS_PROXY="${PROXY_HOST}"
export http_proxy="${PROXY_HOST}"
export https_proxy="${PROXY_HOST}"
fi
}
# 环境变量补全
# 优先级: 系统环境变量 -> .env 文件 (即使为空字符串) -> 预设默认值
# 精准适配 Python 端 set_key (quote_mode="always", 单引号包裹, \' 转义)
@@ -39,6 +56,7 @@ function load_config_from_app_env() {
declare -A vars_and_default_values=(
# update.sh
["PIP_PROXY"]=""
["PACKAGE_CACHE_ROOT"]=""
["GITHUB_PROXY"]=""
["PROXY_HOST"]=""
["GITHUB_TOKEN"]=""
@@ -276,11 +294,10 @@ function ensure_backend_runtime_dependencies() {
fi
WARN "→ 检测到后端核心依赖异常，开始尝试恢复主程序依赖..."
apply_package_proxy_env
local -a pip_cmd=("${VENV_PATH}/bin/pip" "install" "-r" "/app/requirements.txt")
if [ -n "${PIP_PROXY}" ]; then
pip_cmd+=("-i" "${PIP_PROXY}")
elif [ -n "${PROXY_HOST}" ]; then
pip_cmd+=("--proxy" "${PROXY_HOST}")
fi
if ! "${pip_cmd[@]}" > /dev/stdout 2> /dev/stderr; then
@@ -298,6 +315,7 @@ function ensure_backend_runtime_dependencies() {
# 使用env配置
load_config_from_app_env
apply_package_cache_env
# 一次性升级标记仅影响本次启动，避免把临时升级模式带入运行中的 Python 进程
ONE_SHOT_UPDATE_FLAG="${CONFIG_DIR}/temp/moviepilot.pending_update"

36
docker/update.sh
View File

@@ -24,6 +24,27 @@ function WARN() {
VENV_PATH="${VENV_PATH:-/opt/venv}"
export PATH="${VENV_PATH}/bin:$PATH"
CONFIG_DIR="${CONFIG_DIR:-/config}"
function apply_package_cache_env() {
PACKAGE_CACHE_ROOT="${PACKAGE_CACHE_ROOT:-${CONFIG_DIR}/.cache}"
export PACKAGE_CACHE_ROOT
export PIP_CACHE_DIR="${PIP_CACHE_DIR:-${PACKAGE_CACHE_ROOT}/pip}"
export UV_CACHE_DIR="${UV_CACHE_DIR:-${PACKAGE_CACHE_ROOT}/uv}"
mkdir -p "${PIP_CACHE_DIR}" "${UV_CACHE_DIR}"
}
apply_package_cache_env
function apply_package_proxy_env() {
if [[ -n "${PROXY_HOST}" ]]; then
export HTTP_PROXY="${PROXY_HOST}"
export HTTPS_PROXY="${PROXY_HOST}"
export http_proxy="${PROXY_HOST}"
export https_proxy="${PROXY_HOST}"
fi
}
# 下载及解压
function download_and_unzip() {
local retries=0
@@ -176,9 +197,16 @@ function test_connectivity_pip() {
case "$1" in
0)
if [[ -n "${PIP_PROXY}" ]]; then
if ${VENV_PATH}/bin/pip install -i ${PIP_PROXY} pip-hello-world > /dev/null 2>&1; then
if [[ -n "${PROXY_HOST}" ]]; then
HTTP_PROXY="${PROXY_HOST}" HTTPS_PROXY="${PROXY_HOST}" http_proxy="${PROXY_HOST}" https_proxy="${PROXY_HOST}" \
${VENV_PATH}/bin/pip install -i ${PIP_PROXY} pip-hello-world > /dev/null 2>&1
else
${VENV_PATH}/bin/pip install -i ${PIP_PROXY} pip-hello-world > /dev/null 2>&1
fi
if [[ $? -eq 0 ]]; then
PIP_OPTIONS="-i ${PIP_PROXY}"
PIP_LOG="镜像代理模式"
apply_package_proxy_env
return 0
fi
fi
@@ -186,9 +214,11 @@ function test_connectivity_pip() {
;;
1)
if [[ -n "${PROXY_HOST}" ]]; then
if ${VENV_PATH}/bin/pip install --proxy=${PROXY_HOST} pip-hello-world > /dev/null 2>&1; then
PIP_OPTIONS="--proxy=${PROXY_HOST}"
if HTTP_PROXY="${PROXY_HOST}" HTTPS_PROXY="${PROXY_HOST}" http_proxy="${PROXY_HOST}" https_proxy="${PROXY_HOST}" \
${VENV_PATH}/bin/pip install pip-hello-world > /dev/null 2>&1; then
PIP_OPTIONS=""
PIP_LOG="全局代理模式"
apply_package_proxy_env
return 0
fi
fi

1
requirements-dev.in
View File

@@ -5,3 +5,4 @@ pylint~=4.0.6
pytest~=9.0.3
pytest-cov~=7.1.0
pytest-timeout~=2.4.0
uv~=0.11.23

196
scripts/dev/simulate_docker_package_env.sh Normal file
View File

@@ -0,0 +1,196 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
TMP_DIR="$(mktemp -d)"
trap 'rm -rf "${TMP_DIR}"' EXIT
mkdir -p "${TMP_DIR}/venv/bin" "${TMP_DIR}/config"
cat > "${TMP_DIR}/venv/bin/pip" <<'SH'
#!/usr/bin/env bash
printf 'argv=%s\n' "$*" >> "${MP_FAKE_PIP_LOG}"
printf 'HTTP_PROXY=%s\n' "${HTTP_PROXY:-}" >> "${MP_FAKE_PIP_LOG}"
printf 'HTTPS_PROXY=%s\n' "${HTTPS_PROXY:-}" >> "${MP_FAKE_PIP_LOG}"
printf 'PACKAGE_CACHE_ROOT=%s\n' "${PACKAGE_CACHE_ROOT:-}" >> "${MP_FAKE_PIP_LOG}"
printf 'PIP_CACHE_DIR=%s\n' "${PIP_CACHE_DIR:-}" >> "${MP_FAKE_PIP_LOG}"
printf 'UV_CACHE_DIR=%s\n' "${UV_CACHE_DIR:-}" >> "${MP_FAKE_PIP_LOG}"
if [ "${MP_FAKE_PIP_FAIL:-}" = "1" ]; then
exit 1
fi
exit 0
SH
chmod +x "${TMP_DIR}/venv/bin/pip"
assert_contains() {
local needle="$1"
local file="$2"
if ! grep -Fq -- "$needle" "$file"; then
echo "missing expected text: $needle" >&2
cat "$file" >&2
exit 1
fi
}
assert_not_contains() {
local needle="$1"
local file="$2"
if grep -Fq -- "$needle" "$file"; then
echo "unexpected text: $needle" >&2
cat "$file" >&2
exit 1
fi
}
UPDATE_FUNCS="${TMP_DIR}/update-functions.sh"
awk '
BEGIN {capture=1}
/^if \[\[ "\$\{MOVIEPILOT_AUTO_UPDATE\}"/ {capture=0}
capture {print}
' "${ROOT}/docker/update.sh" > "${UPDATE_FUNCS}"
MP_FAKE_PIP_LOG="${TMP_DIR}/update.log"
export MP_FAKE_PIP_LOG
export VENV_PATH="${TMP_DIR}/venv"
export CONFIG_DIR="${TMP_DIR}/config"
export MOVIEPILOT_AUTO_UPDATE=false
export PIP_PROXY="https://mirror.example/simple"
export PROXY_HOST="http://proxy.example:7890"
unset PACKAGE_CACHE_ROOT PIP_CACHE_DIR UV_CACHE_DIR
source "${UPDATE_FUNCS}" >/dev/null
: > "${MP_FAKE_PIP_LOG}"
test_connectivity_pip 0
assert_contains "argv=install -i https://mirror.example/simple pip-hello-world" "${MP_FAKE_PIP_LOG}"
assert_contains "HTTPS_PROXY=http://proxy.example:7890" "${MP_FAKE_PIP_LOG}"
assert_contains "PACKAGE_CACHE_ROOT=${TMP_DIR}/config/.cache" "${MP_FAKE_PIP_LOG}"
assert_contains "PIP_CACHE_DIR=${TMP_DIR}/config/.cache/pip" "${MP_FAKE_PIP_LOG}"
assert_contains "UV_CACHE_DIR=${TMP_DIR}/config/.cache/uv" "${MP_FAKE_PIP_LOG}"
if [[ "${PIP_OPTIONS}" != "-i ${PIP_PROXY}" ]]; then
echo "mirror branch must preserve index option: ${PIP_OPTIONS}" >&2
exit 1
fi
if [[ "${PIP_OPTIONS}" == *"--proxy"* ]]; then
echo "PIP_OPTIONS must not contain --proxy: ${PIP_OPTIONS}" >&2
exit 1
fi
assert_not_contains "user:pass" "${MP_FAKE_PIP_LOG}"
: > "${MP_FAKE_PIP_LOG}"
PIP_PROXY=""
test_connectivity_pip 1
assert_contains "argv=install pip-hello-world" "${MP_FAKE_PIP_LOG}"
assert_contains "HTTPS_PROXY=http://proxy.example:7890" "${MP_FAKE_PIP_LOG}"
if [[ -n "${PIP_OPTIONS}" ]]; then
echo "proxy branch must keep PIP_OPTIONS empty: ${PIP_OPTIONS}" >&2
exit 1
fi
MP_FAKE_PIP_LOG="${TMP_DIR}/update-explicit-cache.log"
export MP_FAKE_PIP_LOG
(
export VENV_PATH="${TMP_DIR}/venv"
export CONFIG_DIR="${TMP_DIR}/config"
export MOVIEPILOT_AUTO_UPDATE=false
export PACKAGE_CACHE_ROOT="${TMP_DIR}/update-custom-package-cache"
export PIP_CACHE_DIR="${TMP_DIR}/explicit-pip-cache"
export UV_CACHE_DIR="${TMP_DIR}/explicit-uv-cache"
export PIP_PROXY="https://mirror.example/simple"
export PROXY_HOST="http://proxy.example:7890"
source "${UPDATE_FUNCS}" >/dev/null
test_connectivity_pip 0
)
assert_contains "PACKAGE_CACHE_ROOT=${TMP_DIR}/update-custom-package-cache" "${MP_FAKE_PIP_LOG}"
assert_contains "PIP_CACHE_DIR=${TMP_DIR}/explicit-pip-cache" "${MP_FAKE_PIP_LOG}"
assert_contains "UV_CACHE_DIR=${TMP_DIR}/explicit-uv-cache" "${MP_FAKE_PIP_LOG}"
MP_FAKE_PIP_LOG="${TMP_DIR}/update-fallback-no-proxy.log"
export MP_FAKE_PIP_LOG
(
export VENV_PATH="${TMP_DIR}/venv"
export CONFIG_DIR="${TMP_DIR}/config"
export MOVIEPILOT_AUTO_UPDATE=false
export PIP_PROXY="https://mirror.example/simple"
export PROXY_HOST="http://proxy.example:7890"
unset PACKAGE_CACHE_ROOT PIP_CACHE_DIR UV_CACHE_DIR HTTP_PROXY HTTPS_PROXY http_proxy https_proxy
source "${UPDATE_FUNCS}" >/dev/null
MP_FAKE_PIP_FAIL=1 test_connectivity_pip 0 && exit 1
if [[ -n "${HTTPS_PROXY:-}" || -n "${https_proxy:-}" ]]; then
echo "mirror failure must not leak proxy env" >&2
env | grep -E '^(HTTP_PROXY|HTTPS_PROXY|http_proxy|https_proxy)=' >&2 || true
exit 1
fi
test_connectivity_pip 2
if [[ "${PIP_LOG}" != "不使用代理" ]]; then
echo "fallback branch must report direct mode: ${PIP_LOG}" >&2
exit 1
fi
)
ENTRYPOINT_FUNCS="${TMP_DIR}/entrypoint-functions.sh"
awk '
BEGIN {capture=1}
/^# 使用env配置/ {capture=0}
capture {print}
' "${ROOT}/docker/entrypoint.sh" > "${ENTRYPOINT_FUNCS}"
cat > "${TMP_DIR}/venv/bin/python3" <<'SH'
#!/usr/bin/env bash
count_file="${MP_FAKE_PYTHON_COUNT}"
count=0
if [ -f "$count_file" ]; then
count="$(cat "$count_file")"
fi
count=$((count + 1))
printf '%s' "$count" > "$count_file"
if [ "$count" -eq 1 ]; then
exit 1
fi
exit 0
SH
chmod +x "${TMP_DIR}/venv/bin/python3"
MP_FAKE_PIP_LOG="${TMP_DIR}/entrypoint.log"
MP_FAKE_PYTHON_COUNT="${TMP_DIR}/python-count"
export MP_FAKE_PIP_LOG MP_FAKE_PYTHON_COUNT
(
export VENV_PATH="${TMP_DIR}/venv"
export CONFIG_DIR="${TMP_DIR}/config"
unset PACKAGE_CACHE_ROOT PIP_CACHE_DIR UV_CACHE_DIR
export PIP_PROXY=""
export PROXY_HOST="http://proxy.example:7890"
source "${ENTRYPOINT_FUNCS}"
apply_package_cache_env
ensure_backend_runtime_dependencies
) >/dev/null
assert_contains "argv=install -r /app/requirements.txt" "${MP_FAKE_PIP_LOG}"
assert_contains "HTTPS_PROXY=http://proxy.example:7890" "${MP_FAKE_PIP_LOG}"
assert_contains "PACKAGE_CACHE_ROOT=${TMP_DIR}/config/.cache" "${MP_FAKE_PIP_LOG}"
assert_contains "PIP_CACHE_DIR=${TMP_DIR}/config/.cache/pip" "${MP_FAKE_PIP_LOG}"
assert_contains "UV_CACHE_DIR=${TMP_DIR}/config/.cache/uv" "${MP_FAKE_PIP_LOG}"
assert_not_contains "--proxy" "${MP_FAKE_PIP_LOG}"
MP_FAKE_PIP_LOG="${TMP_DIR}/entrypoint-app-env.log"
MP_FAKE_PYTHON_COUNT="${TMP_DIR}/python-count-app-env"
cat > "${TMP_DIR}/config/app.env" <<EOF
PACKAGE_CACHE_ROOT='${TMP_DIR}/app-env-custom-package-cache'
PROXY_HOST='http://proxy.example:7890'
EOF
export MP_FAKE_PIP_LOG MP_FAKE_PYTHON_COUNT
(
export VENV_PATH="${TMP_DIR}/venv"
export CONFIG_DIR="${TMP_DIR}/config"
unset PACKAGE_CACHE_ROOT PIP_CACHE_DIR UV_CACHE_DIR PIP_PROXY PROXY_HOST
source "${ENTRYPOINT_FUNCS}"
load_config_from_app_env
apply_package_cache_env
ensure_backend_runtime_dependencies
) >/dev/null
assert_contains "PACKAGE_CACHE_ROOT=${TMP_DIR}/app-env-custom-package-cache" "${MP_FAKE_PIP_LOG}"
assert_contains "PIP_CACHE_DIR=${TMP_DIR}/app-env-custom-package-cache/pip" "${MP_FAKE_PIP_LOG}"
assert_contains "UV_CACHE_DIR=${TMP_DIR}/app-env-custom-package-cache/uv" "${MP_FAKE_PIP_LOG}"
echo "Docker package env simulation passed"

74
scripts/dev/simulate_package_installer.py Normal file
View File

@@ -0,0 +1,74 @@
#!/usr/bin/env python3
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
from app.helper.package_installer import PackageInstallRequest, build_package_install_strategies
def sample(name: str, request: PackageInstallRequest) -> None:
print(f"## {name}")
strategies = build_package_install_strategies(request)
assert strategies, f"{name}: no strategies generated"
for strategy in strategies:
rendered = " ".join(strategy.safe_log_command)
print(strategy.strategy_name)
print(rendered)
assert all("--proxy" not in arg for arg in strategy.command)
assert "user:pass" not in rendered
assert strategy.env["PIP_CACHE_DIR"].endswith("/.cache/pip")
assert strategy.env["UV_CACHE_DIR"].endswith("/.cache/uv")
if strategy.strategy_name.endswith("代理") or strategy.strategy_name.endswith("镜像+代理"):
assert strategy.env["HTTPS_PROXY"] == "http://proxy.example:7890"
def main() -> None:
root = ROOT
config_dir = root / "config"
python_bin = root.parent / ".venv-test" / "bin" / "python"
requirements = root / "requirements.txt"
samples = {
"plain": PackageInstallRequest(
requirements_file=requirements,
python_bin=python_bin,
config_dir=config_dir,
),
"mirror": PackageInstallRequest(
requirements_file=requirements,
python_bin=python_bin,
config_dir=config_dir,
pip_index_url="https://user:pass@mirror.example/simple",
),
"proxy": PackageInstallRequest(
requirements_file=requirements,
python_bin=python_bin,
config_dir=config_dir,
proxy_url="http://proxy.example:7890",
),
"mirror_proxy_wheels": PackageInstallRequest(
requirements_file=requirements,
python_bin=python_bin,
config_dir=config_dir,
find_links_dirs=[
root / "plugins.v2" / "demo" / "wheels",
root / "plugins.v2" / "other" / "wheels",
],
pip_index_url="https://user:pass@mirror.example/simple",
proxy_url="http://proxy.example:7890",
),
}
for name, request in samples.items():
sample(name, request)
print("Package installer simulation passed")
if __name__ == "__main__":
main()

57
scripts/local_setup.py
View File

@@ -17,6 +17,7 @@ import subprocess
import sys
import tarfile
import textwrap
import urllib.parse
import uuid
import zipfile
from datetime import datetime
@@ -493,15 +494,60 @@ def print_step(message: str) -> None:
print(f"==> {message}")
def _redact_url(value: str) -> str:
parsed = urllib.parse.urlsplit(value)
if "@" not in parsed.netloc:
return value
host = parsed.netloc.rsplit("@", 1)[-1]
return urllib.parse.urlunsplit(
(parsed.scheme, host, parsed.path, parsed.query, parsed.fragment)
)
def redact_command(command: list[str]) -> list[str]:
redacted: list[str] = []
for item in command:
value = str(item)
url_marker = value.find("://")
equals_marker = value.find("=")
if url_marker >= 0 and 0 <= equals_marker < url_marker:
key, separator, url = value.partition("=")
value = f"{key}{separator}{_redact_url(url)}"
elif url_marker >= 0:
value = _redact_url(value)
redacted.append(value)
return redacted
def build_package_install_env() -> dict[str, str]:
env = os.environ.copy()
package_cache_root = env.get("PACKAGE_CACHE_ROOT", "").strip() or str(CONFIG_DIR / ".cache")
env.setdefault("PACKAGE_CACHE_ROOT", package_cache_root)
env.setdefault("PIP_CACHE_DIR", os.path.join(package_cache_root, "pip"))
env.setdefault("UV_CACHE_DIR", os.path.join(package_cache_root, "uv"))
index_url = env.get("PIP_PROXY", "").strip()
if index_url:
env["PIP_INDEX_URL"] = index_url
env["UV_DEFAULT_INDEX"] = index_url
proxy = env.get("PROXY_HOST", "").strip()
if proxy:
for key in ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"):
env[key] = proxy
return env
def run(
command: list[str],
cwd: Optional[Path] = None,
env: Optional[dict[str, str]] = None,
safe_command: Optional[list[str]] = None,
) -> None:
"""
执行安装步骤中的外部命令，并在失败时让调用方中断流程。
"""
pretty = " ".join(command)
pretty = " ".join(safe_command or redact_command(command))
print(f"+ {pretty}")
subprocess.run(command, cwd=str(cwd or ROOT), check=True, env=env)
@@ -597,7 +643,8 @@ def _ensure_uv_available_for_venv(venv_dir: Path, venv_python: Path) -> Optional
return uv_bin
print_step("当前未检测到 uv，先在虚拟环境内安装 uv")
run([str(venv_python), "-m", "pip", "install", "--upgrade", "pip", "uv"])
command = [str(venv_python), "-m", "pip", "install", "--upgrade", "pip", "uv"]
run(command, env=build_package_install_env(), safe_command=redact_command(command))
if uv_bin.exists():
return uv_bin
raise RuntimeError("uv 安装完成，但虚拟环境中未找到 uv 可执行文件")
@@ -2677,13 +2724,15 @@ def install_deps(*, python_bin: str, venv_dir: Path, recreate: bool) -> Path:
if os.name == "nt":
print_step("升级 pip")
run([str(venv_python), "-m", "pip", "install", "--upgrade", "pip"])
command = [str(venv_python), "-m", "pip", "install", "--upgrade", "pip"]
run(command, env=build_package_install_env(), safe_command=redact_command(command))
else:
print_step("为虚拟环境配置 uv 兼容 pip 命令")
venv_pip = configure_venv_pip_compat(venv_dir, venv_python)
print_step("安装项目依赖")
run([str(venv_pip), "install", "-r", str(ROOT / "requirements.txt")])
command = [str(venv_pip), "install", "-r", str(ROOT / "requirements.txt")]
run(command, env=build_package_install_env(), safe_command=redact_command(command))
install_browser_runtime(venv_python)
return venv_python

52
scripts/uv-pip-compat.sh
View File

@@ -40,6 +40,49 @@ has_environment_option() {
return 1
}
normalize_pip_proxy_args() {
output_file="$1"
shift
original_args_file=$(mktemp)
: > "${output_file}"
trap 'rm -f "${original_args_file}"' EXIT HUP INT TERM
for arg in "$@"; do
printf '%s\n' "$arg" >> "${original_args_file}"
done
skip_next=0
while IFS= read -r arg; do
if [ "${skip_next}" -eq 1 ]; then
proxy_value="${arg}"
export HTTP_PROXY="${proxy_value}"
export HTTPS_PROXY="${proxy_value}"
export http_proxy="${proxy_value}"
export https_proxy="${proxy_value}"
skip_next=0
continue
fi
case "$arg" in
--proxy)
skip_next=1
;;
--proxy=*)
proxy_value="${arg#--proxy=}"
export HTTP_PROXY="${proxy_value}"
export HTTPS_PROXY="${proxy_value}"
export http_proxy="${proxy_value}"
export https_proxy="${proxy_value}"
;;
*)
printf '%s\n' "$arg" >> "${output_file}"
;;
esac
done < "${original_args_file}"
rm -f "${original_args_file}"
trap - EXIT HUP INT TERM
}
uv_pip_with_venv_python() {
command_name="$1"
shift
@@ -69,6 +112,15 @@ case "${COMMAND_NAME}" in
check|freeze|install|list|show|sync|tree|uninstall)
pip_command="$1"
shift
if [ "${pip_command}" = "install" ]; then
normalized_file=$(mktemp)
normalize_pip_proxy_args "${normalized_file}" "$@"
set --
while IFS= read -r arg; do
set -- "$@" "$arg"
done < "${normalized_file}"
rm -f "${normalized_file}"
fi
uv_pip_with_venv_python "${pip_command}" "$@"
;;
*)

134
tests/test_cache_system.py
View File

@@ -1,4 +1,6 @@
import asyncio
import os
import time
from app.core.cache import AsyncFileBackend, FileBackend, MemoryBackend
from app.core.config import settings
@@ -19,6 +21,138 @@ def test_file_backend_items_keep_relative_keys_and_bytes(tmp_path):
assert not cache.exists("nested/poster.jpg", region="images")
def test_clear_package_tool_cache_only_removes_pip_and_uv_old_files(tmp_path, monkeypatch):
"""
包安装工具缓存清理只处理 pip/uv 子目录，不接管整个 .cache 或业务缓存。
"""
from app.startup.modules_initializer import clear_package_tool_cache
old_time = time.time() - 40 * 24 * 3600
cache_root = tmp_path / ".cache"
old_pip = cache_root / "pip" / "old.whl"
old_uv = cache_root / "uv" / "old.archive"
unknown = cache_root / "other" / "old.bin"
business = tmp_path / "cache" / "images" / "old.jpg"
for path in (old_pip, old_uv, unknown, business):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("x", encoding="utf-8")
os.utime(path, (old_time, old_time))
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
monkeypatch.setattr(settings, "PACKAGE_CACHE_ROOT", None)
monkeypatch.setattr(settings, "PACKAGE_CACHE_DAYS", 30)
clear_package_tool_cache()
assert not old_pip.exists()
assert not old_uv.exists()
assert unknown.exists()
assert business.exists()
def test_clear_package_tool_cache_disabled_when_days_non_positive(tmp_path, monkeypatch):
"""
PACKAGE_CACHE_DAYS 小于等于 0 时不清理包安装缓存。
"""
from app.startup.modules_initializer import clear_package_tool_cache
old_time = time.time() - 40 * 24 * 3600
old_pip = tmp_path / ".cache" / "pip" / "old.whl"
old_pip.parent.mkdir(parents=True, exist_ok=True)
old_pip.write_text("x", encoding="utf-8")
os.utime(old_pip, (old_time, old_time))
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
monkeypatch.setattr(settings, "PACKAGE_CACHE_ROOT", None)
monkeypatch.setattr(settings, "PACKAGE_CACHE_DAYS", 0)
clear_package_tool_cache()
assert old_pip.exists()
def test_clear_package_tool_cache_isolates_subdir_errors(tmp_path, monkeypatch):
"""
单个工具缓存目录清理失败，不影响另一个工具缓存目录。
"""
from app.startup.modules_initializer import clear_package_tool_cache
calls = []
def fake_clear(path, days):
calls.append((path.name, days))
if path.name == "pip":
raise OSError("pip cache locked")
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
monkeypatch.setattr(settings, "PACKAGE_CACHE_ROOT", str(tmp_path / "custom-package-cache"))
monkeypatch.setattr(settings, "PACKAGE_CACHE_DAYS", 30)
monkeypatch.setattr("app.startup.modules_initializer.SystemUtils.clear", fake_clear)
clear_package_tool_cache()
assert calls == [("pip", 30), ("uv", 30)]
def test_clear_package_tool_cache_uses_package_cache_root(tmp_path, monkeypatch):
"""
PACKAGE_CACHE_ROOT 用作 pip/uv 清理根目录，不扩大到配置目录下其他缓存。
"""
from app.startup.modules_initializer import clear_package_tool_cache
old_time = time.time() - 40 * 24 * 3600
package_cache_root = tmp_path / "custom-package-cache"
old_pip = package_cache_root / "pip" / "old.whl"
default_pip = tmp_path / ".cache" / "pip" / "old.whl"
for path in (old_pip, default_pip):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("x", encoding="utf-8")
os.utime(path, (old_time, old_time))
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
monkeypatch.setattr(settings, "PACKAGE_CACHE_ROOT", str(package_cache_root))
monkeypatch.setattr(settings, "PACKAGE_CACHE_DAYS", 30)
clear_package_tool_cache()
assert not old_pip.exists()
assert default_pip.exists()
def test_init_modules_does_not_clear_package_tool_cache(monkeypatch):
"""
包安装缓存清理由通用临时清理入口触发，模块启动路径不直接执行清理。
"""
from app.startup import modules_initializer
called = False
def fail_if_called():
nonlocal called
called = True
raise AssertionError("init_modules must not clear package tool cache directly")
monkeypatch.setattr(modules_initializer, "clear_package_tool_cache", fail_if_called)
monkeypatch.setattr(modules_initializer, "DisplayHelper", lambda: None)
monkeypatch.setattr(modules_initializer, "DohHelper", lambda: None)
monkeypatch.setattr(modules_initializer, "SitesHelper", lambda: None)
monkeypatch.setattr(modules_initializer, "ResourceHelper", lambda: None)
monkeypatch.setattr(modules_initializer, "user_auth", lambda: None)
monkeypatch.setattr(modules_initializer, "ModuleManager", lambda: None)
monkeypatch.setattr(modules_initializer.EventManager, "start", lambda self: None)
monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "init_plugin_report", lambda: None)
monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "init_subscribe_report", lambda: None)
monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "get_user_uuid", lambda: None)
monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "get_github_user", lambda: None)
monkeypatch.setattr(modules_initializer, "init_agent", lambda: None)
monkeypatch.setattr(modules_initializer, "start_frontend", lambda: None)
monkeypatch.setattr(modules_initializer, "check_auth", lambda: None)
modules_initializer.init_modules()
assert called is False
def test_file_backend_delete_missing_key_is_noop(tmp_path):
"""
删除不存在的文件缓存 key 应保持幂等，不向调用方抛出文件系统异常。

47
tests/test_cli_auto_update.py
View File

@@ -32,10 +32,12 @@ def load_cli_module():
ROOT_PATH=root,
FRONTEND_PATH=str(root / "public"),
CONFIG_PATH=root / "config",
PACKAGE_CACHE_PATH=root / "custom-package-cache",
HOST="127.0.0.1",
PORT=3001,
NGINX_PORT=3000,
PROXY_HOST="",
PIP_PROXY="",
GITHUB_TOKEN="",
PROXY={},
REPO_GITHUB_HEADERS=lambda _repo: {},
@@ -110,3 +112,48 @@ class CliAutoUpdateTests(unittest.TestCase):
command = run_mock.call_args.args[0]
self.assertEqual(command[1:5], [str(module._repo_root() / "scripts" / "local_setup.py"), "update", "all", "--ref"])
self.assertNotIn("--frontend-version", command)
def test_best_effort_auto_update_passes_package_env_and_overrides_proxy(self):
module = load_cli_module()
module.settings.PROXY_HOST = "http://proxy.example:7890"
module.settings.PIP_PROXY = "https://mirror.example/simple"
run_result = SimpleNamespace(returncode=0, stdout="ok")
with patch.dict(module.os.environ, {"HTTPS_PROXY": "http://old.example:8080"}, clear=False), patch.object(
module, "_auto_update_mode", return_value="release"
), patch.object(module, "_resolve_auto_update_targets", return_value="v2.10.12"), patch.object(
module.subprocess, "run", return_value=run_result
) as run_mock, patch.object(
module.click, "echo"
):
module._best_effort_auto_update()
env = run_mock.call_args.kwargs["env"]
self.assertEqual(env["HTTPS_PROXY"], "http://proxy.example:7890")
self.assertEqual(env["PIP_PROXY"], "https://mirror.example/simple")
self.assertEqual(env["PACKAGE_CACHE_ROOT"], str(module.settings.PACKAGE_CACHE_PATH))
self.assertEqual(env["PIP_CACHE_DIR"], str(module.settings.PACKAGE_CACHE_PATH / "pip"))
self.assertEqual(env["UV_CACHE_DIR"], str(module.settings.PACKAGE_CACHE_PATH / "uv"))
def test_best_effort_auto_update_derives_tool_cache_from_existing_root(self):
module = load_cli_module()
run_result = SimpleNamespace(returncode=0, stdout="ok")
package_cache_root = Path("/custom/package-cache-root")
with patch.dict(
module.os.environ,
{
"PACKAGE_CACHE_ROOT": str(package_cache_root),
},
clear=False,
), patch.object(module, "_auto_update_mode", return_value="release"), patch.object(
module, "_resolve_auto_update_targets", return_value="v2.10.12"
), patch.object(module.subprocess, "run", return_value=run_result) as run_mock, patch.object(
module.click, "echo"
):
module._best_effort_auto_update()
env = run_mock.call_args.kwargs["env"]
self.assertEqual(env["PACKAGE_CACHE_ROOT"], str(package_cache_root))
self.assertEqual(env["PIP_CACHE_DIR"], str(package_cache_root / "pip"))
self.assertEqual(env["UV_CACHE_DIR"], str(package_cache_root / "uv"))

234
tests/test_local_setup_config_dir.py
View File

@@ -83,7 +83,237 @@ class LocalSetupConfigDirTests(unittest.TestCase):
self.assertEqual(result, venv_python)
run_mock.assert_any_call(["python3", "-m", "venv", str(venv_dir)])
run_mock.assert_any_call(
[str(venv_pip), "install", "-r", str(module.ROOT / "requirements.txt")]
self.assertTrue(
any(
call.args[0] == [str(venv_pip), "install", "-r", str(module.ROOT / "requirements.txt")]
for call in run_mock.call_args_list
)
)
install_browser.assert_called_once_with(venv_python)
def test_package_install_env_maps_proxy_cache_and_index(self):
module = load_local_setup_module()
with tempfile.TemporaryDirectory() as temp_dir, patch.dict(
module.os.environ,
{
"PROXY_HOST": "http://proxy.example:7890",
"PIP_PROXY": "https://user:pass@mirror.example/simple",
"PACKAGE_CACHE_ROOT": str(Path(temp_dir) / "custom-package-cache"),
},
clear=False,
):
module.CONFIG_DIR = Path(temp_dir)
env = module.build_package_install_env()
self.assertEqual(env["HTTPS_PROXY"], "http://proxy.example:7890")
self.assertEqual(env["PACKAGE_CACHE_ROOT"], str(Path(temp_dir) / "custom-package-cache"))
self.assertEqual(env["PIP_CACHE_DIR"], str(Path(temp_dir) / "custom-package-cache" / "pip"))
self.assertEqual(env["UV_CACHE_DIR"], str(Path(temp_dir) / "custom-package-cache" / "uv"))
self.assertEqual(env["PIP_INDEX_URL"], "https://user:pass@mirror.example/simple")
self.assertEqual(env["UV_DEFAULT_INDEX"], "https://user:pass@mirror.example/simple")
def test_package_install_env_defaults_cache_to_config_dir(self):
module = load_local_setup_module()
with tempfile.TemporaryDirectory() as temp_dir, patch.dict(
module.os.environ,
{},
clear=True,
):
module.CONFIG_DIR = Path(temp_dir)
env = module.build_package_install_env()
self.assertEqual(env["PACKAGE_CACHE_ROOT"], str(Path(temp_dir) / ".cache"))
self.assertEqual(env["PIP_CACHE_DIR"], str(Path(temp_dir) / ".cache" / "pip"))
self.assertEqual(env["UV_CACHE_DIR"], str(Path(temp_dir) / ".cache" / "uv"))
def test_package_install_env_preserves_explicit_cache_dirs(self):
module = load_local_setup_module()
with tempfile.TemporaryDirectory() as temp_dir, patch.dict(
module.os.environ,
{
"PIP_CACHE_DIR": "/custom/pip-cache",
"UV_CACHE_DIR": "/custom/uv-cache",
"PACKAGE_CACHE_ROOT": "/custom/custom-package-cache",
},
clear=False,
):
module.CONFIG_DIR = Path(temp_dir)
env = module.build_package_install_env()
self.assertEqual(env["PACKAGE_CACHE_ROOT"], "/custom/custom-package-cache")
self.assertEqual(env["PIP_CACHE_DIR"], "/custom/pip-cache")
self.assertEqual(env["UV_CACHE_DIR"], "/custom/uv-cache")
def test_run_redacts_safe_command(self):
module = load_local_setup_module()
with patch.object(module.subprocess, "run"), patch("builtins.print") as print_mock:
module.run(
[
"python",
"-m",
"pip",
"install",
"-i",
"https://user:pass@mirror.example/simple",
],
safe_command=[
"python",
"-m",
"pip",
"install",
"-i",
"https://mirror.example/simple",
],
)
printed = " ".join(str(call.args[0]) for call in print_mock.call_args_list)
self.assertIn("https://mirror.example/simple", printed)
self.assertNotIn("user:pass", printed)
def test_redact_command_handles_inline_index_url(self):
module = load_local_setup_module()
command = [
"pip",
"install",
"--index-url=https://user:pass@mirror.example/simple",
]
redacted = module.redact_command(command)
self.assertIn("--index-url=https://mirror.example/simple", redacted)
self.assertNotIn("user:pass", " ".join(redacted))
def test_redact_command_handles_url_query_equals(self):
module = load_local_setup_module()
command = [
"pip",
"install",
"https://user:pass@mirror.example/simple?token=abc",
]
redacted = module.redact_command(command)
self.assertIn("https://mirror.example/simple?token=abc", redacted)
self.assertNotIn("user:pass", " ".join(redacted))
def test_uv_bootstrap_uses_package_env_and_index_without_visible_secret(self):
module = load_local_setup_module()
calls = []
with tempfile.TemporaryDirectory() as temp_dir, patch.dict(
module.os.environ,
{
"PROXY_HOST": "http://proxy.example:7890",
"PIP_PROXY": "https://user:pass@mirror.example/simple",
"PACKAGE_CACHE_ROOT": str(Path(temp_dir) / "custom-package-cache"),
},
clear=False,
):
venv_dir = Path(temp_dir) / "venv"
venv_python = venv_dir / "bin" / "python"
uv_bin = venv_dir / "bin" / "uv"
venv_python.parent.mkdir(parents=True)
venv_python.write_text("", encoding="utf-8")
module.CONFIG_DIR = Path(temp_dir) / "config"
def fake_run(command, cwd=None, env=None, safe_command=None):
calls.append((command, env, safe_command))
uv_bin.write_text("", encoding="utf-8")
with patch.object(module.shutil, "which", return_value=None), \
patch.object(module, "run", side_effect=fake_run):
module._ensure_uv_available_for_venv(venv_dir, venv_python)
command, env, safe_command = calls[0]
self.assertEqual(command, [str(venv_python), "-m", "pip", "install", "--upgrade", "pip", "uv"])
self.assertEqual(env["PIP_INDEX_URL"], "https://user:pass@mirror.example/simple")
self.assertEqual(env["UV_DEFAULT_INDEX"], "https://user:pass@mirror.example/simple")
self.assertEqual(env["HTTPS_PROXY"], "http://proxy.example:7890")
self.assertEqual(env["PACKAGE_CACHE_ROOT"], str(Path(temp_dir) / "custom-package-cache"))
self.assertEqual(env["PIP_CACHE_DIR"], str(Path(temp_dir) / "custom-package-cache" / "pip"))
self.assertEqual(env["UV_CACHE_DIR"], str(Path(temp_dir) / "custom-package-cache" / "uv"))
self.assertNotIn("user:pass", " ".join(safe_command or command))
def test_windows_pip_upgrade_uses_package_env(self):
module = load_local_setup_module()
calls = []
with tempfile.TemporaryDirectory() as temp_dir, patch.dict(
module.os.environ,
{
"PROXY_HOST": "http://proxy.example:7890",
"PIP_PROXY": "https://user:pass@mirror.example/simple",
"PACKAGE_CACHE_ROOT": str(Path(temp_dir) / "custom-package-cache"),
},
clear=False,
):
root = Path(temp_dir)
venv_dir = root / "venv"
venv_python = venv_dir / "Scripts" / "python.exe"
venv_pip = venv_dir / "Scripts" / "pip.exe"
venv_pip.parent.mkdir(parents=True)
venv_python.write_text("", encoding="utf-8")
venv_pip.write_text("", encoding="utf-8")
module.CONFIG_DIR = root / "config"
def fake_run(command, cwd=None, env=None, safe_command=None):
calls.append((command, env, safe_command))
with patch.object(module.os, "name", "nt"), \
patch.object(module, "ensure_supported_python"), \
patch.object(module, "install_browser_runtime"), \
patch.object(module, "run", side_effect=fake_run):
module.install_deps(python_bin="python", venv_dir=venv_dir, recreate=False)
pip_upgrade = [
item for item in calls
if item[0][1:] == ["-m", "pip", "install", "--upgrade", "pip"]
][0]
self.assertEqual(pip_upgrade[1]["PIP_INDEX_URL"], "https://user:pass@mirror.example/simple")
self.assertEqual(pip_upgrade[1]["UV_DEFAULT_INDEX"], "https://user:pass@mirror.example/simple")
self.assertEqual(pip_upgrade[1]["HTTPS_PROXY"], "http://proxy.example:7890")
self.assertEqual(pip_upgrade[1]["PACKAGE_CACHE_ROOT"], str(Path(temp_dir) / "custom-package-cache"))
self.assertEqual(pip_upgrade[1]["PIP_CACHE_DIR"], str(Path(temp_dir) / "custom-package-cache" / "pip"))
self.assertEqual(pip_upgrade[1]["UV_CACHE_DIR"], str(Path(temp_dir) / "custom-package-cache" / "uv"))
self.assertNotIn("user:pass", " ".join(pip_upgrade[2] or pip_upgrade[0]))
def test_install_deps_uses_package_env_for_project_requirements(self):
module = load_local_setup_module()
calls = []
with tempfile.TemporaryDirectory() as temp_dir, patch.dict(
module.os.environ,
{"PIP_PROXY": "https://user:pass@mirror.example/simple"},
clear=False,
):
root = Path(temp_dir)
venv_dir = root / "venv"
venv_python = venv_dir / "bin" / "python"
venv_pip = venv_dir / "bin" / "pip"
venv_pip.parent.mkdir(parents=True)
venv_python.write_text("", encoding="utf-8")
venv_pip.write_text("", encoding="utf-8")
module.CONFIG_DIR = root / "config"
def fake_run(command, cwd=None, env=None, safe_command=None):
calls.append((command, env, safe_command))
with patch.object(module, "ensure_supported_python"), \
patch.object(module, "configure_venv_pip_compat", return_value=venv_pip), \
patch.object(module, "install_browser_runtime"), \
patch.object(module, "run", side_effect=fake_run):
module.install_deps(python_bin="python3", venv_dir=venv_dir, recreate=False)
project_install = [
item for item in calls
if item[0][:2] == [str(venv_pip), "install"] and "-r" in item[0]
][0]
self.assertEqual(project_install[1]["PIP_INDEX_URL"], "https://user:pass@mirror.example/simple")
self.assertEqual(project_install[1]["UV_DEFAULT_INDEX"], "https://user:pass@mirror.example/simple")
self.assertNotIn("user:pass", " ".join(project_install[2] or project_install[0]))

123
tests/test_package_installer.py Normal file
View File

@@ -0,0 +1,123 @@
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
from app.helper.package_installer import (
PackageInstallRequest,
build_package_install_env,
build_package_install_strategies,
redact_url,
)
def test_build_env_maps_proxy_and_cache(tmp_path, monkeypatch):
monkeypatch.delenv("PIP_CACHE_DIR", raising=False)
monkeypatch.delenv("UV_CACHE_DIR", raising=False)
monkeypatch.delenv("PACKAGE_CACHE_ROOT", raising=False)
monkeypatch.setenv("HTTP_PROXY", "http://old.example:8080")
request = PackageInstallRequest(
requirements_file=tmp_path / "requirements.txt",
python_bin=Path("/venv/bin/python"),
config_dir=tmp_path / "config",
pip_index_url="https://user:pass@mirror.example/simple",
proxy_url="http://proxy.example:7890",
)
env = build_package_install_env(request)
assert env["HTTP_PROXY"] == "http://proxy.example:7890"
assert env["HTTPS_PROXY"] == "http://proxy.example:7890"
assert env["http_proxy"] == "http://proxy.example:7890"
assert env["https_proxy"] == "http://proxy.example:7890"
assert env["PACKAGE_CACHE_ROOT"] == str(tmp_path / "config" / ".cache")
assert env["PIP_CACHE_DIR"] == str(tmp_path / "config" / ".cache" / "pip")
assert env["UV_CACHE_DIR"] == str(tmp_path / "config" / ".cache" / "uv")
def test_build_env_uses_package_cache_root_and_preserves_tool_cache_overrides(tmp_path, monkeypatch):
monkeypatch.setenv("PACKAGE_CACHE_ROOT", str(tmp_path / "custom-package-cache"))
monkeypatch.setenv("PIP_CACHE_DIR", "/custom/pip")
monkeypatch.delenv("UV_CACHE_DIR", raising=False)
request = PackageInstallRequest(
requirements_file=tmp_path / "requirements.txt",
python_bin=Path("/venv/bin/python"),
config_dir=tmp_path / "config",
)
env = build_package_install_env(request)
assert env["PACKAGE_CACHE_ROOT"] == str(tmp_path / "custom-package-cache")
assert env["PIP_CACHE_DIR"] == "/custom/pip"
assert env["UV_CACHE_DIR"] == str(tmp_path / "custom-package-cache" / "uv")
def test_build_strategies_prefers_uv_network_matrix_and_preserves_find_links(tmp_path):
req = tmp_path / "requirements.txt"
req.write_text("demo\n", encoding="utf-8")
wheels = tmp_path / "wheels"
wheels.mkdir()
uv_bin = tmp_path / "venv" / "bin" / "uv"
uv_bin.parent.mkdir(parents=True)
uv_bin.write_text("", encoding="utf-8")
request = PackageInstallRequest(
requirements_file=req,
python_bin=tmp_path / "venv" / "bin" / "python",
find_links_dirs=[wheels],
config_dir=tmp_path / "config",
pip_index_url="https://mirror.example/simple",
proxy_url="http://proxy.example:7890",
)
strategies = build_package_install_strategies(request)
assert [strategy.strategy_name for strategy in strategies] == [
"uv:镜像+代理",
"uv:镜像",
"uv:代理",
"uv:直连",
"pip:镜像+代理",
"pip:镜像",
"pip:代理",
"pip:直连",
]
assert strategies[0].command[:3] == [str(uv_bin), "pip", "install"]
assert "--python" in strategies[0].command
assert "--find-links" in strategies[0].command
assert "--default-index" in strategies[0].command
assert "--no-index" not in strategies[0].command
assert strategies[0].env["HTTPS_PROXY"] == "http://proxy.example:7890"
assert "--default-index" in strategies[1].command
assert "HTTPS_PROXY" not in {
key for key, value in strategies[1].env.items() if value == "http://proxy.example:7890"
}
assert "--default-index" not in strategies[2].command
assert strategies[4].backend == "pip"
assert "-i" in strategies[4].command
def test_build_strategies_uses_pip_only_when_uv_missing(tmp_path):
req = tmp_path / "requirements.txt"
req.write_text("demo\n", encoding="utf-8")
request = PackageInstallRequest(
requirements_file=req,
python_bin=tmp_path / "venv" / "bin" / "python",
config_dir=tmp_path / "config",
)
with patch("app.helper.package_installer._find_uv", return_value=None):
strategies = build_package_install_strategies(request)
assert [strategy.strategy_name for strategy in strategies] == ["pip:直连"]
def test_redact_url_removes_userinfo():
assert redact_url("https://user:pass@mirror.example/simple") == "https://mirror.example/simple"
def test_redact_url_removes_userinfo_with_invalid_port():
assert (
redact_url("https://user:pass@example.com:notaport/simple")
== "https://example.com:notaport/simple"
)

195
tests/test_plugin_helper.py
View File

@@ -625,7 +625,7 @@ class TestPluginHelper:
module_names = ["app.plugins.dynamicwechat.helper", "Crypto.Cipher._mode_cbc"]
def fake_execute(_cmd):
def fake_execute(_cmd, env=None, safe_command=None):
for module_name in module_names:
sys.modules[module_name] = ModuleType(module_name)
return True, "ok"
@@ -644,6 +644,46 @@ class TestPluginHelper:
for module_name in module_names:
assert module_name in sys.modules
def test_pip_install_builds_uv_strategy_without_proxy_argument(self):
"""
插件依赖安装优先使用 uv 时，传输代理只进入子进程环境。
"""
try:
from app.helper.plugin import PluginHelper
except ModuleNotFoundError as exc:
pytest.skip(f"missing dependency: {exc}")
seen = []
def fake_execute(command, env=None, safe_command=None):
seen.append((command, env, safe_command))
return True, "ok"
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
req = root / "requirements.txt"
req.write_text("demo\n", encoding="utf-8")
uv_bin = root / "venv" / "bin" / "uv"
uv_bin.parent.mkdir(parents=True)
uv_bin.write_text("", encoding="utf-8")
with patch("app.helper.package_installer._find_uv", return_value=uv_bin), \
patch.object(PluginHelper, "_PluginHelper__get_protected_runtime_packages", return_value={}), \
patch.object(PluginHelper, "_PluginHelper__run_runtime_healthcheck", return_value=(True, "")), \
patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute), \
patch("app.helper.plugin.settings.PROXY_HOST", "http://proxy.example:7890"), \
patch("app.helper.plugin.settings.PIP_PROXY", "https://user:pass@mirror.example/simple"):
success, message = PluginHelper.pip_install_with_fallback(req)
assert success
assert message == "ok"
assert seen
command, env, safe_command = seen[0]
assert command[:3] == [str(uv_bin), "pip", "install"]
assert "--proxy" not in command
assert env["HTTPS_PROXY"] == "http://proxy.example:7890"
assert "user:pass" not in " ".join(safe_command)
def test_pip_install_serializes_concurrent_calls(self):
"""
验证多个依赖安装请求会复用同一把锁串行执行 pip。
@@ -660,7 +700,7 @@ class TestPluginHelper:
start_event = threading.Event()
errors = []
def fake_execute(_cmd):
def fake_execute(_cmd, env=None, safe_command=None):
nonlocal active_installs, max_active_installs
with state_lock:
active_installs += 1
@@ -769,7 +809,7 @@ class TestPluginHelper:
seen_install_commands = []
def fake_execute(cmd):
def fake_execute(cmd, env=None, safe_command=None):
if cmd[:4] == [sys.executable, "-m", "pip", "install"]:
seen_install_commands.append(cmd)
assert "-c" not in cmd
@@ -790,7 +830,8 @@ class TestPluginHelper:
return_value={}
):
with patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
with patch("app.helper.package_installer._find_uv", return_value=None):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
assert success
assert "ok" == message
@@ -807,7 +848,7 @@ class TestPluginHelper:
seen_constraints = []
def fake_execute(cmd):
def fake_execute(cmd, env=None, safe_command=None):
if cmd[:4] == [sys.executable, "-m", "pip", "install"]:
constraint_index = cmd.index("-c") + 1
constraint_file = Path(cmd[constraint_index])
@@ -826,7 +867,8 @@ class TestPluginHelper:
return_value={"fastapi": Version("0.115.14")}
):
with patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
with patch("app.helper.package_installer._find_uv", return_value=None):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
assert success
assert "ok" == message
@@ -846,7 +888,7 @@ class TestPluginHelper:
healthcheck_failed = False
pip_check_cmd = PluginHelper._PluginHelper__build_runtime_pip_command("check")
def fake_execute(cmd):
def fake_execute(cmd, env=None, safe_command=None):
nonlocal healthcheck_failed
if cmd[:4] == [sys.executable, "-m", "pip", "install"]:
if "-c" not in cmd:
@@ -871,13 +913,150 @@ class TestPluginHelper:
return_value={"fastapi": Version("0.115.14")}
):
with patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
with patch("app.helper.package_installer._find_uv", return_value=None):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
assert not success
assert "已自动恢复主程序依赖" in message
assert 1 == len(repair_commands)
assert "runtime-constraints-" in repair_commands[0][-1]
def test_failed_install_repairs_runtime_before_returning_error(self):
"""
安装策略失败后如果主运行环境异常，应先恢复主程序依赖再返回失败。
"""
try:
from app.helper.plugin import PluginHelper
except ModuleNotFoundError as exc:
pytest.skip(f"missing dependency: {exc}")
repair_calls = []
def fake_execute(command, env=None, safe_command=None):
if "install" in command and "-r" in command and "plugin" in str(command[-1]):
return False, "partial failure"
return True, "ok"
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
req = root / "plugin-requirements.txt"
req.write_text("demo\n", encoding="utf-8")
with patch("app.helper.package_installer._find_uv", return_value=None), \
patch.object(PluginHelper, "_PluginHelper__get_protected_runtime_packages", return_value={}), \
patch.object(
PluginHelper,
"_PluginHelper__run_runtime_healthcheck",
side_effect=[(False, "broken"), (True, "")],
), \
patch.object(
PluginHelper,
"_PluginHelper__repair_main_runtime_dependencies",
side_effect=lambda snapshot_file=None: repair_calls.append(snapshot_file)
or (True, "runtime repaired"),
), \
patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
success, message = PluginHelper.pip_install_with_fallback(req)
assert not success
assert "partial failure" in message or "恢复" in message
assert repair_calls
def test_failed_strategy_stops_after_runtime_repair_even_if_later_strategy_could_succeed(self):
"""
一旦失败策略污染主运行环境并触发恢复，不能继续 fallback 后把安装结果伪装成成功。
"""
try:
from app.helper.plugin import PluginHelper
except ModuleNotFoundError as exc:
pytest.skip(f"missing dependency: {exc}")
seen_install_commands = []
repair_calls = []
def fake_execute(command, env=None, safe_command=None):
if "install" in command and "-r" in command:
seen_install_commands.append(command)
if len(seen_install_commands) == 1:
return False, "resolver failed"
return True, "later success"
return True, "ok"
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
req = root / "requirements.txt"
req.write_text("demo\n", encoding="utf-8")
uv_bin = root / "venv" / "bin" / "uv"
uv_bin.parent.mkdir(parents=True)
uv_bin.write_text("", encoding="utf-8")
with patch("app.helper.package_installer._find_uv", return_value=uv_bin), \
patch.object(PluginHelper, "_PluginHelper__get_protected_runtime_packages", return_value={}), \
patch.object(
PluginHelper,
"_PluginHelper__run_runtime_healthcheck",
side_effect=[(False, "broken"), (True, "")],
), \
patch.object(
PluginHelper,
"_PluginHelper__repair_main_runtime_dependencies",
side_effect=lambda snapshot_file=None: repair_calls.append(snapshot_file)
or (True, "runtime repaired"),
), \
patch("app.helper.plugin.settings.PIP_PROXY", "https://mirror.example/simple"), \
patch("app.helper.plugin.settings.PROXY_HOST", "http://proxy.example:7890"), \
patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
success, message = PluginHelper.pip_install_with_fallback(req)
assert not success
assert "resolver failed" in message
assert "主运行环境已恢复" in message
assert len(seen_install_commands) == 1
assert repair_calls
def test_repair_main_runtime_dependencies_uses_package_installer_semantics(self):
"""
主运行环境恢复与插件安装使用同一套 cache、index、proxy 和安全日志语义。
"""
try:
from app.helper.plugin import PluginHelper
except ModuleNotFoundError as exc:
pytest.skip(f"missing dependency: {exc}")
seen = []
def fake_execute(command, env=None, safe_command=None):
seen.append((command, env, safe_command))
return True, "ok"
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
req = root / "requirements.txt"
req.write_text("fastapi==1.0\n", encoding="utf-8")
uv_bin = root / "venv" / "bin" / "uv"
uv_bin.parent.mkdir(parents=True)
uv_bin.write_text("", encoding="utf-8")
with patch("app.helper.package_installer._find_uv", return_value=uv_bin), \
patch("app.helper.plugin.settings.CONFIG_DIR", str(root / "config")), \
patch("app.helper.plugin.settings.PACKAGE_CACHE_ROOT", str(root / "custom-package-cache")), \
patch("app.helper.plugin.settings.PIP_PROXY", "https://user:pass@mirror.example/simple"), \
patch("app.helper.plugin.settings.PROXY_HOST", "http://proxy.example:7890"), \
patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
success, message = PluginHelper._PluginHelper__repair_main_runtime_dependencies(req)
assert success
assert message == "ok"
assert seen
command, env, safe_command = seen[0]
assert command[:3] == [str(uv_bin), "pip", "install"]
assert "--proxy" not in command
assert env["PACKAGE_CACHE_ROOT"] == str(root / "custom-package-cache")
assert env["PIP_CACHE_DIR"] == str(root / "custom-package-cache" / "pip")
assert env["UV_CACHE_DIR"] == str(root / "custom-package-cache" / "uv")
assert env["HTTPS_PROXY"] == "http://proxy.example:7890"
assert "user:pass" not in " ".join(safe_command)
def test_async_pip_install_runs_in_threadpool(self):
"""
验证异步安装路径会把同步 pip 安装派发到线程池，避免阻塞事件循环。

98
tests/test_system_utils.py
View File

@@ -74,3 +74,101 @@ class SystemHelperRestartTest(TestCase):
finally:
SystemHelper._SystemHelper__docker_restart_intent_file = original_intent_file
settings.CONFIG_DIR = original_config_dir
def test_execute_with_subprocess_passes_env_to_subprocess():
with patch("app.utils.system.subprocess.run") as run_mock:
run_mock.return_value.stdout = "ok"
run_mock.return_value.stderr = ""
success, message = SystemUtils.execute_with_subprocess(
["pip", "check"],
env={"PIP_CACHE_DIR": "/config/.cache/pip"},
)
assert success
assert message == "ok"
assert run_mock.call_args.kwargs["env"]["PIP_CACHE_DIR"] == "/config/.cache/pip"
def test_execute_with_subprocess_uses_safe_command_in_failure_message():
error = subprocess.CalledProcessError(
returncode=1,
cmd=["pip", "install", "-i", "https://user:pass@mirror.example/simple"],
output="",
stderr="failed",
)
command = ["pip", "install", "-i", "https://user:pass@mirror.example/simple"]
with patch("app.utils.system.subprocess.run", side_effect=error) as run_mock:
success, message = SystemUtils.execute_with_subprocess(
command,
safe_command=["pip", "install", "-i", "https://mirror.example/simple"],
)
assert not success
assert "https://mirror.example/simple" in message
assert "user:pass" not in message
assert run_mock.call_args.args[0] == command
def test_execute_with_subprocess_redacts_userinfo_from_stdout_and_stderr():
error = subprocess.CalledProcessError(
returncode=1,
cmd=["pip", "install"],
output="Looking in indexes: https://user:pass@mirror.example/simple",
stderr="Proxy failed: http://proxy_user:proxy_pass@proxy.example:7890",
)
with patch("app.utils.system.subprocess.run", side_effect=error):
success, message = SystemUtils.execute_with_subprocess(["pip", "install"])
assert not success
assert "https://mirror.example/simple" in message
assert "http://proxy.example:7890" in message
assert "user:pass" not in message
assert "proxy_user:proxy_pass" not in message
def test_execute_with_subprocess_redacts_userinfo_from_non_http_scheme():
error = subprocess.CalledProcessError(
returncode=1,
cmd=["pip", "install"],
output="Proxy failed: socks5://proxy_user:proxy_pass@proxy.example:7890",
stderr="Resolved direct URL: git+https://git_user:git_pass@example.com/org/repo.git",
)
with patch("app.utils.system.subprocess.run", side_effect=error):
success, message = SystemUtils.execute_with_subprocess(["pip", "install"])
assert not success
assert "socks5://proxy.example:7890" in message
assert "git+https://example.com/org/repo.git" in message
assert "proxy_user:proxy_pass" not in message
assert "git_user:git_pass" not in message
def test_execute_with_subprocess_redacts_success_output_userinfo():
with patch("app.utils.system.subprocess.run") as run_mock:
run_mock.return_value.stdout = "Using https://user:pass@mirror.example/simple\n"
run_mock.return_value.stderr = "Proxy socks5://proxy_user:proxy_pass@proxy.example:7890\n"
success, message = SystemUtils.execute_with_subprocess(["pip", "install"])
assert success
assert "https://mirror.example/simple" in message
assert "socks5://proxy.example:7890" in message
assert "user:pass" not in message
assert "proxy_user:proxy_pass" not in message
def test_execute_with_subprocess_redacts_unknown_error_userinfo_and_invalid_port():
with patch(
"app.utils.system.subprocess.run",
side_effect=RuntimeError("bad url https://user:pass@example.com:notaport/simple"),
):
success, message = SystemUtils.execute_with_subprocess(["pip", "install"])
assert not success
assert "https://example.com:notaport/simple" in message
assert "user:pass" not in message

56
tests/test_uv_pip_compat.py
View File

@@ -12,6 +12,62 @@ ROOT = Path(__file__).resolve().parents[1]
WRAPPER = ROOT / "scripts" / "uv-pip-compat.sh"
def run_wrapper_with_env(link_name: str, *args: str) -> tuple[list[str], dict[str, str]]:
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
venv_bin = temp_path / "venv" / "bin"
venv_bin.mkdir(parents=True)
(venv_bin / "python").write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
(venv_bin / "python").chmod(0o755)
argv_file = temp_path / "argv.txt"
env_file = temp_path / "env.txt"
uv_bin = venv_bin / "uv"
uv_bin.write_text(
"#!/bin/sh\n"
f"for arg in \"$@\"; do printf '%s\\n' \"$arg\" >> '{argv_file}'; done\n"
"for name in HTTP_PROXY HTTPS_PROXY http_proxy https_proxy; do\n"
" eval \"value=\\${$name:-}\"\n"
f" printf '%s=%s\\n' \"$name\" \"$value\" >> '{env_file}'\n"
"done\n",
encoding="utf-8",
)
uv_bin.chmod(0o755)
wrapper_path = venv_bin / "uv-pip-compat"
shutil.copy2(WRAPPER, wrapper_path)
wrapper_path.chmod(0o755)
link_path = venv_bin / link_name
link_path.symlink_to(wrapper_path.name)
subprocess.run(
[str(link_path), *args],
check=True,
env={
**os.environ,
"PATH": f"{venv_bin}{os.pathsep}{os.environ.get('PATH', '')}",
},
)
env_lines = dict(line.split("=", 1) for line in env_file.read_text(encoding="utf-8").splitlines())
return argv_file.read_text(encoding="utf-8").splitlines(), env_lines
def test_pip_install_converts_proxy_argument_to_env():
argv, env_lines = run_wrapper_with_env("pip", "install", "--proxy", "http://proxy.example:7890", "demo")
assert "--proxy" not in argv
assert "http://proxy.example:7890" not in argv
assert env_lines["HTTPS_PROXY"] == "http://proxy.example:7890"
assert env_lines["HTTP_PROXY"] == "http://proxy.example:7890"
def test_pip_install_converts_proxy_equals_argument_to_env():
argv, env_lines = run_wrapper_with_env("pip", "install", "--proxy=http://proxy.example:7890", "demo")
assert "--proxy=http://proxy.example:7890" not in argv
assert env_lines["https_proxy"] == "http://proxy.example:7890"
class UvPipCompatTests(unittest.TestCase):
def run_wrapper(self, link_name: str, *args: str) -> list[str]:
with tempfile.TemporaryDirectory() as temp_dir: