Fix: Fix issue where temporary directory is not cleaned and error tracking false positives when repeatedly installing plugins (#8148)

* fix(star): 修复重复安装插件时临时目录未清理及错误追踪误报问题

- 引入 skip_failed_tracking 标志,当目标目录已存在时跳过失败安装追踪,避免将预期冲突错误误记为安装失败
- 修复 finally 块中临时目录清理逻辑,确保冲突场景下 plugin_upload_* 临时目录也能被正确删除
- 新增测试用例 test_install_plugin_from_file_conflict_keeps_failed_plugins_clean,验证重复安装同名插件时 failed_plugin_dict 为空且无残留临时目录

Ref #8122

* style(star): ruff 格式化
This commit is contained in:
NayukiMeko
2026-05-11 16:36:03 +08:00
committed by GitHub
parent ad516950f2
commit 041c35c35b
2 changed files with 50 additions and 8 deletions

View File

@@ -1798,6 +1798,7 @@ class PluginManager:
dir=self.plugin_store_path, prefix="plugin_upload_"
)
temp_desti_dir = desti_dir
skip_failed_tracking = False
try:
self.updator.unzip_file(zip_file_path, desti_dir)
@@ -1807,6 +1808,7 @@ class PluginManager:
metadata_dir_name,
)
if target_plugin_path != desti_dir and os.path.exists(target_plugin_path):
skip_failed_tracking = True
raise Exception(f"安装失败:目录 {metadata_dir_name} 已存在。")
if target_plugin_path != desti_dir:
os.rename(desti_dir, target_plugin_path)
@@ -1870,17 +1872,20 @@ class PluginManager:
return plugin_info
except Exception as e:
self._track_failed_install_dir(
dir_name=dir_name,
plugin_path=desti_dir,
error=e,
)
if not skip_failed_tracking:
self._track_failed_install_dir(
dir_name=dir_name,
plugin_path=desti_dir,
error=e,
)
logger.warning(
f"安装插件 {dir_name} 失败,插件安装目录:{desti_dir}",
)
raise
finally:
if temp_desti_dir != desti_dir and os.path.isdir(temp_desti_dir):
if (skip_failed_tracking or temp_desti_dir != desti_dir) and os.path.isdir(
temp_desti_dir
):
try:
remove_dir(temp_desti_dir)
except Exception as e:

View File

@@ -28,13 +28,15 @@ class MockStar:
self.info = {"repo": TEST_PLUGIN_REPO, "readme": ""}
def _write_local_test_plugin(plugin_path: Path, repo_url: str):
def _write_local_test_plugin(
plugin_path: Path, repo_url: str, version: str = "1.0.0"
):
"""Creates a minimal valid plugin structure."""
plugin_path.mkdir(parents=True, exist_ok=True)
metadata = {
"name": TEST_PLUGIN_NAME,
"repo": repo_url,
"version": "1.0.0",
"version": version,
"author": "AstrBot Team",
"desc": "Local test plugin",
"short_desc": "Local test short description",
@@ -386,6 +388,41 @@ async def test_install_plugin_from_file_dependency_install_flow(
assert ("load", TEST_PLUGIN_DIR) in events
@pytest.mark.asyncio
async def test_install_plugin_from_file_conflict_keeps_failed_plugins_clean(
plugin_manager_pm: PluginManager,
local_updator: Path,
monkeypatch,
tmp_path: Path,
):
zip_file_path = tmp_path / "plugin_upload_helloworld_v2.zip"
zip_file_path.write_text("placeholder", encoding="utf-8")
plugin_store_path = Path(plugin_manager_pm.plugin_store_path)
existing_upload_dirs = set(plugin_store_path.glob("plugin_upload_*"))
def mock_unzip_file(zip_path: str, target_dir: str) -> None:
assert zip_path == str(zip_file_path)
_write_local_test_plugin(
Path(target_dir),
TEST_PLUGIN_REPO,
version="2.0.0",
)
assert local_updator.is_dir()
monkeypatch.setattr(plugin_manager_pm.updator, "unzip_file", mock_unzip_file)
with pytest.raises(Exception, match=f"安装失败:目录 {TEST_PLUGIN_DIR} 已存在。"):
await plugin_manager_pm.install_plugin_from_file(str(zip_file_path))
new_upload_dirs = [
upload_dir
for upload_dir in plugin_store_path.glob("plugin_upload_*")
if upload_dir not in existing_upload_dirs
]
assert plugin_manager_pm.failed_plugin_dict == {}
assert new_upload_dirs == []
@pytest.mark.asyncio
@pytest.mark.parametrize("dependency_install_fails", [False, True])
async def test_reload_failed_plugin_dependency_install_flow(