GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-09-05 07:27:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 使用 uv 锁定主程序依赖并强化插件恢复边界 (#6364)

Browse Source
This commit is contained in:
InfinityPacer
2026-08-20 12:17:19 +08:00
committed by GitHub
parent 27ae1b5290
commit 23f5d59c74
59 changed files with 6804 additions and 1797 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+2
View File
@@ -148,6 +148,7 @@ def configure_plugin_system_services():
)
from app.adapters.external.plugin.client import PluginMarketClient
from app.adapters.system.plugin.dependency import PluginDependencyInstaller
from app.adapters.system.plugin.manifest import dependency_manifest_status
from app.adapters.system.plugin.package import PluginPackageManager
from app.runtime.extensions.plugin.system import (
PluginSystemServices,
@@ -160,6 +161,7 @@ def configure_plugin_system_services():
market=PluginMarketClient(helper),
package=PluginPackageManager(helper),
dependency=PluginDependencyInstaller(helper),
dependency_manifest_status=dependency_manifest_status,
compatible_flags=lambda flag: (
[flag] + VERSION_BACKWARD_COMPATIBLE_FLAGS.get(flag, [])
if flag else []
tests/fixtures/architecture/dependency-baseline.json
Vendored
+12 -3
View File
@@ -13,8 +13,8 @@
"runtime_to_db": [],
"workflow_to_db": []
},
"edge_count": 6061,
"edge_sha256": "022ba984b711776539b4dc120d0f2e92645f61b28b59dd6be3c32a64499d08ce",
"edge_count": 6069,
"edge_sha256": "a47f8e0ee6d4855e106fefae05d5fa27beea9a1fd5710009260615d232af3c03",
"edges": [
"app -> app.runtime",
"app -> app.runtime.compat",
@@ -52,6 +52,8 @@
"app.adapters.external.market -> app.adapters.system",
"app.adapters.external.market -> app.adapters.system.host",
"app.adapters.external.market -> app.adapters.system.package",
"app.adapters.external.market -> app.adapters.system.plugin",
"app.adapters.external.market -> app.adapters.system.plugin.manifest",
"app.adapters.external.market -> app.foundation",
"app.adapters.external.market -> app.foundation.singleton",
"app.adapters.external.market -> app.foundation.url",
@@ -122,9 +124,14 @@
"app.adapters.system.plugin.dependency -> app.adapters",
"app.adapters.system.plugin.dependency -> app.adapters.external",
"app.adapters.system.plugin.dependency -> app.adapters.external.market",
"app.adapters.system.plugin.dependency -> app.adapters.system",
"app.adapters.system.plugin.dependency -> app.adapters.system.plugin",
"app.adapters.system.plugin.dependency -> app.adapters.system.plugin.manifest",
"app.adapters.system.plugin.dependency -> app.runtime",
"app.adapters.system.plugin.dependency -> app.runtime.config",
"app.adapters.system.plugin.dependency -> app.runtime.log",
"app.adapters.system.plugin.manifest -> app.runtime",
"app.adapters.system.plugin.manifest -> app.runtime.log",
"app.adapters.system.plugin.package -> app.adapters",
"app.adapters.system.plugin.package -> app.adapters.external",
"app.adapters.system.plugin.package -> app.adapters.external.market",
@@ -5862,6 +5869,7 @@
"app.startup.plugins_initializer -> app.adapters.system.host",
"app.startup.plugins_initializer -> app.adapters.system.plugin",
"app.startup.plugins_initializer -> app.adapters.system.plugin.dependency",
"app.startup.plugins_initializer -> app.adapters.system.plugin.manifest",
"app.startup.plugins_initializer -> app.adapters.system.plugin.package",
"app.startup.plugins_initializer -> app.api",
"app.startup.plugins_initializer -> app.api.endpoints",
@@ -6078,7 +6086,7 @@
"app.workflow.actions.transfer_file -> app.workflow",
"app.workflow.actions.transfer_file -> app.workflow.actions"
],
"module_count": 752,
"module_count": 753,
"modules": [
"app",
"app.adapters",
@@ -6112,6 +6120,7 @@
"app.adapters.system.package",
"app.adapters.system.plugin",
"app.adapters.system.plugin.dependency",
"app.adapters.system.plugin.manifest",
"app.adapters.system.plugin.package",
"app.adapters.system.resource",
"app.adapters.system.rust",
tests/test_cli_auto_update.py
+2 -4
View File
@@ -119,7 +119,7 @@ class CliAutoUpdateTests(unittest.TestCase):
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(
with patch.dict(module.os.environ, {"HTTPS_PROXY": "http://old.example:8080"}, clear=True), 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
@@ -132,7 +132,6 @@ class CliAutoUpdateTests(unittest.TestCase):
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):
@@ -145,7 +144,7 @@ class CliAutoUpdateTests(unittest.TestCase):
{
"PACKAGE_CACHE_ROOT": str(package_cache_root),
},
clear=False,
clear=True,
), 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(
@@ -155,5 +154,4 @@ class CliAutoUpdateTests(unittest.TestCase):
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"))
tests/test_docker_bootstrap.py
+145 -103
View File
@@ -27,6 +27,20 @@ def _write_bundle(path: Path, label: str, *, extra_files: tuple[str, ...] = ())
def test_dockerfile_control_bundle_build_checks_fail_closed() -> None:
dockerfile = (ROOT / "docker" / "Dockerfile").read_text(encoding="utf-8")
assert (
"FROM ghcr.io/astral-sh/uv:0.12.5@sha256:"
"e85be844203885286c60ffad8a858d48afb6c5a5c237ca0e67f12e74b8f174b1 AS uv"
in dockerfile
)
assert "COPY --from=uv /uv /usr/local/bin/uv" in dockerfile
assert "COPY pyproject.toml uv.lock ./" in dockerfile
assert "python3 -m venv --without-pip ${VENV_PATH}" in dockerfile
assert "UV_PROJECT_ENVIRONMENT=${VENV_PATH} uv sync" in dockerfile
for option in ("--locked", "--no-dev", "--no-install-project"):
assert option in dockerfile
assert "uv-pip-compat" not in dockerfile
assert "requirements.in" not in dockerfile
assert "${VENV_PATH}/bin/pip" not in dockerfile
assert "-exec cp -f -t /usr/local/lib/moviepilot/control {} +" in dockerfile
assert "bash -n /entrypoint.sh" in dockerfile
assert 'ENTRYPOINT [ "/usr/bin/tini", "-g", "--", "/entrypoint.sh" ]' in dockerfile
@@ -540,7 +554,7 @@ def test_updater_package_proxy_stays_command_scoped(tmp_path: Path) -> None:
source {UPDATER!s}
set_package_proxy_env
printf '%s|%s|%s|%s\\n' "${{HTTP_PROXY-unset}}" "${{HTTPS_PROXY-unset}}" "${{http_proxy-unset}}" "${{https_proxy-unset}}"
printf '%s\\n' "${{PIP_ENV[*]}}"
printf '%s\\n' "${{PACKAGE_ENV[*]}}"
"""
)
env = dict(os.environ)
@@ -578,7 +592,7 @@ def test_updater_exposes_explicit_result(
INFO() {{ :; }}
WARN() {{ :; }}
ERROR() {{ :; }}
test_connectivity_pip() {{ PIP_LOG=test; return 0; }}
test_connectivity_package() {{ PACKAGE_LOG=test; return 0; }}
test_connectivity_github() {{ GITHUB_LOG=test; return 0; }}
install_backend_and_download_resources() {{
if [ "${{INSTALL_RESULT}}" = success ]; then
@@ -605,7 +619,7 @@ def test_updater_exposes_explicit_result(
def test_release_noop_preserves_prerelease_selection_without_probing_package_index(
tmp_path: Path,
) -> None:
pip_probe = tmp_path / "pip-probe"
package_probe = tmp_path / "package-probe"
curl_log = tmp_path / "curl.log"
comparison_log = tmp_path / "comparison.log"
script = textwrap.dedent(
@@ -613,14 +627,14 @@ def test_release_noop_preserves_prerelease_selection_without_probing_package_ind
CONFIG_DIR="$1"
MOVIEPILOT_AUTO_UPDATE=release
PIP_PROXY= PROXY_HOST= GITHUB_PROXY= GITHUB_TOKEN=
PIP_PROBE="$2"
PACKAGE_PROBE="$2"
CURL_LOG="$3"
COMPARISON_LOG="$4"
source {UPDATER!s}
INFO() {{ :; }}
WARN() {{ :; }}
ERROR() {{ :; }}
test_connectivity_pip() {{ touch "${{PIP_PROBE}}"; return 0; }}
test_connectivity_package() {{ touch "${{PACKAGE_PROBE}}"; return 0; }}
test_connectivity_github() {{ CURL_OPTIONS=-sL; GITHUB_LOG=test; return 0; }}
compare_versions() {{ printf '%s|%s\n' "$1" "$2" > "${{COMPARISON_LOG}}"; return 1; }}
grep() {{
@@ -653,7 +667,7 @@ def test_release_noop_preserves_prerelease_selection_without_probing_package_ind
script,
"release-noop-test",
str(tmp_path / "config"),
str(pip_probe),
str(package_probe),
str(curl_log),
str(comparison_log),
],
@@ -663,7 +677,7 @@ def test_release_noop_preserves_prerelease_selection_without_probing_package_ind
)
assert result.stdout == "noop\n"
assert not pip_probe.exists()
assert not package_probe.exists()
curl_args = curl_log.read_text(encoding="utf-8")
assert "/releases" in curl_args
assert "/releases/latest" not in curl_args
@@ -675,51 +689,68 @@ def test_release_noop_preserves_prerelease_selection_without_probing_package_ind
@pytest.mark.parametrize(
("dependencies_changed", "expected_route_calls", "expected_install_calls"),
((False, 0, 0), (True, 1, 1)),
("pyproject_changed", "lock_changed", "expected_route_calls", "expected_sync_calls"),
(
(False, False, 0, 0),
(True, False, 1, 1),
(False, True, 1, 1),
(True, True, 1, 1),
),
)
def test_package_route_is_only_configured_for_changed_dependencies(
tmp_path: Path,
dependencies_changed: bool,
pyproject_changed: bool,
lock_changed: bool,
expected_route_calls: int,
expected_install_calls: int,
expected_sync_calls: int,
) -> None:
venv_bin = tmp_path / "venv" / "bin"
venv_bin.mkdir(parents=True)
pip_log = tmp_path / "pip.log"
uv_bin = tmp_path / "bin" / "uv"
uv_bin.parent.mkdir(parents=True)
uv_log = tmp_path / "uv.log"
route_log = tmp_path / "route.log"
for executable in ("pip", "pip-compile"):
path = venv_bin / executable
path.write_text(
"#!/bin/bash\nprintf '%s\\n' \"$*\" >> \"${PIP_TEST_LOG}\"\n",
encoding="utf-8",
)
path.chmod(0o755)
uv_bin.write_text(
"#!/bin/bash\n"
"printf '%s|%s\\n' \"${UV_PROJECT_ENVIRONMENT:-}\" \"$*\" >> \"${UV_TEST_LOG}\"\n",
encoding="utf-8",
)
uv_bin.chmod(0o755)
update_tree = tmp_path / "update" / "App"
update_tree.mkdir(parents=True)
(update_tree / "requirements.in").write_text("new-package==1\n", encoding="utf-8")
(update_tree / "version.py").write_text("FRONTEND_VERSION = ''\n", encoding="utf-8")
(update_tree / "pyproject.toml").write_text("[project]\n", encoding="utf-8")
(update_tree / "uv.lock").write_text("version = 1\n", encoding="utf-8")
script = textwrap.dedent(
f"""\
CONFIG_DIR="$1"
VENV_PATH="$2"
TMP_PATH="$3"
ROUTE_LOG="$4"
DEPENDENCIES_CHANGED="$5"
PYPROJECT_CHANGED="$5"
LOCK_CHANGED="$6"
UV_BIN="$7"
PIP_PROXY= PROXY_HOST=
source {UPDATER!s}
INFO() {{ :; }}
WARN() {{ :; }}
ERROR() {{ :; }}
download_and_unzip() {{ return 0; }}
cmp() {{ [ "${{DEPENDENCIES_CHANGED}}" = false ]; }}
cp() {{ return 0; }}
configure_pip_route() {{ printf 'route\n' >> "${{ROUTE_LOG}}"; PIP_LOG=test; }}
install_backend_and_download_resources tags/v3.0.1.zip || true
cmp() {{
case "$2" in
*/pyproject.toml) [ "${{PYPROJECT_CHANGED}}" = false ] ;;
*/uv.lock) [ "${{LOCK_CHANGED}}" = false ] ;;
esac
}}
configure_package_route() {{
printf 'route\n' >> "${{ROUTE_LOG}}"
PACKAGE_LOG=test
PACKAGE_ENV=()
UV_OPTIONS=()
}}
if dependency_manifests_changed; then
sync_project_dependencies
fi
"""
)
env = {**os.environ, "PIP_TEST_LOG": str(pip_log)}
env = {**os.environ, "UV_TEST_LOG": str(uv_log)}
result = subprocess.run(
[
"bash",
@@ -730,7 +761,9 @@ def test_package_route_is_only_configured_for_changed_dependencies(
str(tmp_path / "venv"),
str(tmp_path / "update"),
str(route_log),
str(dependencies_changed).lower(),
str(pyproject_changed).lower(),
str(lock_changed).lower(),
str(uv_bin),
],
text=True,
capture_output=True,
@@ -740,72 +773,98 @@ def test_package_route_is_only_configured_for_changed_dependencies(
assert result.stderr == ""
route_calls = route_log.read_text(encoding="utf-8").splitlines() if route_log.exists() else []
install_calls = pip_log.read_text(encoding="utf-8").splitlines() if pip_log.exists() else []
sync_calls = uv_log.read_text(encoding="utf-8").splitlines() if uv_log.exists() else []
assert len(route_calls) == expected_route_calls
compile_calls = [call for call in install_calls if not call.startswith("install ")]
package_install_calls = [call for call in install_calls if call.startswith("install ")]
assert len(package_install_calls) == expected_install_calls
if dependencies_changed:
assert compile_calls == [
f"{update_tree / 'requirements.in'} -o {tmp_path / 'update' / 'requirements.txt'}"
]
assert package_install_calls == [
f"install -r {tmp_path / 'update' / 'requirements.txt'}"
assert len(sync_calls) == expected_sync_calls
if expected_sync_calls:
assert sync_calls == [
f"{tmp_path / 'venv'}|sync --project {update_tree} "
f"--locked --inexact --no-dev --no-install-project "
f"--python {tmp_path / 'venv' / 'bin' / 'python3'}"
]
@pytest.mark.parametrize("failure", ("compile", "install", "post_install"))
def test_failed_dependency_update_does_not_overwrite_current_manifests(
@pytest.mark.parametrize("missing_manifest", ("pyproject.toml", "uv.lock"))
def test_dependency_update_requires_complete_uv_manifests(
tmp_path: Path,
failure: str,
missing_manifest: str,
) -> None:
venv_bin = tmp_path / "venv" / "bin"
venv_bin.mkdir(parents=True)
command_log = tmp_path / "commands.log"
copy_log = tmp_path / "copies.log"
for executable in ("pip", "pip-compile"):
path = venv_bin / executable
path.write_text(
"#!/bin/bash\n"
'printf \'%s|%s\\n\' "$(basename "$0")" "$*" >> "${COMMAND_LOG}"\n'
f'[[ "$(basename "$0")" == "pip-compile" && "{failure}" == "compile" ]] && exit 1\n'
f'[[ "$(basename "$0")" == "pip" && "{failure}" == "install" ]] && exit 1\n'
"exit 0\n",
encoding="utf-8",
)
path.chmod(0o755)
update_tree = tmp_path / "update" / "App"
update_tree.mkdir(parents=True)
(update_tree / "requirements.in").write_text("new-package==1\n", encoding="utf-8")
(update_tree / "version.py").write_text(
"FRONTEND_VERSION = 'v3.0.0'\n",
(update_tree / "pyproject.toml").write_text("[project]\n", encoding="utf-8")
(update_tree / "uv.lock").write_text("version = 1\n", encoding="utf-8")
(update_tree / missing_manifest).unlink()
route_marker = tmp_path / "route-called"
copy_marker = tmp_path / "copy-called"
script = textwrap.dedent(
f"""\
CONFIG_DIR="$1"
TMP_PATH="$2"
ROUTE_MARKER="$3"
COPY_MARKER="$4"
PIP_PROXY= PROXY_HOST=
source {UPDATER!s}
INFO() {{ :; }}
WARN() {{ :; }}
ERROR() {{ :; }}
download_and_unzip() {{ return 0; }}
configure_package_route() {{ touch "${{ROUTE_MARKER}}"; }}
cp() {{ touch "${{COPY_MARKER}}"; }}
! install_backend_and_download_resources tags/v3.0.1.zip
"""
)
subprocess.run(
[
"bash",
"-c",
script,
"incomplete-manifest-test",
str(tmp_path / "config"),
str(tmp_path / "update"),
str(route_marker),
str(copy_marker),
],
text=True,
capture_output=True,
check=True,
)
assert not route_marker.exists()
assert not copy_marker.exists()
def test_failed_dependency_sync_does_not_replace_program_files(tmp_path: Path) -> None:
uv_bin = tmp_path / "bin" / "uv"
uv_bin.parent.mkdir(parents=True)
uv_bin.write_text(
"#!/bin/bash\nprintf '%s\\n' \"$*\" >> \"${UV_LOG}\"\nexit 1\n",
encoding="utf-8",
)
uv_bin.chmod(0o755)
update_tree = tmp_path / "update" / "App"
update_tree.mkdir(parents=True)
(update_tree / "pyproject.toml").write_text("[project]\n", encoding="utf-8")
(update_tree / "uv.lock").write_text("version = 1\n", encoding="utf-8")
copy_log = tmp_path / "copies.log"
uv_log = tmp_path / "uv.log"
script = textwrap.dedent(
f"""\
CONFIG_DIR="$1"
VENV_PATH="$2"
TMP_PATH="$3"
COPY_LOG="$4"
UV_BIN="$4"
COPY_LOG="$5"
PIP_PROXY= PROXY_HOST=
FAILURE={failure}
source {UPDATER!s}
INFO() {{ :; }}
WARN() {{ :; }}
ERROR() {{ :; }}
download_and_unzip() {{
if [[ "${{FAILURE}}" = post_install ]] && [[ "$2" = dist ]]; then
return 1
fi
return 0
}}
download_and_unzip() {{ return 0; }}
cmp() {{ return 1; }}
cp() {{ printf '%s\n' "$*" >> "${{COPY_LOG}}"; }}
configure_pip_route() {{ PIP_LOG=test; }}
configure_package_route() {{ PACKAGE_LOG=test; PACKAGE_ENV=(); UV_OPTIONS=(); }}
install_backend_and_download_resources tags/v3.0.1.zip || true
if [[ "${{FAILURE}}" = post_install ]]; then
install_backend_and_download_resources tags/v3.0.1.zip || true
fi
"""
)
@@ -818,38 +877,20 @@ def test_failed_dependency_update_does_not_overwrite_current_manifests(
str(tmp_path / "config"),
str(tmp_path / "venv"),
str(tmp_path / "update"),
str(uv_bin),
str(copy_log),
],
text=True,
capture_output=True,
check=True,
env={**os.environ, "COMMAND_LOG": str(command_log)},
env={**os.environ, "UV_LOG": str(uv_log)},
)
assert not copy_log.exists()
commands = command_log.read_text(encoding="utf-8").splitlines()
assert commands[0] == (
f"pip-compile|{update_tree / 'requirements.in'} "
f"-o {tmp_path / 'update' / 'requirements.txt'}"
assert uv_log.read_text(encoding="utf-8") == (
f"sync --project {update_tree} --locked --inexact --no-dev "
f"--no-install-project --python {tmp_path / 'venv' / 'bin' / 'python3'}\n"
)
assert all("/app/requirements" not in command for command in commands)
if failure == "compile":
assert len(commands) == 1
elif failure == "install":
assert commands[1] == f"pip|install -r {tmp_path / 'update' / 'requirements.txt'}"
else:
assert commands == [
(
f"pip-compile|{update_tree / 'requirements.in'} "
f"-o {tmp_path / 'update' / 'requirements.txt'}"
),
f"pip|install -r {tmp_path / 'update' / 'requirements.txt'}",
(
f"pip-compile|{update_tree / 'requirements.in'} "
f"-o {tmp_path / 'update' / 'requirements.txt'}"
),
f"pip|install -r {tmp_path / 'update' / 'requirements.txt'}",
]
def test_package_index_probe_is_cacheless_and_bounded(tmp_path: Path) -> None:
@@ -859,11 +900,12 @@ def test_package_index_probe_is_cacheless_and_bounded(tmp_path: Path) -> None:
CONFIG_DIR="$1"
VENV_PATH="$2"
TIMEOUT_LOG="$3"
UV_BIN="$4"
PIP_PROXY=https://packages.example/simple
PROXY_HOST=
source {UPDATER!s}
timeout() {{ printf '%s\n' "$*" > "${{TIMEOUT_LOG}}"; return 124; }}
test_connectivity_pip 0 || true
test_connectivity_package 0 || true
"""
)
@@ -876,6 +918,7 @@ def test_package_index_probe_is_cacheless_and_bounded(tmp_path: Path) -> None:
str(tmp_path / "config"),
str(tmp_path / "venv"),
str(timeout_log),
"/fake/uv",
],
text=True,
capture_output=True,
@@ -885,13 +928,12 @@ def test_package_index_probe_is_cacheless_and_bounded(tmp_path: Path) -> None:
command = timeout_log.read_text(encoding="utf-8")
assert command.startswith("--kill-after=2s 10s env ")
assert "UV_NO_CACHE=1" in command
assert "PIP_NO_CACHE_DIR=1" in command
assert "UV_HTTP_TIMEOUT=5" in command
assert "PIP_DEFAULT_TIMEOUT=5" in command
assert "UV_HTTP_RETRIES=0" in command
assert "PIP_RETRIES=0" in command
assert "pip install --target " in command
assert " --no-deps -i https://packages.example/simple pip-hello-world" in command
assert "/fake/uv pip install --target " in command
assert (
" --no-deps --default-index https://packages.example/simple pip-hello-world" in command
)
assert "uninstall" not in command
probe_dir = Path(command.split("--target ", 1)[1].split(" ", 1)[0])
assert not probe_dir.exists()
tests/test_docker_entrypoint_permissions.py
+59
View File
@@ -32,6 +32,18 @@ def _write_fake_chown(tmp_path: Path) -> Path:
encoding="utf-8",
)
chown.chmod(0o755)
gosu = fake_bin / "gosu"
gosu.write_text(
textwrap.dedent(
"""\
#!/usr/bin/env bash
shift
exec "$@"
"""
),
encoding="utf-8",
)
gosu.chmod(0o755)
return fake_bin
@@ -78,6 +90,7 @@ def _run_permission_case(tmp_path: Path, body: str, env: dict[str, str] | None =
"PUID": str(os.getuid()),
"PGID": str(os.getgid()),
}
case_env.pop("UV_CACHE_DIR", None)
if env:
case_env.update(env)
@@ -428,6 +441,52 @@ def test_runtime_writable_paths_are_still_corrected(tmp_path: Path) -> None:
assert not any(f"{tmp_path}/public" in line for line in lines)
def test_external_package_cache_is_repaired_without_chowning_parent(
tmp_path: Path,
) -> None:
external_cache = tmp_path / "package-cache" / "uv"
log = _run_permission_case(
tmp_path,
"""
gosu() { shift; "$@"; }
UV_CACHE_DIR="${EXTERNAL_CACHE}" HOME="${HOME_DIR}" correct_file_permissions
""",
env={"EXTERNAL_CACHE": str(external_cache)},
)
assert f"-R moviepilot:moviepilot {external_cache}" in log.splitlines()
assert not any(
line.endswith(str(external_cache.parent)) for line in log.splitlines()
)
def test_external_package_cache_write_probe_failure_is_fatal(tmp_path: Path) -> None:
output = _run_entrypoint_case(
tmp_path,
"""
ERROR() { printf '%s\n' "$1"; }
chown() { :; }
gosu() { return 1; }
CONFIG_DIR="${CASE_CONFIG_DIR}"
VENV_PATH="${CASE_VENV_PATH}"
UV_CACHE_DIR="${CASE_CACHE_DIR}"
if correct_package_cache_permissions; then
printf 'unexpected-success\n'
else
printf 'rejected\n'
fi
""",
env={
"CASE_CONFIG_DIR": str(tmp_path / "config"),
"CASE_VENV_PATH": str(tmp_path / "venv"),
"CASE_CACHE_DIR": str(tmp_path / "external-cache"),
},
)
assert "uv 缓存目录不可写" in output
assert output.endswith("rejected\n")
def test_explicit_browser_cache_subtree_is_not_scanned_by_permission_repair(
tmp_path: Path,
) -> None:
tests/test_local_setup_config_dir.py
+174 -96
View File
@@ -59,20 +59,31 @@ class LocalSetupConfigDirTests(unittest.TestCase):
self.assertIsNone(result)
prompt_mock.assert_not_called()
def test_supported_python_accepts_versions_newer_than_current_ci(self):
module = load_local_setup_module()
with patch.object(module, "get_python_version", return_value=(3, 15, 0)):
module.ensure_supported_python("python3.15")
def test_supported_python_rejects_versions_below_3_12(self):
module = load_local_setup_module()
with patch.object(module, "get_python_version", return_value=(3, 11, 9)):
with self.assertRaisesRegex(RuntimeError, r"Python 3\.12\+"):
module.ensure_supported_python("python3.11")
def test_install_deps_installs_browser_runtime(self):
module = load_local_setup_module()
with tempfile.TemporaryDirectory() as temp_dir:
venv_dir = (Path(temp_dir) / "venv").resolve()
root = Path(temp_dir)
venv_dir = (root / "venv").resolve()
venv_python = venv_dir / "bin" / "python"
venv_pip = venv_dir / "bin" / "pip"
uv_bin = root / "tools" / "uv"
with patch.object(module, "ensure_supported_python"), \
patch.object(
module,
"configure_venv_pip_compat",
return_value=venv_pip,
), \
patch.object(module, "require_uv", return_value=uv_bin), \
patch.object(module, "expose_uv_to_venv") as expose_uv, \
patch.object(module, "run") as run_mock, \
patch.object(module, "install_browser_runtime") as install_browser:
result = module.install_deps(
@@ -82,13 +93,23 @@ class LocalSetupConfigDirTests(unittest.TestCase):
)
self.assertEqual(result, venv_python)
run_mock.assert_any_call(["python3", "-m", "venv", str(venv_dir)])
self.assertTrue(
any(
call.args[0] == [str(venv_pip), "install", "-r", str(module.ROOT / "requirements.txt")]
for call in run_mock.call_args_list
)
command = run_mock.call_args.args[0]
self.assertEqual(
command,
[
str(uv_bin),
"sync",
"--project",
str(module.ROOT),
"--locked",
"--no-dev",
"--no-install-project",
"--python",
"python3",
],
)
self.assertEqual(run_mock.call_args.kwargs["env"]["UV_PROJECT_ENVIRONMENT"], str(venv_dir))
expose_uv.assert_called_once_with(uv_bin, venv_dir)
install_browser.assert_called_once_with(venv_python)
def test_package_install_env_maps_proxy_cache_and_index(self):
@@ -101,17 +122,17 @@ class LocalSetupConfigDirTests(unittest.TestCase):
"PIP_PROXY": "https://user:pass@mirror.example/simple",
"PACKAGE_CACHE_ROOT": str(Path(temp_dir) / "custom-package-cache"),
},
clear=False,
clear=True,
):
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")
self.assertNotIn("PIP_CACHE_DIR", env)
self.assertNotIn("PIP_INDEX_URL", env)
def test_package_install_env_defaults_cache_to_config_dir(self):
module = load_local_setup_module()
@@ -125,26 +146,24 @@ class LocalSetupConfigDirTests(unittest.TestCase):
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"))
self.assertNotIn("PIP_CACHE_DIR", env)
def test_package_install_env_preserves_explicit_cache_dirs(self):
def test_package_install_env_preserves_explicit_uv_cache_dir(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,
clear=True,
):
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):
@@ -153,19 +172,15 @@ class LocalSetupConfigDirTests(unittest.TestCase):
with patch.object(module.subprocess, "run"), patch("builtins.print") as print_mock:
module.run(
[
"python",
"-m",
"pip",
"install",
"-i",
"uv",
"sync",
"--default-index",
"https://user:pass@mirror.example/simple",
],
safe_command=[
"python",
"-m",
"pip",
"install",
"-i",
"uv",
"sync",
"--default-index",
"https://mirror.example/simple",
],
)
@@ -178,22 +193,22 @@ class LocalSetupConfigDirTests(unittest.TestCase):
module = load_local_setup_module()
command = [
"pip",
"install",
"--index-url=https://user:pass@mirror.example/simple",
"uv",
"sync",
"--default-index=https://user:pass@mirror.example/simple",
]
redacted = module.redact_command(command)
self.assertIn("--index-url=https://mirror.example/simple", redacted)
self.assertIn("--default-index=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",
"uv",
"sync",
"https://user:pass@mirror.example/simple?token=abc",
]
@@ -202,45 +217,115 @@ class LocalSetupConfigDirTests(unittest.TestCase):
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):
def test_require_uv_accepts_repository_version(self):
module = load_local_setup_module()
calls = []
uv_bin = Path("/opt/moviepilot/bin/uv")
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,
with patch.object(module.shutil, "which", return_value=str(uv_bin)), patch.object(
module, "capture", return_value=f"uv {module.UV_VERSION} (test-target)"
):
result = module.require_uv()
self.assertEqual(result, uv_bin.resolve())
def test_windows_expose_uv_keeps_existing_source_when_target_is_same(self):
module = load_local_setup_module()
with tempfile.TemporaryDirectory() as temp_dir:
venv_dir = Path(temp_dir) / "venv"
venv_python = venv_dir / "bin" / "python"
uv_bin = venv_dir / "Scripts" / "uv.exe"
uv_bin.parent.mkdir(parents=True)
uv_bin.write_bytes(b"uv-binary")
with patch.object(module.os, "name", "nt"):
result = module.expose_uv_to_venv(uv_bin, venv_dir)
self.assertEqual(result, uv_bin)
self.assertEqual(uv_bin.read_bytes(), b"uv-binary")
def test_recreate_preserves_uv_located_inside_old_venv(self):
module = load_local_setup_module()
commands = []
with tempfile.TemporaryDirectory() as temp_dir:
venv_dir = (Path(temp_dir) / "venv").resolve()
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"
uv_bin.parent.mkdir(parents=True)
uv_bin.write_bytes(b"uv-binary")
def fake_run(command, cwd=None, env=None, safe_command=None):
calls.append((command, env, safe_command))
uv_bin.write_text("", encoding="utf-8")
def fake_run(command, **_kwargs):
self.assertTrue(Path(command[0]).is_file())
commands.append(command)
with patch.object(module.shutil, "which", return_value=None), \
with patch.object(module, "ensure_supported_python"), \
patch.object(module, "require_uv", return_value=uv_bin), \
patch.object(module, "install_browser_runtime"), \
patch.object(module, "run", side_effect=fake_run):
module._ensure_uv_available_for_venv(venv_dir, venv_python)
module.install_deps(
python_bin="python.exe",
venv_dir=venv_dir,
recreate=True,
)
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))
self.assertEqual(len(commands), 1)
self.assertNotEqual(Path(commands[0][0]), uv_bin)
self.assertNotIn("--inexact", commands[0])
self.assertEqual(uv_bin.read_bytes(), b"uv-binary")
def test_windows_pip_upgrade_uses_package_env(self):
def test_recreate_rejects_python_from_target_venv(self):
module = load_local_setup_module()
with tempfile.TemporaryDirectory() as temp_dir:
venv_dir = (Path(temp_dir) / "venv").resolve()
python_bin = venv_dir / "bin" / "python"
python_bin.parent.mkdir(parents=True)
python_bin.touch()
with patch.object(module, "ensure_supported_python"), \
self.assertRaisesRegex(RuntimeError, "venv 外部"):
module.install_deps(
python_bin=str(python_bin),
venv_dir=venv_dir,
recreate=True,
)
def test_recreate_rejects_current_python_inside_target_venv(self):
module = load_local_setup_module()
with tempfile.TemporaryDirectory() as temp_dir:
venv_dir = (Path(temp_dir) / "venv").resolve()
running_python = venv_dir / "bin" / "python"
running_python.parent.mkdir(parents=True)
running_python.touch()
with patch.object(module, "ensure_supported_python"), patch.object(
module.sys, "executable", str(running_python)
), self.assertRaisesRegex(RuntimeError, "venv 外部"):
module.install_deps(
python_bin="/usr/bin/python3",
venv_dir=venv_dir,
recreate=True,
)
def test_recreate_resolves_python_command_through_path(self):
module = load_local_setup_module()
with tempfile.TemporaryDirectory() as temp_dir:
venv_dir = (Path(temp_dir) / "venv").resolve()
path_python = venv_dir / "bin" / "python"
path_python.parent.mkdir(parents=True)
path_python.touch()
with patch.object(module, "ensure_supported_python"), patch.object(
module.shutil, "which", return_value=str(path_python)
), self.assertRaisesRegex(RuntimeError, "venv 外部"):
module.install_deps(
python_bin="python3",
venv_dir=venv_dir,
recreate=True,
)
def test_windows_install_deps_uses_uv_without_pip_bootstrap(self):
module = load_local_setup_module()
calls = []
@@ -251,15 +336,12 @@ class LocalSetupConfigDirTests(unittest.TestCase):
"PIP_PROXY": "https://user:pass@mirror.example/simple",
"PACKAGE_CACHE_ROOT": str(Path(temp_dir) / "custom-package-cache"),
},
clear=False,
clear=True,
):
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")
uv_bin = root / "tools" / "uv.exe"
module.CONFIG_DIR = root / "config"
def fake_run(command, cwd=None, env=None, safe_command=None):
@@ -267,23 +349,24 @@ class LocalSetupConfigDirTests(unittest.TestCase):
with patch.object(module.os, "name", "nt"), \
patch.object(module, "ensure_supported_python"), \
patch.object(module, "require_uv", return_value=uv_bin), \
patch.object(module, "expose_uv_to_venv"), \
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]))
self.assertEqual(len(calls), 1)
command, env, safe_command = calls[0]
self.assertEqual(command[:2], [str(uv_bin), "sync"])
self.assertNotIn("pip", command)
self.assertEqual(env["UV_PROJECT_ENVIRONMENT"], str(venv_dir.resolve()))
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["UV_CACHE_DIR"], str(Path(temp_dir) / "custom-package-cache" / "uv"))
self.assertNotIn("user:pass", " ".join(safe_command or command))
def test_install_deps_uses_package_env_for_project_requirements(self):
def test_install_deps_uses_package_env_for_project_lock(self):
module = load_local_setup_module()
calls = []
@@ -294,26 +377,21 @@ class LocalSetupConfigDirTests(unittest.TestCase):
):
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")
uv_bin = root / "tools" / "uv"
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, "require_uv", return_value=uv_bin), \
patch.object(module, "expose_uv_to_venv"), \
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]))
project_sync = calls[0]
self.assertEqual(project_sync[0][:2], [str(uv_bin), "sync"])
self.assertIn("--locked", project_sync[0])
self.assertEqual(project_sync[1]["UV_DEFAULT_INDEX"], "https://user:pass@mirror.example/simple")
self.assertNotIn("user:pass", " ".join(project_sync[2] or project_sync[0]))
tests/test_moviepilot_launcher.py
+167
View File
@@ -0,0 +1,167 @@
from pathlib import Path
import os
import stat
import subprocess
import pytest
LAUNCHER = Path(__file__).resolve().parents[1] / "moviepilot"
@pytest.mark.parametrize(
"arguments",
[
("install", "deps", "--recreate"),
("setup", "--recreate"),
("update", "backend", "--recreate"),
],
)
def test_recreate_commands_use_external_bootstrap_python(tmp_path, arguments):
"""所有会删除 venv 的 launcher 入口都不能由目标 venv Python 执行。"""
root = tmp_path / "moviepilot"
root.mkdir()
launcher = root / "moviepilot"
launcher.write_text(LAUNCHER.read_text(encoding="utf-8"), encoding="utf-8")
launcher.chmod(launcher.stat().st_mode | stat.S_IXUSR)
(root / "scripts").mkdir()
(root / "scripts" / "local_setup.py").write_text("# test stub\n", encoding="utf-8")
bin_dir = tmp_path / "bin"
bin_dir.mkdir()
record = tmp_path / "record"
external_python = bin_dir / "python3.12"
external_python.write_text(
"#!/bin/sh\n"
"if [ \"$1\" = \"-\" ]; then exit 0; fi\n"
"printf '%s\\n' \"$0 $*\" > \"$MOVIEPILOT_TEST_RECORD\"\n",
encoding="utf-8",
)
external_python.chmod(external_python.stat().st_mode | stat.S_IXUSR)
venv_python = root / "venv" / "bin" / "python"
venv_python.parent.mkdir(parents=True)
venv_script = (
"#!/bin/sh\n"
"printf '%s\\n' \"$0 $*\" > \"$MOVIEPILOT_TEST_RECORD\"\n"
)
venv_python.write_text(venv_script, encoding="utf-8")
venv_python.chmod(venv_python.stat().st_mode | stat.S_IXUSR)
venv_alias = venv_python.with_name("python3.12")
venv_alias.write_text(venv_script, encoding="utf-8")
venv_alias.chmod(venv_alias.stat().st_mode | stat.S_IXUSR)
env = os.environ.copy()
env["PATH"] = f"{venv_python.parent}:{bin_dir}:/usr/bin:/bin"
env["MOVIEPILOT_TEST_RECORD"] = str(record)
subprocess.run(
[str(launcher), *arguments],
cwd=root,
env=env,
check=True,
capture_output=True,
text=True,
)
invocation = record.read_text(encoding="utf-8")
assert invocation.startswith(f"{external_python} ")
def test_recreate_accepts_explicit_external_python(tmp_path):
"""显式指定的外部 Python 可作为重建命令的执行解释器。"""
root = tmp_path / "moviepilot"
root.mkdir()
launcher = root / "moviepilot"
launcher.write_text(LAUNCHER.read_text(encoding="utf-8"), encoding="utf-8")
launcher.chmod(launcher.stat().st_mode | stat.S_IXUSR)
(root / "scripts").mkdir()
(root / "scripts" / "local_setup.py").write_text("# test stub\n", encoding="utf-8")
explicit_python = tmp_path / "custom-python"
record = tmp_path / "record"
explicit_python.write_text(
"#!/bin/sh\n"
"if [ \"$1\" = \"-\" ]; then exit 0; fi\n"
"printf '%s\\n' \"$0 $*\" > \"$MOVIEPILOT_TEST_RECORD\"\n",
encoding="utf-8",
)
explicit_python.chmod(explicit_python.stat().st_mode | stat.S_IXUSR)
env = os.environ.copy()
env["PATH"] = "/usr/bin:/bin"
env["MOVIEPILOT_TEST_RECORD"] = str(record)
subprocess.run(
[
str(launcher),
"install",
"deps",
"--recreate",
"--python",
str(explicit_python),
],
cwd=root,
env=env,
check=True,
capture_output=True,
text=True,
)
assert record.read_text(encoding="utf-8").startswith(f"{explicit_python} ")
@pytest.mark.parametrize(
"arguments",
[
("install", "deps", "--recreate"),
("setup", "--recreate"),
("update", "backend", "--recreate"),
],
)
@pytest.mark.parametrize("python_option", [(), ("--python", "python3.12")])
def test_recreate_excludes_custom_venv_from_external_bootstrap(
tmp_path, arguments, python_option
):
"""自定义 --venv 目录中的解释器也不能执行重建流程。"""
root = tmp_path / "moviepilot"
root.mkdir()
launcher = root / "moviepilot"
launcher.write_text(LAUNCHER.read_text(encoding="utf-8"), encoding="utf-8")
launcher.chmod(launcher.stat().st_mode | stat.S_IXUSR)
(root / "scripts").mkdir()
(root / "scripts" / "local_setup.py").write_text("# test stub\n", encoding="utf-8")
target_bin = tmp_path / "custom-venv" / "bin"
target_bin.mkdir(parents=True)
external_bin = tmp_path / "external-bin"
external_bin.mkdir()
record = tmp_path / "record"
target_python = target_bin / "python3.12"
external_python = external_bin / "python3.12"
for python_path in (target_python, external_python):
python_path.write_text(
"#!/bin/sh\n"
"if [ \"$1\" = \"-\" ]; then exit 0; fi\n"
"printf '%s\\n' \"$0 $*\" > \"$MOVIEPILOT_TEST_RECORD\"\n",
encoding="utf-8",
)
python_path.chmod(python_path.stat().st_mode | stat.S_IXUSR)
env = os.environ.copy()
env["PATH"] = f"{target_bin}:{external_bin}:/usr/bin:/bin"
env["MOVIEPILOT_TEST_RECORD"] = str(record)
subprocess.run(
[
str(launcher),
*arguments,
"--venv",
str(target_bin.parent),
*python_option,
],
cwd=root,
env=env,
check=True,
capture_output=True,
text=True,
)
assert record.read_text(encoding="utf-8").startswith(f"{external_python} ")
tests/test_package_installer.py
+9 -18
View File
@@ -17,10 +17,10 @@ def test_build_env_maps_proxy_and_cache(tmp_path, monkeypatch):
monkeypatch.delenv("PACKAGE_CACHE_ROOT", raising=False)
monkeypatch.setenv("HTTP_PROXY", "http://old.example:8080")
request = PackageInstallRequest(
requirements_file=tmp_path / "requirements.txt",
dependency_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",
package_index_url="https://user:pass@mirror.example/simple",
proxy_url="http://proxy.example:7890",
)
@@ -31,16 +31,14 @@ def test_build_env_maps_proxy_and_cache(tmp_path, monkeypatch):
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",
dependency_file=tmp_path / "requirements.txt",
python_bin=Path("/venv/bin/python"),
config_dir=tmp_path / "config",
)
@@ -48,7 +46,6 @@ def test_build_env_uses_package_cache_root_and_preserves_tool_cache_overrides(tm
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")
@@ -62,11 +59,11 @@ def test_build_strategies_prefers_uv_network_matrix_and_preserves_find_links(tmp
uv_bin.write_text("", encoding="utf-8")
request = PackageInstallRequest(
requirements_file=req,
dependency_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",
package_index_url="https://mirror.example/simple",
proxy_url="http://proxy.example:7890",
)
@@ -77,10 +74,6 @@ def test_build_strategies_prefers_uv_network_matrix_and_preserves_find_links(tmp
"uv:镜像",
"uv:代理",
"uv:直连",
"pip:镜像+代理",
"pip:镜像",
"pip:代理",
"pip:直连",
]
assert strategies[0].command[:3] == [str(uv_bin), "pip", "install"]
assert "--python" in strategies[0].command
@@ -93,23 +86,21 @@ def test_build_strategies_prefers_uv_network_matrix_and_preserves_find_links(tmp
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):
def test_build_strategies_fail_closed_when_uv_missing(tmp_path):
req = tmp_path / "requirements.txt"
req.write_text("demo\n", encoding="utf-8")
request = PackageInstallRequest(
requirements_file=req,
dependency_file=req,
python_bin=tmp_path / "venv" / "bin" / "python",
config_dir=tmp_path / "config",
)
with patch("app.adapters.system.package._find_uv", return_value=None):
with patch("app.adapters.system.package.find_uv", return_value=None):
strategies = build_package_install_strategies(request)
assert [strategy.strategy_name for strategy in strategies] == ["pip:直连"]
assert strategies == []
def test_redact_url_removes_userinfo():
tests/test_plugin_dependency_installer.py
+366 -4
View File
@@ -2,9 +2,13 @@ from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock
import pytest
from packaging.requirements import Requirement
from packaging.utils import canonicalize_name
from packaging.version import Version
from app.adapters.system.plugin.dependency import PluginDependencyInstaller
from app.adapters.system.plugin.manifest import load_dependency_file
def _write_requirements(root: Path, plugin_id: str, content: str) -> None:
@@ -14,6 +18,15 @@ def _write_requirements(root: Path, plugin_id: str, content: str) -> None:
(plugin_dir / "requirements.txt").write_text(content, encoding="utf-8")
def _write_pyproject(root: Path, plugin_id: str, content: str) -> Path:
"""写入一个测试插件的 pyproject 依赖清单。"""
plugin_dir = root / plugin_id.lower()
plugin_dir.mkdir(parents=True, exist_ok=True)
pyproject_file = plugin_dir / "pyproject.toml"
pyproject_file.write_text(content, encoding="utf-8")
return plugin_dir
def test_find_missing_merges_only_installed_plugin_constraints(tmp_path, monkeypatch):
"""依赖扫描只覆盖安装清单，并合并同名包的多插件约束。"""
plugin_root = tmp_path / "plugins"
@@ -58,10 +71,353 @@ def test_find_missing_skips_satisfied_constraints(tmp_path, monkeypatch):
assert installer.find_missing() == []
def test_find_missing_preserves_merged_extras(tmp_path, monkeypatch):
"""同一包的多插件约束合并后必须保留全部 extras。"""
plugin_root = tmp_path / "plugins"
_write_requirements(plugin_root, "Alpha", "Demo-Pkg[alpha]>=2\n")
_write_requirements(plugin_root, "Beta", "demo.pkg[beta]<4\n")
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha", "Beta"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(installer, "_installed_packages", lambda: {})
missing = installer.find_missing()
assert len(missing) == 1
requirement = Requirement(missing[0])
assert requirement.name == "demo_pkg"
assert requirement.extras == {"alpha", "beta"}
assert ">=2" in str(requirement.specifier)
assert "<4" in str(requirement.specifier)
def test_find_missing_preserves_direct_url(tmp_path, monkeypatch):
"""缺失的 direct URL 依赖必须按原安装来源返回。"""
plugin_root = tmp_path / "plugins"
direct_url = "https://example.com/packages/demo_pkg-2.0.0-py3-none-any.whl"
_write_requirements(
plugin_root,
"Alpha",
f"Demo-Pkg[feature] @ {direct_url}\n",
)
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(installer, "_installed_packages", lambda: {})
missing = installer.find_missing()
assert len(missing) == 1
requirement = Requirement(missing[0])
assert canonicalize_name(requirement.name) == canonicalize_name("Demo-Pkg")
assert requirement.extras == {"feature"}
assert requirement.url == direct_url
def test_find_missing_does_not_accept_base_package_for_extra(
tmp_path, monkeypatch
):
"""已安装基础包但未安装其 extra 依赖时必须继续恢复。"""
plugin_root = tmp_path / "plugins"
_write_requirements(plugin_root, "Alpha", "Demo[feature]>=1\n")
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(
installer, "_installed_packages", lambda: {"demo": Version("2.0")}
)
metadata = SimpleNamespace(
get_all=lambda key: {"Provides-Extra": ["feature"], "Requires-Dist": [
"feature-dependency>=1; extra == 'feature'"
]}.get(key, []),
)
monkeypatch.setattr(
installer,
"_installed_distribution",
lambda package_name: SimpleNamespace(metadata=metadata)
if package_name == "demo"
else None,
)
assert installer.find_missing() == ["demo[feature]>=1"]
def test_find_missing_accepts_satisfied_extra_dependencies(tmp_path, monkeypatch):
"""已安装 extra 及其依赖时不得重复恢复。"""
plugin_root = tmp_path / "plugins"
_write_requirements(plugin_root, "Alpha", "Demo[feature]>=1\n")
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(
installer,
"_installed_packages",
lambda: {"demo": Version("2.0"), "feature_dependency": Version("1.2")},
)
metadata = SimpleNamespace(
get_all=lambda key: {"Provides-Extra": ["feature"], "Requires-Dist": [
"feature-dependency>=1; extra == 'feature'"
]}.get(key, []),
)
monkeypatch.setattr(
installer,
"_installed_distribution",
lambda package_name: SimpleNamespace(metadata=metadata)
if package_name == "demo"
else None,
)
assert installer.find_missing() == []
def test_find_missing_rejects_missing_transitive_extra_dependency(
tmp_path, monkeypatch
):
"""extra 的传递依赖缺失时不能只因根包已安装就跳过恢复。"""
plugin_root = tmp_path / "plugins"
_write_requirements(plugin_root, "Alpha", "Demo[feature]>=1\n")
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(
installer,
"_installed_packages",
lambda: {"demo": Version("2.0"), "bridge": Version("1.0")},
)
metadata_by_name = {
"demo": SimpleNamespace(
metadata=SimpleNamespace(
get_all=lambda key: {
"Provides-Extra": ["feature"],
"Requires-Dist": ["bridge>=1; extra == 'feature'"],
}.get(key, [])
)
),
"bridge": SimpleNamespace(
metadata=SimpleNamespace(
get_all=lambda key: {
"Requires-Dist": ["leaf>=1"],
}.get(key, [])
)
),
}
monkeypatch.setattr(
installer,
"_installed_distribution",
lambda package_name: metadata_by_name.get(package_name),
)
assert installer.find_missing() == ["demo[feature]>=1"]
def test_find_missing_rejects_same_name_package_from_wrong_direct_url(
tmp_path, monkeypatch
):
"""存在不同 PEP 610 来源时，同名包不能满足 direct URL 依赖。"""
plugin_root = tmp_path / "plugins"
required_url = "https://example.com/packages/demo-2.0.0-py3-none-any.whl"
installed_url = "https://mirror.example.com/packages/demo-2.0.0-py3-none-any.whl"
_write_requirements(plugin_root, "Alpha", f"Demo @ {required_url}\n")
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(
installer, "_installed_packages", lambda: {"demo": Version("2.0")}
)
metadata = SimpleNamespace(get_all=lambda _key: [])
monkeypatch.setattr(
installer,
"_installed_distribution",
lambda _package_name: SimpleNamespace(
metadata=metadata,
read_text=lambda _name: '{"url": "' + installed_url + '"}',
),
)
assert installer.find_missing() == [f"demo @ {required_url}"]
def test_find_missing_accepts_matching_direct_url(tmp_path, monkeypatch):
"""同名包且 PEP 610 来源一致时应视为已满足。"""
plugin_root = tmp_path / "plugins"
direct_url = "https://example.com/packages/demo-2.0.0-py3-none-any.whl"
_write_requirements(plugin_root, "Alpha", f"Demo @ {direct_url}\n")
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(
installer, "_installed_packages", lambda: {"demo": Version("2.0")}
)
metadata = SimpleNamespace(get_all=lambda _key: [])
monkeypatch.setattr(
installer,
"_installed_distribution",
lambda _package_name: SimpleNamespace(
metadata=metadata,
read_text=lambda _name: '{"url": "' + direct_url + '"}',
),
)
assert installer.find_missing() == []
def test_find_missing_prefers_pyproject_project_dependencies(
tmp_path,
monkeypatch,
):
"""现代清单优先，且只消费 project.dependencies。"""
plugin_root = tmp_path / "plugins"
plugin_dir = _write_pyproject(
plugin_root,
"Alpha",
"""
[project]
name = "alpha"
version = "1.0.0"
dependencies = ["Modern-Pkg>=2"]
[dependency-groups]
dev = ["group-only>=1"]
""",
)
(plugin_dir / "requirements.txt").write_text(
"legacy-only>=1\n",
encoding="utf-8",
)
(plugin_dir / "uv.lock").write_text(
'package = [{ name = "lock-only", version = "1.0.0" }]\n',
encoding="utf-8",
)
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(installer, "_installed_packages", lambda: {})
missing = installer.find_missing()
assert missing == ["modern_pkg>=2"]
@pytest.mark.parametrize(
"pyproject",
[
"[project\n",
'[project]\ndependencies = "demo>=2"\n',
'[project]\ndependencies = ["not a requirement !!!"]\n',
'[project]\ndynamic = ["dependencies"]\n',
],
)
def test_find_missing_fails_closed_for_invalid_pyproject(
tmp_path,
monkeypatch,
pyproject,
):
"""现代清单无效时不得回退并消费旧 requirements。"""
plugin_root = tmp_path / "plugins"
plugin_dir = _write_pyproject(plugin_root, "Alpha", pyproject)
(plugin_dir / "requirements.txt").write_text(
"legacy-only>=1\n",
encoding="utf-8",
)
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(installer, "_installed_packages", lambda: {})
with pytest.raises(ValueError, match="pyproject.toml"):
installer.find_missing()
@pytest.mark.parametrize(
"pyproject",
[
'[project]\nversion = "1.0.0"\ndependencies = ["demo>=2"]\n',
'[project]\nname = "alpha"\ndependencies = ["demo>=2"]\n',
'[project]\nname = " "\nversion = "1.0.0"\n'
'dependencies = ["demo>=2"]\n',
'[project]\nname = "alpha"\nversion = " "\n'
'dependencies = ["demo>=2"]\n',
],
)
def test_find_missing_fails_closed_without_required_project_identity(
tmp_path,
monkeypatch,
pyproject,
):
"""现代清单缺少 uv 消费所需的 name 或 version 时必须拒绝安装。"""
plugin_root = tmp_path / "plugins"
_write_pyproject(plugin_root, "Alpha", pyproject)
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(installer, "_installed_packages", lambda: {})
with pytest.raises(ValueError, match="pyproject.toml"):
installer.find_missing()
def test_find_missing_accepts_dynamic_project_version(tmp_path, monkeypatch):
"""version 由构建后端动态提供时仍可消费静态 dependencies。"""
plugin_root = tmp_path / "plugins"
_write_pyproject(
plugin_root,
"Alpha",
'[project]\nname = "alpha"\ndynamic = ["version"]\n'
'dependencies = ["demo>=2"]\n',
)
installer = PluginDependencyInstaller(
Mock(),
installed_plugins_provider=lambda: ["Alpha"],
plugin_dir=plugin_root,
)
monkeypatch.setattr(installer, "_installed_packages", lambda: {})
assert installer.find_missing() == ["demo>=2"]
def test_load_dependency_file_accepts_custom_legacy_filename(tmp_path):
"""临时或自定义命名的旧格式依赖文件复用统一解析器。"""
dependency_file = tmp_path / "plugin-dependencies.txt"
dependency_file.write_text("Demo-Pkg>=2\n", encoding="utf-8")
manifest = load_dependency_file(dependency_file)
assert manifest.path == dependency_file
assert [str(requirement) for requirement in manifest.dependencies] == [
"Demo-Pkg>=2"
]
def test_install_uses_adapter_owned_temporary_requirements(tmp_path, monkeypatch):
"""批量依赖文件由依赖适配器创建并在 pip 返回后清理。"""
"""批量依赖文件由依赖适配器创建并在安装返回后清理。"""
helper = Mock()
helper.pip_install_with_fallback.return_value = (True, "installed")
installed_contents = []
def _install_packages(dependency_file, _wheels_dirs):
installed_contents.append(dependency_file.read_text(encoding="utf-8"))
return True, "installed"
helper.install_packages_with_fallback.side_effect = _install_packages
monkeypatch.setattr(
"app.adapters.system.plugin.dependency.settings",
SimpleNamespace(ROOT_PATH=tmp_path, TEMP_PATH=tmp_path / "temp"),
@@ -72,9 +428,15 @@ def test_install_uses_adapter_owned_temporary_requirements(tmp_path, monkeypatch
plugin_dir=tmp_path / "plugins",
)
result = installer.install(["demo>=2", "other"])
result = installer.install([
"demo[feature] @ https://example.com/demo.whl",
"other",
])
assert result == (True, "installed")
requirements_file = helper.pip_install_with_fallback.call_args.args[0]
assert installed_contents == [
"demo[feature] @ https://example.com/demo.whl\nother\n"
]
requirements_file = helper.install_packages_with_fallback.call_args.args[0]
assert requirements_file.name == "requirements.txt"
assert not requirements_file.exists()
tests/test_plugin_helper.py
+224 -144
View File
@@ -1,5 +1,6 @@
import asyncio
import io
import os
import stat
import sys
import tempfile
@@ -102,6 +103,14 @@ def _build_release_zip_member(name: str, *, symlink: bool = False) -> bytes:
return buffer.getvalue()
def _create_fake_uv(root: Path) -> Path:
"""创建仅供命令构造测试定位的 uv 可执行文件。"""
uv_bin = root / "venv" / "bin" / "uv"
uv_bin.parent.mkdir(parents=True, exist_ok=True)
uv_bin.write_text("", encoding="utf-8")
return uv_bin
def _patch_release_install_settings(monkeypatch, tmp_path: Path) -> None:
"""隔离 release 安装根目录，并阻止测试误触真实根路径。"""
monkeypatch.setattr("app.adapters.external.market.settings", SimpleNamespace(
@@ -979,7 +988,7 @@ class TestPluginHelper:
assert not annotated["system_version_compatible"]
assert "当前版本" in annotated["system_version_message"]
def test_pip_install_keeps_modules_imported_during_install(self):
def test_uv_install_keeps_modules_imported_during_install(self):
"""
验证依赖安装窗口内被其他任务导入的运行态模块不会被误删。
"""
@@ -1002,14 +1011,14 @@ class TestPluginHelper:
requirements_file = Path(temp_dir) / "requirements.txt"
requirements_file.write_text("demo-package\n", encoding="utf-8")
with patch("app.adapters.external.market.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
success, message = PluginHelper.install_packages_with_fallback(requirements_file)
assert success
assert "ok" == message
for module_name in module_names:
assert module_name in sys.modules
def test_pip_install_builds_uv_strategy_without_proxy_argument(self):
def test_uv_install_builds_uv_strategy_without_proxy_argument(self):
"""
插件依赖安装优先使用 uv 时，传输代理只进入子进程环境。
"""
@@ -1032,17 +1041,17 @@ class TestPluginHelper:
uv_bin.parent.mkdir(parents=True)
uv_bin.write_text("", encoding="utf-8")
with patch("app.adapters.system.package._find_uv", return_value=uv_bin), \
with patch("app.adapters.system.package.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={"pip check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
return_value={"uv check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
), \
patch("app.adapters.external.market.SystemUtils.execute_with_subprocess", side_effect=fake_execute), \
patch("app.adapters.external.market.settings.PROXY_HOST", "http://proxy.example:7890"), \
patch("app.adapters.external.market.settings.PIP_PROXY", "https://user:pass@mirror.example/simple"):
success, message = PluginHelper.pip_install_with_fallback(req)
success, message = PluginHelper.install_packages_with_fallback(req)
assert success
assert message == "ok"
@@ -1053,9 +1062,9 @@ class TestPluginHelper:
assert env["HTTPS_PROXY"] == "http://proxy.example:7890"
assert "user:pass" not in " ".join(safe_command)
def test_pip_install_serializes_concurrent_calls(self):
def test_uv_install_serializes_concurrent_calls(self):
"""
验证多个依赖安装请求会复用同一把锁串行执行 pip。
验证多个依赖安装请求会复用同一把锁串行执行 uv。
"""
try:
from app.adapters.external.market import PluginHelper
@@ -1082,7 +1091,7 @@ class TestPluginHelper:
def worker(requirements_file: Path):
try:
start_event.wait()
PluginHelper.pip_install_with_fallback(requirements_file)
PluginHelper.install_packages_with_fallback(requirements_file)
except Exception as err: # pragma: no cover - 仅用于并发测试失败诊断
errors.append(err)
@@ -1144,9 +1153,9 @@ class TestPluginHelper:
"bcrypt": Version("4.0.1"),
} == protected_packages
def test_pip_install_rejects_conflicting_runtime_dependency(self):
def test_uv_install_rejects_conflicting_runtime_dependency(self):
"""
验证插件如果试图覆盖主程序核心依赖，会在真正执行 pip 前被直接拒绝。
验证插件如果试图覆盖主程序核心依赖，会在真正执行安装前被直接拒绝。
"""
try:
from app.adapters.external.market import PluginHelper
@@ -1161,13 +1170,13 @@ class TestPluginHelper:
"_PluginHelper__get_protected_runtime_packages",
return_value={"fastapi": Version("0.115.14")}
):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
success, message = PluginHelper.install_packages_with_fallback(requirements_file)
assert not success
assert "主程序核心依赖" in message
assert "fastapi" in message
def test_pip_install_allows_changing_non_runtime_dependency(self):
def test_uv_install_allows_changing_non_runtime_dependency(self):
"""
验证非主程序依赖即便已安装，插件后续仍可调整其版本约束。
"""
@@ -1178,16 +1187,19 @@ class TestPluginHelper:
seen_install_commands = []
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
return True, "ok"
return True, "ok"
with tempfile.TemporaryDirectory() as temp_dir:
requirements_file = Path(temp_dir) / "requirements.txt"
root = Path(temp_dir)
requirements_file = root / "requirements.txt"
requirements_file.write_text("demo-package>=2\n", encoding="utf-8")
uv_bin = _create_fake_uv(root)
def fake_execute(cmd, env=None, safe_command=None):
if cmd[:3] == [str(uv_bin), "pip", "install"]:
seen_install_commands.append(cmd)
assert "-c" not in cmd
return True, "ok"
return True, "ok"
with patch.object(
PluginHelper,
"_PluginHelper__get_installed_packages",
@@ -1199,14 +1211,14 @@ class TestPluginHelper:
return_value={}
):
with patch("app.adapters.external.market.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
with patch("app.adapters.system.package._find_uv", return_value=None):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
with patch("app.adapters.system.package.find_uv", return_value=uv_bin):
success, message = PluginHelper.install_packages_with_fallback(requirements_file)
assert success
assert "ok" == message
assert 1 == len(seen_install_commands)
def test_pip_install_uses_runtime_constraints_file(self):
def test_uv_install_uses_runtime_constraints_file(self):
"""
验证插件依赖安装会固定主程序依赖的当前版本，防止共享 venv 被改写。
"""
@@ -1217,34 +1229,37 @@ class TestPluginHelper:
seen_constraints = []
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])
seen_constraints.append(constraint_file)
assert constraint_file.exists()
assert "fastapi==0.115.14" in constraint_file.read_text(encoding="utf-8")
return True, "ok"
return True, "ok"
with tempfile.TemporaryDirectory() as temp_dir:
requirements_file = Path(temp_dir) / "requirements.txt"
root = Path(temp_dir)
requirements_file = root / "requirements.txt"
requirements_file.write_text("demo-package\n", encoding="utf-8")
uv_bin = _create_fake_uv(root)
def fake_execute(cmd, env=None, safe_command=None):
if cmd[:3] == [str(uv_bin), "pip", "install"]:
constraint_index = cmd.index("-c") + 1
constraint_file = Path(cmd[constraint_index])
seen_constraints.append(constraint_file)
assert constraint_file.exists()
assert "fastapi==0.115.14" in constraint_file.read_text(encoding="utf-8")
return True, "ok"
return True, "ok"
with patch.object(
PluginHelper,
"_PluginHelper__get_protected_runtime_packages",
return_value={"fastapi": Version("0.115.14")}
):
with patch("app.adapters.external.market.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
with patch("app.adapters.system.package._find_uv", return_value=None):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
with patch("app.adapters.system.package.find_uv", return_value=uv_bin):
success, message = PluginHelper.install_packages_with_fallback(requirements_file)
assert success
assert "ok" == message
assert 1 == len(seen_constraints)
assert not seen_constraints[0].exists()
def test_pip_install_repairs_runtime_when_healthcheck_fails(self):
def test_uv_install_repairs_runtime_when_healthcheck_fails(self):
"""
验证插件依赖安装后若破坏运行环境，会先恢复主程序依赖，再向上层返回失败。
"""
@@ -1254,43 +1269,45 @@ class TestPluginHelper:
pytest.skip(f"missing dependency: {exc}")
repair_commands = []
pip_check_count = 0
pip_check_cmd = PluginHelper._PluginHelper__build_runtime_pip_command("check")
def fake_execute(cmd, env=None, safe_command=None):
nonlocal pip_check_count
if cmd[:4] == [sys.executable, "-m", "pip", "install"]:
if "-c" not in cmd:
repair_commands.append(cmd)
return True, "repaired"
return True, "installed"
if cmd == pip_check_cmd:
pip_check_count += 1
if pip_check_count == 2:
return False, "broken"
return True, "healthy"
if len(cmd) >= 3 and cmd[1] == "-c":
return True, "probe ok"
raise AssertionError(f"unexpected command: {cmd}")
uv_check_count = 0
with tempfile.TemporaryDirectory() as temp_dir:
requirements_file = Path(temp_dir) / "requirements.txt"
root = Path(temp_dir)
requirements_file = root / "requirements.txt"
requirements_file.write_text("demo-package\n", encoding="utf-8")
uv_bin = _create_fake_uv(root)
def fake_execute(cmd, env=None, safe_command=None):
nonlocal uv_check_count
if cmd[:3] == [str(uv_bin), "pip", "install"]:
if "-c" not in cmd:
repair_commands.append(cmd)
return True, "repaired"
return True, "installed"
if cmd[1:3] == ["pip", "check"]:
uv_check_count += 1
if uv_check_count == 2:
return False, "broken"
return True, "healthy"
if len(cmd) >= 3 and cmd[1] == "-c":
return True, "probe ok"
raise AssertionError(f"unexpected command: {cmd}")
with patch.object(
PluginHelper,
"_PluginHelper__get_protected_runtime_packages",
return_value={"fastapi": Version("0.115.14")}
):
with patch("app.adapters.external.market.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
with patch("app.adapters.system.package._find_uv", return_value=None):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
with patch("app.adapters.system.package.find_uv", return_value=uv_bin):
success, message = PluginHelper.install_packages_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_pip_install_allows_preexisting_healthcheck_failure(self):
def test_uv_install_allows_preexisting_healthcheck_failure(self):
"""
安装前已存在且安装后未新增的环境异常不应归因于本次插件依赖安装。
"""
@@ -1301,19 +1318,21 @@ class TestPluginHelper:
health_snapshots = [
{
"pip check": (False, "existing issue before install"),
"uv check": (False, "existing issue before install"),
"核心依赖导入检查": (True, "ok"),
},
{
"pip check": (False, "same issue with different command summary"),
"uv check": (False, "same issue with different command summary"),
"核心依赖导入检查": (True, "ok"),
},
]
with tempfile.TemporaryDirectory() as temp_dir:
requirements_file = Path(temp_dir) / "requirements.txt"
root = Path(temp_dir)
requirements_file = root / "requirements.txt"
requirements_file.write_text("demo-package\n", encoding="utf-8")
with patch("app.adapters.system.package._find_uv", return_value=None), \
uv_bin = _create_fake_uv(root)
with patch("app.adapters.system.package.find_uv", return_value=uv_bin), \
patch.object(PluginHelper, "_PluginHelper__get_protected_runtime_packages", return_value={}), \
patch.object(
PluginHelper,
@@ -1325,7 +1344,7 @@ class TestPluginHelper:
"app.adapters.external.market.SystemUtils.execute_with_subprocess",
return_value=(True, "installed"),
):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
success, message = PluginHelper.install_packages_with_fallback(requirements_file)
assert success
assert message == "installed"
@@ -1342,23 +1361,25 @@ class TestPluginHelper:
health_snapshots = [
{
"pip check": (False, "existing issue"),
"uv check": (False, "existing issue"),
"核心依赖导入检查": (True, "ok"),
},
{
"pip check": (False, "existing issue"),
"uv check": (False, "existing issue"),
"核心依赖导入检查": (False, "import failed"),
},
{
"pip check": (False, "existing issue"),
"uv check": (False, "existing issue"),
"核心依赖导入检查": (True, "ok"),
},
]
with tempfile.TemporaryDirectory() as temp_dir:
requirements_file = Path(temp_dir) / "requirements.txt"
root = Path(temp_dir)
requirements_file = root / "requirements.txt"
requirements_file.write_text("demo-package\n", encoding="utf-8")
with patch("app.adapters.system.package._find_uv", return_value=None), \
uv_bin = _create_fake_uv(root)
with patch("app.adapters.system.package.find_uv", return_value=uv_bin), \
patch.object(PluginHelper, "_PluginHelper__get_protected_runtime_packages", return_value={}), \
patch.object(
PluginHelper,
@@ -1374,7 +1395,7 @@ class TestPluginHelper:
"app.adapters.external.market.SystemUtils.execute_with_subprocess",
return_value=(True, "installed"),
):
success, message = PluginHelper.pip_install_with_fallback(requirements_file)
success, message = PluginHelper.install_packages_with_fallback(requirements_file)
assert not success
assert "核心依赖导入检查失败" in message
@@ -1400,16 +1421,17 @@ class TestPluginHelper:
root = Path(temp_dir)
req = root / "plugin-requirements.txt"
req.write_text("demo\n", encoding="utf-8")
uv_bin = _create_fake_uv(root)
with patch("app.adapters.system.package._find_uv", return_value=None), \
with patch("app.adapters.system.package.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=[
{"pip check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
{"pip check": (False, "broken"), "核心依赖导入检查": (True, "ok")},
{"pip check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
{"uv check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
{"uv check": (False, "broken"), "核心依赖导入检查": (True, "ok")},
{"uv check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
],
), \
patch.object(
@@ -1419,7 +1441,7 @@ class TestPluginHelper:
or (True, "runtime repaired"),
), \
patch("app.adapters.external.market.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
success, message = PluginHelper.pip_install_with_fallback(req)
success, message = PluginHelper.install_packages_with_fallback(req)
assert not success
assert "partial failure" in message or "恢复" in message
@@ -1453,15 +1475,15 @@ class TestPluginHelper:
uv_bin.parent.mkdir(parents=True)
uv_bin.write_text("", encoding="utf-8")
with patch("app.adapters.system.package._find_uv", return_value=uv_bin), \
with patch("app.adapters.system.package.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=[
{"pip check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
{"pip check": (False, "broken"), "核心依赖导入检查": (True, "ok")},
{"pip check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
{"uv check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
{"uv check": (False, "broken"), "核心依赖导入检查": (True, "ok")},
{"uv check": (True, "ok"), "核心依赖导入检查": (True, "ok")},
],
), \
patch.object(
@@ -1473,7 +1495,7 @@ class TestPluginHelper:
patch("app.adapters.external.market.settings.PIP_PROXY", "https://mirror.example/simple"), \
patch("app.adapters.external.market.settings.PROXY_HOST", "http://proxy.example:7890"), \
patch("app.adapters.external.market.SystemUtils.execute_with_subprocess", side_effect=fake_execute):
success, message = PluginHelper.pip_install_with_fallback(req)
success, message = PluginHelper.install_packages_with_fallback(req)
assert not success
assert "resolver failed" in message
@@ -1504,7 +1526,8 @@ class TestPluginHelper:
uv_bin.parent.mkdir(parents=True)
uv_bin.write_text("", encoding="utf-8")
with patch("app.adapters.system.package._find_uv", return_value=uv_bin), \
with patch("app.adapters.system.package.find_uv", return_value=uv_bin), \
patch.dict(os.environ, {}, clear=True), \
patch("app.adapters.external.market.settings.CONFIG_DIR", str(root / "config")), \
patch("app.adapters.external.market.settings.PACKAGE_CACHE_ROOT", str(root / "custom-package-cache")), \
patch("app.adapters.external.market.settings.PIP_PROXY", "https://user:pass@mirror.example/simple"), \
@@ -1519,14 +1542,13 @@ class TestPluginHelper:
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):
def test_async_package_install_runs_in_threadpool(self):
"""
验证异步安装路径会把同步 pip 安装派发到线程池，避免阻塞事件循环。
验证异步安装路径会把同步包安装派发到线程池，避免阻塞事件循环。
"""
try:
from app.adapters.external.market import PluginHelper
@@ -1539,7 +1561,7 @@ class TestPluginHelper:
calls = []
async def run_install():
return await helper._PluginHelper__async_pip_install_with_fallback(
return await helper._PluginHelper__async_install_packages_with_fallback(
requirements_file,
find_links_dirs
)
@@ -1554,7 +1576,7 @@ class TestPluginHelper:
assert success
assert "ok" == message
assert 1 == len(calls)
assert helper.pip_install_with_fallback == calls[0][0]
assert helper.install_packages_with_fallback == calls[0][0]
assert (requirements_file, find_links_dirs) == calls[0][1]
assert {} == calls[0][2]
@@ -2468,10 +2490,99 @@ class TestPluginHelper:
assert "dependency failed" == message
assert ["remove", "restore"] == calls
def test_prepare_content_via_filelist_sync_preinstalls_requirements_and_downloads(self, monkeypatch):
"""
文件列表安装会先尝试 requirements 预安装，再下载插件文件。
"""
def test_install_flow_sync_restores_backup_for_invalid_modern_manifest(self, tmp_path, monkeypatch):
"""现代清单无效时恢复旧插件目录。"""
from app.adapters.external import market as market_module
plugin_root = tmp_path / "plugins"
plugin_dir = plugin_root / PLUGIN_ID.lower()
plugin_dir.mkdir(parents=True)
(plugin_dir / "old.txt").write_text("old", encoding="utf-8")
monkeypatch.setattr(market_module, "PLUGIN_DIR", plugin_root)
monkeypatch.setattr(market_module.settings, "CONFIG_DIR", str(tmp_path))
def prepare_content():
plugin_dir.mkdir(parents=True)
(plugin_dir / "pyproject.toml").write_text(
"[project]\nname = 'demo'\n",
encoding="utf-8",
)
return True, ""
success, message = market_module.PluginHelper()._PluginHelper__install_flow_sync(
PLUGIN_ID,
False,
prepare_content,
)
assert not success
assert "project.version" in message
assert (plugin_dir / "old.txt").read_text(encoding="utf-8") == "old"
assert not (plugin_dir / "pyproject.toml").exists()
def test_install_dependencies_prefers_plugin_pyproject(self, tmp_path, monkeypatch):
"""同步安装入口只消费双清单中的 pyproject。"""
from app.adapters.external import market as market_module
plugin_root = tmp_path / "plugins"
plugin_dir = plugin_root / "demoplugin"
plugin_dir.mkdir(parents=True)
pyproject_file = plugin_dir / "pyproject.toml"
pyproject_file.write_text(
'[project]\nname = "demo"\nversion = "1.0.0"\ndependencies = ["modern>=1"]\n',
encoding="utf-8",
)
(plugin_dir / "requirements.txt").write_text("legacy>=1\n", encoding="utf-8")
helper = market_module.PluginHelper()
seen = []
monkeypatch.setattr(market_module, "PLUGIN_DIR", plugin_root)
monkeypatch.setattr(
helper,
"install_packages_with_fallback",
lambda path: seen.append(path) or (True, ""),
)
result = helper._PluginHelper__install_dependencies_if_required("DemoPlugin")
assert result == (True, True, "")
assert seen == [pyproject_file]
def test_async_install_dependencies_prefers_plugin_pyproject(self, tmp_path, monkeypatch):
"""异步安装入口只消费双清单中的 pyproject。"""
from app.adapters.external import market as market_module
plugin_root = tmp_path / "plugins"
plugin_dir = plugin_root / "demoplugin"
plugin_dir.mkdir(parents=True)
pyproject_file = plugin_dir / "pyproject.toml"
pyproject_file.write_text(
'[project]\nname = "demo"\nversion = "1.0.0"\ndependencies = ["modern>=1"]\n',
encoding="utf-8",
)
(plugin_dir / "requirements.txt").write_text("legacy>=1\n", encoding="utf-8")
helper = market_module.PluginHelper()
seen = []
async def fake_install(path, _find_links=None):
seen.append(path)
return True, ""
monkeypatch.setattr(market_module, "PLUGIN_DIR", plugin_root)
monkeypatch.setattr(
helper,
"_PluginHelper__async_install_packages_with_fallback",
fake_install,
)
result = asyncio.run(
helper._PluginHelper__async_install_dependencies_if_required("DemoPlugin")
)
assert result == (True, True, "")
assert seen == [pyproject_file]
def test_prepare_content_via_filelist_sync_downloads_dependency_manifests_once(self, monkeypatch):
"""文件列表准备会完整下载内容，依赖由统一安装流程处理。"""
try:
from app.adapters.external.market import PluginHelper
except ModuleNotFoundError as exc:
@@ -2479,55 +2590,28 @@ class TestPluginHelper:
helper = PluginHelper()
calls = []
requirements = {"name": "requirements.txt", "download_url": "https://example.com/requirements.txt"}
file_list = [requirements, {"name": "__init__.py", "download_url": "https://example.com/__init__.py"}]
file_list = [
{"name": "pyproject.toml", "download_url": "https://example.com/pyproject.toml"},
{"name": "requirements.txt", "download_url": "https://example.com/requirements.txt"},
{"name": "__init__.py", "download_url": "https://example.com/__init__.py"},
]
monkeypatch.setattr(helper, "_PluginHelper__get_file_list", lambda *_args: (file_list, ""))
monkeypatch.setattr(
helper,
"_PluginHelper__download_and_install_requirements",
lambda *_args: calls.append("requirements") or (True, ""),
)
def fake_download(*args):
calls.append(args)
return True, ""
monkeypatch.setattr(
helper,
"_PluginHelper__download_files",
lambda *_args: calls.append("download") or (True, ""),
fake_download,
)
success, message = helper._PluginHelper__prepare_content_via_filelist_sync("demoplugin", "demo/repo", "v2")
assert success
assert "" == message
assert ["requirements", "download"] == calls
def test_prepare_content_via_filelist_sync_continues_when_requirements_preinstall_fails(self, monkeypatch):
"""
requirements 预安装失败不阻断文件下载，最终依赖安装由统一流程兜底。
"""
try:
from app.adapters.external.market import PluginHelper
except ModuleNotFoundError as exc:
pytest.skip(f"missing dependency: {exc}")
helper = PluginHelper()
calls = []
file_list = [{"name": "requirements.txt"}, {"name": "__init__.py"}]
monkeypatch.setattr(helper, "_PluginHelper__get_file_list", lambda *_args: (file_list, ""))
monkeypatch.setattr(
helper,
"_PluginHelper__download_and_install_requirements",
lambda *_args: calls.append("requirements") or (False, "preinstall failed"),
)
monkeypatch.setattr(
helper,
"_PluginHelper__download_files",
lambda *_args: calls.append("download") or (True, ""),
)
success, message = helper._PluginHelper__prepare_content_via_filelist_sync("demoplugin", "demo/repo", "v2")
assert success
assert "" == message
assert ["requirements", "download"] == calls
assert calls == [("demoplugin", file_list, "demo/repo", "v2")]
def test_prepare_content_via_filelist_sync_reports_missing_file_list(self, monkeypatch):
"""
@@ -2564,10 +2648,8 @@ class TestPluginHelper:
assert not success
assert "download failed" == message
def test_async_prepare_content_via_filelist_preinstalls_requirements_and_downloads(self, monkeypatch):
"""
异步文件列表安装会先尝试 requirements 预安装，再下载插件文件。
"""
def test_async_prepare_content_via_filelist_downloads_dependency_manifests_once(self, monkeypatch):
"""异步文件列表准备会完整下载内容，依赖由统一安装流程处理。"""
try:
from app.adapters.external.market import PluginHelper
except ModuleNotFoundError as exc:
@@ -2575,22 +2657,20 @@ class TestPluginHelper:
helper = PluginHelper()
calls = []
requirements = {"name": "requirements.txt", "download_url": "https://example.com/requirements.txt"}
file_list = [requirements, {"name": "__init__.py", "download_url": "https://example.com/__init__.py"}]
file_list = [
{"name": "pyproject.toml", "download_url": "https://example.com/pyproject.toml"},
{"name": "requirements.txt", "download_url": "https://example.com/requirements.txt"},
{"name": "__init__.py", "download_url": "https://example.com/__init__.py"},
]
async def fake_file_list(*_args):
return file_list, ""
async def fake_requirements(*_args):
calls.append("requirements")
return True, ""
async def fake_download(*_args):
calls.append("download")
async def fake_download(*args):
calls.append(args)
return True, ""
monkeypatch.setattr(helper, "_PluginHelper__async_get_file_list", fake_file_list)
monkeypatch.setattr(helper, "_PluginHelper__async_download_and_install_requirements", fake_requirements)
monkeypatch.setattr(helper, "_PluginHelper__async_download_files", fake_download)
success, message = asyncio.run(
@@ -2599,7 +2679,7 @@ class TestPluginHelper:
assert success
assert "" == message
assert ["requirements", "download"] == calls
assert calls == [("demoplugin", file_list, "demo/repo", "v2")]
def test_async_prepare_content_via_filelist_reports_missing_file_list(self, monkeypatch):
"""
tests/test_plugin_local_sync.py
+160
View File
@@ -452,11 +452,171 @@ def test_local_requirements_change_still_does_not_sync_or_reload(
reload_spy = Mock()
monkeypatch.setattr(plugin_manager, "_sync_local_plugin_if_installed", sync_spy)
monkeypatch.setattr(plugin_manager, "reload_plugin", reload_spy)
log = Mock()
monkeypatch.setattr("app.runtime.extensions.plugin_manager.logger", log)
plugin_manager._run_file_watcher()
sync_spy.assert_not_called()
reload_spy.assert_not_called()
log.warning.assert_called_once()
def test_local_pyproject_change_prompts_reinstall_without_sync_or_reload(
tmp_path,
monkeypatch,
plugin_manager: PluginManager,
) -> None:
"""生效的现代依赖清单变化只提示重新安装。"""
repo_path, source_file = _build_local_plugin_repo(tmp_path)
pyproject_file = source_file.parent / "pyproject.toml"
pyproject_file.write_text(
'[project]\ndependencies = ["example==1.0.0"]\n',
encoding="utf-8",
)
_configure_local_watcher(
monkeypatch,
tmp_path,
repo_path,
{(Change.modified, str(pyproject_file))},
)
monkeypatch.setattr(PluginHelper, "get_current_system_version", lambda: Version("2.13.11"))
sync_spy = Mock()
reload_spy = Mock()
monkeypatch.setattr(plugin_manager, "_sync_local_plugin_if_installed", sync_spy)
monkeypatch.setattr(plugin_manager, "reload_plugin", reload_spy)
log = Mock()
monkeypatch.setattr("app.runtime.extensions.plugin_manager.logger", log)
plugin_manager._run_file_watcher()
sync_spy.assert_not_called()
reload_spy.assert_not_called()
log.warning.assert_called_once()
def test_local_inactive_requirements_change_is_debug_only(
tmp_path,
monkeypatch,
plugin_manager: PluginManager,
) -> None:
"""现代清单生效时，旧 requirements 变化不提示重新安装。"""
repo_path, source_file = _build_local_plugin_repo(tmp_path)
(source_file.parent / "pyproject.toml").write_text(
'[project]\ndependencies = ["example==1.0.0"]\n',
encoding="utf-8",
)
requirements_file = source_file.parent / "requirements.txt"
requirements_file.write_text("legacy==1.0.0\n", encoding="utf-8")
_configure_local_watcher(
monkeypatch,
tmp_path,
repo_path,
{(Change.modified, str(requirements_file))},
)
monkeypatch.setattr(PluginHelper, "get_current_system_version", lambda: Version("2.13.11"))
sync_spy = Mock()
reload_spy = Mock()
monkeypatch.setattr(plugin_manager, "_sync_local_plugin_if_installed", sync_spy)
monkeypatch.setattr(plugin_manager, "reload_plugin", reload_spy)
log = Mock()
monkeypatch.setattr("app.runtime.extensions.plugin_manager.logger", log)
plugin_manager._run_file_watcher()
sync_spy.assert_not_called()
reload_spy.assert_not_called()
log.warning.assert_not_called()
log.debug.assert_called_once()
def test_deleting_active_pyproject_prompts_for_requirements_takeover(
tmp_path,
monkeypatch,
plugin_manager: PluginManager,
) -> None:
"""删除现代清单后旧清单接管时必须提示重新安装。"""
repo_path, source_file = _build_local_plugin_repo(tmp_path)
requirements_file = source_file.parent / "requirements.txt"
requirements_file.write_text("legacy==1.0.0\n", encoding="utf-8")
pyproject_file = source_file.parent / "pyproject.toml"
pyproject_file.write_text(
'[project]\nname = "demo"\nversion = "1.0.0"\n'
'dependencies = ["modern==2.0.0"]\n',
encoding="utf-8",
)
pyproject_file.unlink()
_configure_local_watcher(
monkeypatch,
tmp_path,
repo_path,
{(Change.deleted, str(pyproject_file))},
)
monkeypatch.setattr(PluginHelper, "get_current_system_version", lambda: Version("2.13.11"))
log = Mock()
monkeypatch.setattr("app.runtime.extensions.plugin_manager.logger", log)
plugin_manager._run_file_watcher()
log.warning.assert_called_once()
log.debug.assert_not_called()
def test_deleting_only_active_requirements_prompts_reinstall(
tmp_path,
monkeypatch,
plugin_manager: PluginManager,
) -> None:
"""删除唯一生效的旧清单时必须提示依赖集合已变化。"""
repo_path, source_file = _build_local_plugin_repo(tmp_path)
requirements_file = source_file.parent / "requirements.txt"
requirements_file.write_text("legacy==1.0.0\n", encoding="utf-8")
requirements_file.unlink()
_configure_local_watcher(
monkeypatch,
tmp_path,
repo_path,
{(Change.deleted, str(requirements_file))},
)
monkeypatch.setattr(PluginHelper, "get_current_system_version", lambda: Version("2.13.11"))
log = Mock()
monkeypatch.setattr("app.runtime.extensions.plugin_manager.logger", log)
plugin_manager._run_file_watcher()
log.warning.assert_called_once()
log.debug.assert_not_called()
def test_deleting_inactive_requirements_is_debug_only(
tmp_path,
monkeypatch,
plugin_manager: PluginManager,
) -> None:
"""现代清单仍生效时，删除旧清单不得提示重新安装。"""
repo_path, source_file = _build_local_plugin_repo(tmp_path)
(source_file.parent / "pyproject.toml").write_text(
'[project]\nname = "demo"\nversion = "1.0.0"\n'
'dependencies = ["modern==2.0.0"]\n',
encoding="utf-8",
)
requirements_file = source_file.parent / "requirements.txt"
requirements_file.write_text("legacy==1.0.0\n", encoding="utf-8")
requirements_file.unlink()
_configure_local_watcher(
monkeypatch,
tmp_path,
repo_path,
{(Change.deleted, str(requirements_file))},
)
monkeypatch.setattr(PluginHelper, "get_current_system_version", lambda: Version("2.13.11"))
log = Mock()
monkeypatch.setattr("app.runtime.extensions.plugin_manager.logger", log)
plugin_manager._run_file_watcher()
log.warning.assert_not_called()
log.debug.assert_called_once()
def test_local_python_change_still_syncs_and_reloads_plugin(
tests/test_plugin_sync_service.py
+32
View File
@@ -0,0 +1,32 @@
"""插件市场同步服务用例。"""
from types import SimpleNamespace
from unittest.mock import Mock
from app.runtime.extensions.plugin.sync import PluginSyncService
def test_market_sync_keeps_install_rollback_enabled() -> None:
"""自动更新插件时保留旧版本，失败后可由安装器恢复。"""
plugin = SimpleNamespace(
id="DemoPlugin",
repo_url="https://example.com/plugins",
plugin_name="Demo",
plugin_version="1.0.0",
system_version_compatible=True,
)
install = Mock(return_value=(True, ""))
service = PluginSyncService(
frozen=lambda: False,
installed_plugins=lambda: [plugin.id],
online_plugins=lambda: [plugin],
local_plugins=lambda: [],
merge_plugins=lambda items, *_args: items,
plugin_exists=lambda *_args: False,
install=install,
report=Mock(),
log=Mock(),
)
assert service.sync() == [plugin.id]
install.assert_called_once_with(plugin.id, plugin.repo_url, False)
tests/test_uv_pip_compat.py
-131
View File
@@ -1,131 +0,0 @@
from __future__ import annotations
import os
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path
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:
venv_bin = Path(temp_dir) / "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 = Path(temp_dir) / "argv.txt"
uv_bin = venv_bin / "uv"
uv_bin.write_text(
"#!/bin/sh\n"
# 测试只关心兼容层传给 uv 的参数，逐行记录可以避免 shell 转义差异干扰断言。
f"for arg in \"$@\"; do printf '%s\\n' \"$arg\" >> '{argv_file}'; 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', '')}",
},
)
return argv_file.read_text(encoding="utf-8").splitlines()
def test_pip_install_binds_venv_python(self):
argv = self.run_wrapper("pip", "install", "-r", "requirements.txt")
self.assertEqual(
[
"pip",
"install",
"--python",
argv[3],
"-r",
"requirements.txt",
],
argv,
)
self.assertTrue(argv[3].endswith("/venv/bin/python"))
def test_pip_install_keeps_explicit_environment(self):
argv = self.run_wrapper("pip", "install", "--system", "demo-package")
self.assertEqual(["pip", "install", "--system", "demo-package"], argv)
def test_pip_sync_binds_venv_python(self):
argv = self.run_wrapper("pip-sync", "requirements.txt")
self.assertEqual(["pip", "sync", "--python", argv[3], "requirements.txt"], argv)
self.assertTrue(argv[3].endswith("/venv/bin/python"))