mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 11:54:50 +08:00
212 lines
7.7 KiB
Python
212 lines
7.7 KiB
Python
"""
|
||
验证 OpenList 自动分页、目录可见性和文件读写边界。
|
||
"""
|
||
|
||
import hashlib
|
||
from pathlib import Path
|
||
from tempfile import TemporaryDirectory
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from app.modules.filemanager.storages import alist as alist_module
|
||
from app.modules.filemanager.storages.alist import Alist
|
||
from app.schemas.file import FileItem
|
||
|
||
|
||
@pytest.fixture
|
||
def storage() -> Alist:
|
||
"""
|
||
为每个用例创建独立的 OpenList 存储实例，HTTP 请求由用例自行替换。
|
||
"""
|
||
return Alist()
|
||
|
||
|
||
class _FakeResponse:
|
||
"""
|
||
提供预设 OpenList JSON 响应与状态码，不执行真实请求。
|
||
"""
|
||
|
||
def __init__(self, payload: dict, status_code: int = 200):
|
||
"""
|
||
保存预设响应载荷和 HTTP 状态码。
|
||
"""
|
||
self._payload = payload
|
||
self.status_code = status_code
|
||
|
||
def json(self):
|
||
"""
|
||
返回预设 JSON 响应正文。
|
||
"""
|
||
return self._payload
|
||
|
||
|
||
def _dir_item(path: str = "/"):
|
||
"""
|
||
构造指定路径的 OpenList 目录项，默认使用根目录。
|
||
"""
|
||
return FileItem(storage="alist", type="dir", path=path)
|
||
|
||
|
||
def _page_payload(start: int, count: int, total: int) -> dict:
|
||
"""
|
||
构造连续目录项及总数的分页响应，用于验证自动翻页。
|
||
"""
|
||
return {
|
||
"code": 200,
|
||
"message": "success",
|
||
"data": {
|
||
"content": [
|
||
{
|
||
"name": f"dir-{index}",
|
||
"size": 0,
|
||
"is_dir": True,
|
||
"modified": "2024-05-17T13:47:55.4174917+08:00",
|
||
"thumb": "",
|
||
}
|
||
for index in range(start, start + count)
|
||
],
|
||
"total": total,
|
||
},
|
||
}
|
||
|
||
|
||
def test_list_fetches_all_pages_when_per_page_is_default(storage) -> None:
|
||
"""
|
||
使用默认分页大小时遍历全部页并合并目录项。
|
||
"""
|
||
responses = [
|
||
_FakeResponse(_page_payload(0, 500, 505)),
|
||
_FakeResponse(_page_payload(500, 5, 505)),
|
||
]
|
||
request_utils = MagicMock()
|
||
request_utils.post_res.side_effect = responses
|
||
|
||
with patch.object(Alist, "get_conf", return_value={"url": "http://openlist.test", "token": "token"}):
|
||
with patch.object(alist_module, "RequestUtils", return_value=request_utils):
|
||
items = storage.list(_dir_item())
|
||
|
||
assert 505 == len(items)
|
||
assert "/dir-0/" == items[0].path
|
||
assert "/dir-504/" == items[-1].path
|
||
assert 2 == request_utils.post_res.call_count
|
||
assert 1 == request_utils.post_res.call_args_list[0].kwargs["json"]["page"]
|
||
assert 2 == request_utils.post_res.call_args_list[1].kwargs["json"]["page"]
|
||
assert 500 == request_utils.post_res.call_args_list[0].kwargs["json"]["per_page"]
|
||
assert 500 == request_utils.post_res.call_args_list[1].kwargs["json"]["per_page"]
|
||
|
||
|
||
def test_list_respects_explicit_per_page_without_auto_paging(storage) -> None:
|
||
"""
|
||
显式指定每页数量时只请求一页，不触发自动翻页。
|
||
"""
|
||
request_utils = MagicMock()
|
||
request_utils.post_res.return_value = _FakeResponse(_page_payload(0, 50, 205))
|
||
|
||
with patch.object(Alist, "get_conf", return_value={"url": "http://openlist.test", "token": "token"}):
|
||
with patch.object(alist_module, "RequestUtils", return_value=request_utils):
|
||
items = storage.list(_dir_item(), per_page=50)
|
||
|
||
assert 50 == len(items)
|
||
assert 1 == request_utils.post_res.call_count
|
||
assert 50 == request_utils.post_res.call_args.kwargs["json"]["per_page"]
|
||
|
||
|
||
def test_create_folder_returns_target_when_openlist_metadata_is_delayed(storage) -> None:
|
||
"""
|
||
OpenList 创建目录成功但元数据延迟可见时，应返回可用的目标目录项。
|
||
"""
|
||
request_utils = MagicMock()
|
||
request_utils.post_res.return_value = _FakeResponse({"code": 200, "message": "success", "data": None})
|
||
|
||
with patch.object(Alist, "get_conf", return_value={"url": "http://openlist.test", "token": "token"}):
|
||
with patch.object(storage, "_Alist__get_header_with_token", return_value={}):
|
||
with patch.object(alist_module, "RequestUtils", return_value=request_utils):
|
||
with patch.object(storage, "_delay_get_item", return_value=None):
|
||
folder = storage.create_folder(
|
||
_dir_item("/library/Test Show (2026)"),
|
||
"Season 1",
|
||
)
|
||
|
||
assert folder is not None
|
||
assert "/library/Test Show (2026)/Season 1/" == folder.path
|
||
assert "alist" == folder.storage
|
||
assert "dir" == folder.type
|
||
|
||
|
||
def test_move_item_returns_target_when_openlist_metadata_is_delayed(storage) -> None:
|
||
"""
|
||
OpenList 操作成功但目标元数据延迟可见时，应返回可用的目标文件项。
|
||
"""
|
||
source = FileItem(
|
||
storage="alist",
|
||
type="file",
|
||
path="/downloads/Test.Show.S01E01.mkv",
|
||
name="Test.Show.S01E01.mkv",
|
||
basename="Test.Show.S01E01",
|
||
extension="mkv",
|
||
size=1024,
|
||
modify_time=1715939275.0,
|
||
)
|
||
request_utils = MagicMock()
|
||
request_utils.post_res.return_value = _FakeResponse({"code": 200, "message": "success", "data": None})
|
||
|
||
with patch.object(Alist, "get_conf", return_value={"url": "http://openlist.test", "token": "token"}):
|
||
with patch.object(storage, "_Alist__get_header_with_token", return_value={}):
|
||
with patch.object(alist_module, "RequestUtils", return_value=request_utils):
|
||
with patch.object(storage, "_delay_get_item", return_value=None):
|
||
target = storage.move_item(
|
||
source,
|
||
Path("/library/Test Show (2026)/Season 1"),
|
||
"Test.Show.S01E01.mkv",
|
||
)
|
||
|
||
assert target is not None
|
||
assert "/library/Test Show (2026)/Season 1/Test.Show.S01E01.mkv" == target.path
|
||
assert "alist" == target.storage
|
||
assert "file" == target.type
|
||
assert 1024 == target.size
|
||
|
||
|
||
def test_upload_sends_hash_headers_for_rapid_upload(storage) -> None:
|
||
"""
|
||
OpenList 上传应附带文件哈希头，供服务端尝试秒传。
|
||
"""
|
||
with TemporaryDirectory() as temp_dir:
|
||
local_path = Path(temp_dir) / "rapid.bin"
|
||
content = b"moviepilot-openlist-rapid-upload"
|
||
local_path.write_bytes(content)
|
||
upload_dir = FileItem(
|
||
storage="alist",
|
||
type="dir",
|
||
path="/library/",
|
||
name="library",
|
||
basename="library",
|
||
)
|
||
uploaded_item = FileItem(
|
||
storage="alist",
|
||
type="file",
|
||
path="/library/rapid.bin",
|
||
name="rapid.bin",
|
||
basename="rapid",
|
||
extension="bin",
|
||
size=len(content),
|
||
)
|
||
request_utils = MagicMock()
|
||
request_utils.put_res.return_value = _FakeResponse({"code": 200, "message": "success", "data": None})
|
||
|
||
with patch.object(Alist, "get_conf", return_value={"url": "http://openlist.test", "token": "token"}):
|
||
with patch.object(alist_module, "RequestUtils", return_value=request_utils) as request_utils_factory:
|
||
with patch.object(storage, "_delay_get_item", return_value=uploaded_item):
|
||
result = storage.upload(upload_dir, local_path)
|
||
|
||
assert uploaded_item == result
|
||
request_utils.put_res.assert_called_once()
|
||
headers = request_utils_factory.call_args.kwargs["headers"]
|
||
assert hashlib.md5(content).hexdigest() == headers["X-File-Md5"]
|
||
assert hashlib.sha1(content).hexdigest() == headers["X-File-Sha1"]
|
||
assert hashlib.sha256(content).hexdigest() == headers["X-File-Sha256"]
|
||
assert str(len(content)) == headers["Content-Length"]
|
||
assert "application/octet-stream" == headers["Content-Type"]
|
||
assert "/library/rapid.bin" == headers["File-Path"]
|