mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
492 lines
17 KiB
Python
492 lines
17 KiB
Python
"""
|
|
验证插件包安装、替换和回滚事务。
|
|
"""
|
|
|
|
import errno
|
|
import shutil
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
|
|
from app.adapters.system.plugin.package import PluginPackageManager
|
|
from app.runtime.dependencies.native import LoadedNativeDependencySnapshot
|
|
|
|
|
|
def _manager(monkeypatch, tmp_path: Path) -> PluginPackageManager:
|
|
"""
|
|
构造使用隔离运行目录和事务目录的插件包管理器。
|
|
"""
|
|
settings = SimpleNamespace(
|
|
ROOT_PATH=tmp_path,
|
|
TEMP_PATH=tmp_path / "temp",
|
|
CONFIG_PATH=tmp_path / "config",
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.adapters.system.plugin.package.get_runtime_setting",
|
|
lambda key: getattr(settings, key),
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.adapters.system.plugin.package.capture_loaded_native_dependencies",
|
|
LoadedNativeDependencySnapshot,
|
|
)
|
|
return PluginPackageManager(source=Mock())
|
|
|
|
|
|
def test_checkpoint_rollback_restores_existing_package(monkeypatch, tmp_path):
|
|
"""
|
|
已存在插件在后续阶段失败时应完整恢复原文件。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
source_file = plugin_dir / "__init__.py"
|
|
source_file.write_text("old", encoding="utf-8")
|
|
|
|
checkpoint = manager.checkpoint("DemoPlugin")
|
|
source_file.write_text("new", encoding="utf-8")
|
|
(plugin_dir / "partial.py").write_text("partial", encoding="utf-8")
|
|
manager.rollback(checkpoint)
|
|
|
|
assert source_file.read_text(encoding="utf-8") == "old"
|
|
assert not (plugin_dir / "partial.py").exists()
|
|
assert not checkpoint.transaction_dir.exists()
|
|
|
|
|
|
def test_checkpoint_rollback_handles_cross_device_plugin_directory(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
"""
|
|
overlayfs 拒绝目录 rename 时仍应恢复旧插件并清理事务。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
source_file = plugin_dir / "__init__.py"
|
|
source_file.write_text("old", encoding="utf-8")
|
|
|
|
checkpoint = manager.checkpoint("DemoPlugin")
|
|
source_file.write_text("new", encoding="utf-8")
|
|
(plugin_dir / "partial.py").write_text("partial", encoding="utf-8")
|
|
|
|
original_replace = Path.replace
|
|
|
|
def replace(self: Path, target: Path) -> Path:
|
|
"""
|
|
只模拟运行目录跨设备 rename 失败，保留其他替换操作。
|
|
"""
|
|
if self == plugin_dir and target.name.startswith(
|
|
".demoplugin.previous-"
|
|
):
|
|
raise OSError(errno.EXDEV, "Invalid cross-device link")
|
|
return original_replace(self, target)
|
|
|
|
monkeypatch.setattr(Path, "replace", replace)
|
|
manager.rollback(checkpoint)
|
|
|
|
assert source_file.read_text(encoding="utf-8") == "old"
|
|
assert not (plugin_dir / "partial.py").exists()
|
|
assert not checkpoint.transaction_dir.exists()
|
|
|
|
|
|
def test_checkpoint_uses_injected_plugin_root(monkeypatch, tmp_path):
|
|
"""
|
|
显式装配的插件根目录必须覆盖全局运行目录设置。
|
|
"""
|
|
_manager(monkeypatch, tmp_path)
|
|
plugin_root = tmp_path / "custom-plugins"
|
|
plugin_dir = plugin_root / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
(plugin_dir / "__init__.py").write_text("custom", encoding="utf-8")
|
|
manager = PluginPackageManager(source=Mock(), plugin_root=plugin_root)
|
|
|
|
checkpoint = manager.checkpoint("DemoPlugin")
|
|
|
|
assert checkpoint.plugin_dir == plugin_dir.resolve()
|
|
assert (checkpoint.transaction_dir / "package" / "__init__.py").read_text(
|
|
encoding="utf-8"
|
|
) == "custom"
|
|
manager.commit(checkpoint)
|
|
|
|
|
|
def test_remove_plugin_uses_package_owner_path_boundary(tmp_path):
|
|
"""
|
|
物理卸载只删除注入根目录内的目标插件。
|
|
"""
|
|
plugin_root = tmp_path / "plugins"
|
|
plugin_dir = plugin_root / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
(plugin_dir / "__init__.py").write_text("plugin", encoding="utf-8")
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
manager = PluginPackageManager(source=Mock(), plugin_root=plugin_root)
|
|
|
|
assert manager.remove_plugin("DemoPlugin") is True
|
|
assert not plugin_dir.exists()
|
|
assert outside.exists()
|
|
assert manager.remove_plugin("DemoPlugin") is False
|
|
with pytest.raises(ValueError, match="非法插件ID"):
|
|
manager.remove_plugin("../outside")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("remote_path", "package_version"),
|
|
[
|
|
("../escaped.py", None),
|
|
("/tmp/escaped.py", None),
|
|
("C:\\escaped.py", None),
|
|
("plugins/other/file.py", None),
|
|
("plugins.v2/demoplugin/../escaped.py", "v2"),
|
|
],
|
|
)
|
|
def test_file_list_download_rejects_paths_outside_plugin_root(
|
|
monkeypatch,
|
|
tmp_path,
|
|
remote_path,
|
|
package_version,
|
|
):
|
|
"""
|
|
同步文件列表安装不得把远端路径写到当前插件目录之外。
|
|
"""
|
|
plugin_root = tmp_path / "plugins"
|
|
manager = PluginPackageManager(source=Mock(), plugin_root=plugin_root)
|
|
request = Mock()
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_PluginPackageManager__request_with_fallback",
|
|
request,
|
|
)
|
|
|
|
result = manager._PluginPackageManager__download_files(
|
|
"DemoPlugin",
|
|
[{"path": remote_path, "download_url": "https://example.invalid/file"}],
|
|
"owner/repo",
|
|
package_version,
|
|
)
|
|
|
|
assert result == (False, "插件文件路径无效")
|
|
request.assert_not_called()
|
|
assert not (tmp_path / "escaped.py").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_file_list_download_rejects_path_outside_plugin_root(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
"""
|
|
异步文件列表安装复用同一受控路径边界。
|
|
"""
|
|
manager = PluginPackageManager(source=Mock(), plugin_root=tmp_path / "plugins")
|
|
request = AsyncMock()
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_PluginPackageManager__async_request_with_fallback",
|
|
request,
|
|
)
|
|
|
|
result = await manager._PluginPackageManager__async_download_files(
|
|
"DemoPlugin",
|
|
[
|
|
{
|
|
"path": "plugins.v2/demoplugin/../../escaped.py",
|
|
"download_url": "https://example.invalid/file",
|
|
}
|
|
],
|
|
"owner/repo",
|
|
"v2",
|
|
)
|
|
|
|
assert result == (False, "插件文件路径无效")
|
|
request.assert_not_awaited()
|
|
assert not (tmp_path / "escaped.py").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_list_download_rejects_traversal_directory_names(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
"""
|
|
目录项名称不得扩大后续市场查询到当前插件树之外。
|
|
"""
|
|
manager = PluginPackageManager(source=Mock(), plugin_root=tmp_path / "plugins")
|
|
sync_query = Mock()
|
|
async_query = AsyncMock()
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_PluginPackageManager__get_file_list",
|
|
sync_query,
|
|
)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_PluginPackageManager__async_get_file_list",
|
|
async_query,
|
|
)
|
|
item = {"name": "..", "download_url": None}
|
|
|
|
assert manager._PluginPackageManager__download_files(
|
|
"DemoPlugin", [item], "owner/repo"
|
|
) == (False, "插件目录路径无效")
|
|
assert await manager._PluginPackageManager__async_download_files(
|
|
"DemoPlugin", [item], "owner/repo"
|
|
) == (False, "插件目录路径无效")
|
|
sync_query.assert_not_called()
|
|
async_query.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_list_download_preserves_binary_payload_in_injected_plugin_root(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
"""
|
|
同步与异步文件列表都应在受控根目录内保留原始文件字节。
|
|
"""
|
|
plugin_root = tmp_path / "plugins"
|
|
payload = b"\x00\xffwheel-payload"
|
|
response = SimpleNamespace(status_code=200, content=payload, text="wrong-text")
|
|
manager = PluginPackageManager(source=Mock(), plugin_root=plugin_root)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_PluginPackageManager__request_with_fallback",
|
|
Mock(return_value=response),
|
|
)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_PluginPackageManager__async_request_with_fallback",
|
|
AsyncMock(return_value=response),
|
|
)
|
|
item = {
|
|
"path": "plugins.v2/demoplugin/nested/file.py",
|
|
"download_url": "https://example.invalid/file",
|
|
}
|
|
|
|
assert manager._PluginPackageManager__download_files(
|
|
"DemoPlugin", [item], "owner/repo", "v2"
|
|
) == (True, "")
|
|
assert await manager._PluginPackageManager__async_download_files(
|
|
"DemoPlugin", [item], "owner/repo", "v2"
|
|
) == (True, "")
|
|
assert (plugin_root / "demoplugin" / "nested" / "file.py").read_bytes() == payload
|
|
|
|
|
|
def test_checkpoint_does_not_scan_native_dependencies(monkeypatch, tmp_path):
|
|
"""
|
|
普通插件文件快照不应枚举宿主全部原生发行包。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
capture = Mock()
|
|
monkeypatch.setattr(
|
|
"app.adapters.system.plugin.package.capture_loaded_native_dependencies",
|
|
capture,
|
|
)
|
|
|
|
checkpoint = manager.checkpoint("DemoPlugin")
|
|
|
|
assert checkpoint.native_dependencies is None
|
|
capture.assert_not_called()
|
|
assert manager.native_dependency_changes(checkpoint) == ()
|
|
|
|
|
|
def test_checkpoint_rollback_removes_new_package(monkeypatch, tmp_path):
|
|
"""
|
|
首次安装失败时应删除安装过程创建的不完整目录。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
checkpoint = manager.checkpoint("DemoPlugin")
|
|
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
(plugin_dir / "__init__.py").write_text("partial", encoding="utf-8")
|
|
|
|
manager.rollback(checkpoint)
|
|
|
|
assert not plugin_dir.exists()
|
|
assert not checkpoint.transaction_dir.exists()
|
|
|
|
|
|
def test_rollback_does_not_delete_package_when_snapshot_is_missing(monkeypatch, tmp_path):
|
|
"""
|
|
补偿快照损坏时先失败，不能先删除当前可用插件。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
(plugin_dir / "__init__.py").write_text("old", encoding="utf-8")
|
|
|
|
checkpoint = manager.checkpoint("DemoPlugin")
|
|
shutil.rmtree(checkpoint.transaction_dir / "package")
|
|
(plugin_dir / "__init__.py").write_text("new", encoding="utf-8")
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
manager.rollback(checkpoint)
|
|
|
|
assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "new"
|
|
|
|
|
|
def test_durable_checkpoint_stages_backup_without_overwriting_current_backup(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
"""
|
|
数据库提交前只准备新备份，现有容器恢复材料保持可用。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(
|
|
"app.adapters.system.plugin.package.SystemUtils.is_docker",
|
|
lambda: True,
|
|
)
|
|
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
|
|
backup_dir = tmp_path / "config" / "plugins_backup" / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
backup_dir.mkdir(parents=True)
|
|
(plugin_dir / "__init__.py").write_text("new", encoding="utf-8")
|
|
(backup_dir / "__init__.py").write_text("old", encoding="utf-8")
|
|
|
|
checkpoint = manager.checkpoint("DemoPlugin", "txn-1")
|
|
manager.stage_persistent_backup(checkpoint)
|
|
|
|
assert checkpoint.transaction_dir.parent == tmp_path / "config" / "plugin_transactions"
|
|
assert (backup_dir / "__init__.py").read_text(encoding="utf-8") == "old"
|
|
assert checkpoint.backup_staging_dir is not None
|
|
assert (checkpoint.backup_staging_dir / "__init__.py").read_text(
|
|
encoding="utf-8"
|
|
) == "new"
|
|
|
|
|
|
def test_activate_and_finalize_persistent_backup_are_retryable(monkeypatch, tmp_path):
|
|
"""
|
|
备份激活保留旧载荷，数据库提交后的清理可以重复执行。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(
|
|
"app.adapters.system.plugin.package.SystemUtils.is_docker",
|
|
lambda: True,
|
|
)
|
|
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
|
|
backup_dir = tmp_path / "config" / "plugins_backup" / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
backup_dir.mkdir(parents=True)
|
|
(plugin_dir / "__init__.py").write_text("new", encoding="utf-8")
|
|
(backup_dir / "__init__.py").write_text("old", encoding="utf-8")
|
|
checkpoint = manager.checkpoint("DemoPlugin", "txn-2")
|
|
manager.stage_persistent_backup(checkpoint)
|
|
|
|
manager.activate_persistent_backup(checkpoint)
|
|
manager.activate_persistent_backup(checkpoint)
|
|
|
|
assert (backup_dir / "__init__.py").read_text(encoding="utf-8") == "new"
|
|
assert checkpoint.backup_staging_dir is not None
|
|
assert not checkpoint.backup_staging_dir.exists()
|
|
assert checkpoint.backup_previous_dir is not None
|
|
assert (checkpoint.backup_previous_dir / "__init__.py").read_text(
|
|
encoding="utf-8"
|
|
) == "old"
|
|
|
|
manager.finalize_persistent_backup(checkpoint)
|
|
manager.finalize_persistent_backup(checkpoint)
|
|
|
|
assert not checkpoint.backup_previous_dir.exists()
|
|
|
|
|
|
def test_rollback_removes_staging_but_preserves_current_backup(monkeypatch, tmp_path):
|
|
"""
|
|
提交前失败只恢复运行目录，不修改上一份容器恢复备份。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(
|
|
"app.adapters.system.plugin.package.SystemUtils.is_docker",
|
|
lambda: True,
|
|
)
|
|
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
|
|
backup_dir = tmp_path / "config" / "plugins_backup" / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
backup_dir.mkdir(parents=True)
|
|
(plugin_dir / "__init__.py").write_text("old-runtime", encoding="utf-8")
|
|
(backup_dir / "__init__.py").write_text("old-backup", encoding="utf-8")
|
|
checkpoint = manager.checkpoint("DemoPlugin", "txn-3")
|
|
(plugin_dir / "__init__.py").write_text("new-runtime", encoding="utf-8")
|
|
manager.stage_persistent_backup(checkpoint)
|
|
|
|
manager.rollback(checkpoint)
|
|
|
|
assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "old-runtime"
|
|
assert (backup_dir / "__init__.py").read_text(encoding="utf-8") == "old-backup"
|
|
assert checkpoint.backup_staging_dir is not None
|
|
assert not checkpoint.backup_staging_dir.exists()
|
|
|
|
|
|
def test_rollback_after_backup_activation_restores_previous_backup(
|
|
monkeypatch,
|
|
tmp_path,
|
|
):
|
|
"""
|
|
数据库提交前失败时，已激活的新备份必须回退到上一份载荷。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(
|
|
"app.adapters.system.plugin.package.SystemUtils.is_docker",
|
|
lambda: True,
|
|
)
|
|
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
|
|
backup_dir = tmp_path / "config" / "plugins_backup" / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
backup_dir.mkdir(parents=True)
|
|
(plugin_dir / "__init__.py").write_text("old-runtime", encoding="utf-8")
|
|
(backup_dir / "__init__.py").write_text("old-backup", encoding="utf-8")
|
|
checkpoint = manager.checkpoint("DemoPlugin", "txn-4")
|
|
(plugin_dir / "__init__.py").write_text("new-runtime", encoding="utf-8")
|
|
manager.stage_persistent_backup(checkpoint)
|
|
manager.activate_persistent_backup(checkpoint)
|
|
|
|
manager.rollback(checkpoint)
|
|
|
|
assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "old-runtime"
|
|
assert (backup_dir / "__init__.py").read_text(encoding="utf-8") == "old-backup"
|
|
|
|
|
|
def test_restore_checkpoint_derives_only_controlled_paths(monkeypatch, tmp_path):
|
|
"""
|
|
崩溃回放只按事务 ID 在受控根目录内重建文件引用。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(
|
|
"app.adapters.system.plugin.package.SystemUtils.is_docker",
|
|
lambda: True,
|
|
)
|
|
|
|
checkpoint = manager.restore_checkpoint(
|
|
plugin_id="DemoPlugin",
|
|
transaction_id="txn-5",
|
|
plugin_existed=True,
|
|
persistent_backup_existed=False,
|
|
)
|
|
|
|
assert checkpoint.transaction_dir == (
|
|
tmp_path / "config" / "plugin_transactions" / "txn-5"
|
|
)
|
|
assert checkpoint.backup_staging_dir == (
|
|
tmp_path / "config" / "plugins_backup" / ".demoplugin.staging-txn-5"
|
|
)
|
|
assert checkpoint.backup_previous_dir == (
|
|
tmp_path / "config" / "plugins_backup" / ".demoplugin.previous-txn-5"
|
|
)
|
|
|
|
|
|
def test_local_sync_failure_restores_previous_runtime_copy(monkeypatch, tmp_path):
|
|
"""
|
|
本地来源不可复制时不得丢失已经运行的插件副本。
|
|
"""
|
|
manager = _manager(monkeypatch, tmp_path)
|
|
plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
|
|
plugin_dir.mkdir(parents=True)
|
|
source_file = plugin_dir / "__init__.py"
|
|
source_file.write_text("stable", encoding="utf-8")
|
|
missing_source = tmp_path / "missing" / "demoplugin"
|
|
|
|
assert manager.sync_local("DemoPlugin", missing_source) is False
|
|
|
|
assert source_file.read_text(encoding="utf-8") == "stable"
|