mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
123 lines
5.4 KiB
Python
123 lines
5.4 KiB
Python
"""
|
|
验证存储项缺少路径或名称时不会访问远端、删除本地数据或构造错误目标。
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from app.modules.filemanager.storages import StorageBase
|
|
from app.modules.filemanager.storages.alipan import AliPan
|
|
from app.modules.filemanager.storages.alist import Alist
|
|
from app.modules.filemanager.storages.local import LocalStorage
|
|
from app.modules.filemanager.storages.rclone import Rclone
|
|
from app.modules.filemanager.storages.smb import SMB
|
|
from app.schemas.file import FileItem
|
|
|
|
|
|
@pytest.mark.parametrize("storage_class", [Alist, LocalStorage, Rclone, SMB])
|
|
@pytest.mark.parametrize("missing_path", [None, ""])
|
|
def test_missing_path_operations_stop_before_io(storage_class, missing_path, tmp_path, monkeypatch):
|
|
"""
|
|
空路径项不能被当作当前目录或远端根目录执行查询、传输、重命名及删除。
|
|
"""
|
|
storage = storage_class()
|
|
item = FileItem(storage=storage.schema.value, type="file", path=missing_path)
|
|
query = Mock(side_effect=AssertionError("空路径不应触发查询"))
|
|
monkeypatch.setattr(storage, "get_item", query)
|
|
if isinstance(storage, SMB):
|
|
monkeypatch.setattr(storage, "_check_connection", Mock())
|
|
assert storage.detail(item) is None
|
|
assert storage.delete(item) is False
|
|
assert storage.rename(item, "renamed.mkv") is False
|
|
assert storage.upload(item, tmp_path / "source.mkv") is None
|
|
if isinstance(storage, LocalStorage):
|
|
with pytest.raises(ValueError, match="缺少路径"):
|
|
storage.download(item, tmp_path)
|
|
assert storage.copy(item, tmp_path, "copy.mkv") is False
|
|
assert storage.move(item, tmp_path, "move.mkv") is False
|
|
assert storage.link(item, tmp_path / "hardlink.mkv") is False
|
|
assert storage.softlink(item, tmp_path / "symlink.mkv") is False
|
|
else:
|
|
assert storage.download(item, tmp_path) is None
|
|
if isinstance(storage, (Alist, Rclone)):
|
|
assert storage.list(item) == []
|
|
assert storage.create_folder(item, "child") is None
|
|
assert storage.copy(item, tmp_path, "copy.mkv") is False
|
|
assert storage.move(item, tmp_path, "move.mkv") is False
|
|
if isinstance(storage, Alist):
|
|
assert storage.get_parent(item) is None
|
|
assert storage.copy_item(item, tmp_path, "copy.mkv") is None
|
|
query.assert_not_called()
|
|
assert list(tmp_path.iterdir()) == []
|
|
|
|
|
|
@pytest.mark.parametrize("storage_class", [AliPan, Alist, Rclone, SMB])
|
|
def test_cloud_backends_report_unsupported_links(storage_class, tmp_path):
|
|
"""
|
|
不支持链接的存储明确返回失败，调用方不能把空返回值误判为已完成。
|
|
"""
|
|
storage = storage_class()
|
|
item = FileItem(storage=storage.schema.value, type="file", path="/source.mkv")
|
|
assert storage.softlink(item, tmp_path / "symlink.mkv") is False
|
|
if not isinstance(storage, SMB):
|
|
assert storage.link(item, tmp_path / "hardlink.mkv") is False
|
|
assert list(tmp_path.iterdir()) == []
|
|
|
|
|
|
def test_alipan_incomplete_metadata_does_not_query_or_create(monkeypatch):
|
|
"""
|
|
阿里云盘的目录创建、详情及复制拒绝缺失必要元数据的输入。
|
|
"""
|
|
storage = AliPan()
|
|
request = Mock(side_effect=AssertionError("不完整文件项不应触发 API"))
|
|
monkeypatch.setattr(storage, "_request_api", request)
|
|
query = Mock(side_effect=AssertionError("不完整文件项不应查询目标"))
|
|
monkeypatch.setattr(storage, "get_item", query)
|
|
item = FileItem(storage=storage.schema.value, type="file", path=None)
|
|
assert storage.detail(item) is None
|
|
assert storage.create_folder(item, "child") is None
|
|
assert storage.copy(item, Path("/target"), "copy.mkv") is False
|
|
request.assert_not_called()
|
|
query.assert_not_called()
|
|
|
|
|
|
def test_missing_snapshot_path_preserves_previous_files(monkeypatch):
|
|
"""
|
|
缺少目录路径的快照输入保留旧文件索引，避免把未查询到的数据清成空目录。
|
|
"""
|
|
storage = LocalStorage()
|
|
item = FileItem(storage=storage.schema.value, type="dir", path=None)
|
|
files = {"/keep.mkv": {"size": 123}}
|
|
children = {}
|
|
previous = {}
|
|
query = Mock(side_effect=AssertionError("空路径不应遍历目录"))
|
|
monkeypatch.setattr(storage, "_snapshot_list", query)
|
|
assert StorageBase.get_parent(storage, item) is None
|
|
StorageBase._snapshot_remove_deleted_children(storage, item, [], files, children, previous)
|
|
StorageBase._snapshot_file(storage, item, files, children, previous, None, 5)
|
|
assert files == {"/keep.mkv": {"size": 123}}
|
|
query.assert_not_called()
|
|
|
|
|
|
def test_rclone_config_without_content_writes_empty_file(tmp_path, monkeypatch):
|
|
"""
|
|
缺省配置内容保存为空文本，未提供路径时不会尝试打开文件。
|
|
"""
|
|
storage = Rclone()
|
|
monkeypatch.setattr(StorageBase, "set_config", Mock())
|
|
storage.set_config({})
|
|
config_path = tmp_path / "nested" / "rclone.conf"
|
|
storage.set_config({"filepath": str(config_path), "content": None})
|
|
assert config_path.read_text(encoding="utf-8") == ""
|
|
|
|
|
|
@pytest.mark.parametrize("payload", [{}, {"Name": None}, {"Name": 123}])
|
|
def test_rclone_listing_rejects_item_without_name(payload):
|
|
"""
|
|
远端列表中的无名项不能被投影成当前目录或名称为 None 的文件。
|
|
"""
|
|
with pytest.raises(ValueError, match="缺少名称"):
|
|
Rclone()._Rclone__get_rcloneitem(payload)
|