mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-08-15 11:04:12 +08:00
test: 补充 AListGo 与特典整理回归验证
This commit is contained in:
@@ -4,7 +4,7 @@ from app.schemas.types import StorageSchema
|
||||
|
||||
class AlistGo(Alist):
|
||||
"""
|
||||
AList相关操作
|
||||
AListGo 相关操作
|
||||
|
||||
API 文档:https://docs.alistgo.com/
|
||||
"""
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from typing import Generator
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
@@ -9,34 +10,42 @@ from app.schemas.types import StorageSchema
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
def __init__(self, payload: dict, status_code: int = 200):
|
||||
"""模拟 AList 登录接口响应"""
|
||||
|
||||
def __init__(self, payload: dict, status_code: int = 200) -> None:
|
||||
"""保存模拟响应数据和状态码"""
|
||||
self._payload = payload
|
||||
self.status_code = status_code
|
||||
|
||||
def json(self):
|
||||
def json(self) -> dict:
|
||||
"""返回模拟 JSON 响应数据"""
|
||||
return self._payload
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def clear_token_cache():
|
||||
def clear_token_cache() -> Generator[None, None, None]:
|
||||
"""在测试前后清理 AList 登录令牌缓存"""
|
||||
Alist._Alist__login_token.cache_clear() # noqa
|
||||
yield
|
||||
Alist._Alist__login_token.cache_clear() # noqa
|
||||
|
||||
|
||||
def test_alistgo_schema_registered():
|
||||
def test_alistgo_schema_registered() -> None:
|
||||
"""AListGo 存储类型应注册到存储枚举"""
|
||||
assert AlistGo.schema == StorageSchema.AlistGo
|
||||
assert StorageSchema.AlistGo.value == "alistgo"
|
||||
|
||||
|
||||
def test_alistgo_singleton_isolated_from_alist():
|
||||
def test_alistgo_singleton_isolated_from_alist() -> None:
|
||||
"""AListGo 与 OpenList 应使用彼此独立的存储实例"""
|
||||
alist = Alist()
|
||||
alistgo = AlistGo()
|
||||
assert alistgo is not alist
|
||||
assert isinstance(alistgo, Alist)
|
||||
|
||||
|
||||
def test_alistgo_token_isolated_from_alist(clear_token_cache):
|
||||
def test_alistgo_token_isolated_from_alist(clear_token_cache) -> None:
|
||||
"""AListGo 与 OpenList 应按各自凭据缓存登录令牌"""
|
||||
def _conf(storage):
|
||||
return {
|
||||
"url": f"http://{storage.schema.value}.test",
|
||||
@@ -63,7 +72,8 @@ def test_alistgo_token_isolated_from_alist(clear_token_cache):
|
||||
assert request_utils.post_res.call_count == 2
|
||||
|
||||
|
||||
def test_init_storage_keeps_other_storage_token(clear_token_cache):
|
||||
def test_init_storage_keeps_other_storage_token(clear_token_cache) -> None:
|
||||
"""重新初始化单个存储时不应清除另一存储的登录令牌"""
|
||||
def _conf(storage):
|
||||
return {
|
||||
"url": f"http://{storage.schema.value}.test",
|
||||
|
||||
73
tests/test_transhandler_special_extra.py
Normal file
73
tests/test_transhandler_special_extra.py
Normal file
@@ -0,0 +1,73 @@
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.context import MediaInfo
|
||||
from app.core.meta import MetaVideo
|
||||
from app.modules.filemanager.transhandler import TransHandler
|
||||
from app.schemas import FileItem, TransferInfo
|
||||
from app.schemas.types import MediaType
|
||||
|
||||
|
||||
def _transfer_without_episode(file_name: str) -> tuple[TransferInfo, MagicMock, MagicMock]:
|
||||
"""整理一个未识别出集数的剧集视频文件"""
|
||||
fileitem = FileItem(
|
||||
storage="local",
|
||||
path=f"/downloads/{file_name}",
|
||||
type="file",
|
||||
name=file_name,
|
||||
basename=file_name.removesuffix(".mkv"),
|
||||
extension="mkv",
|
||||
size=1024,
|
||||
)
|
||||
meta = MetaVideo(file_name)
|
||||
meta.begin_episode = None
|
||||
mediainfo = MediaInfo()
|
||||
mediainfo.type = MediaType.TV
|
||||
source_oper = MagicMock()
|
||||
target_oper = MagicMock()
|
||||
|
||||
result = TransHandler().transfer_media(
|
||||
fileitem=fileitem,
|
||||
in_meta=meta,
|
||||
mediainfo=mediainfo,
|
||||
target_storage="local",
|
||||
target_path=Path("/library"),
|
||||
transfer_type="copy",
|
||||
source_oper=source_oper,
|
||||
target_oper=target_oper,
|
||||
need_notify=True,
|
||||
)
|
||||
return result, source_oper, target_oper
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"file_name",
|
||||
[
|
||||
"Test.Show.NCOP.mkv",
|
||||
"Test.Show.Menu.mkv",
|
||||
"Test.Show.Featurette.mkv",
|
||||
],
|
||||
)
|
||||
def test_tv_special_extra_without_episode_skips_transfer(file_name: str) -> None:
|
||||
"""未识别出集数的已知特典文件应静默跳过并返回成功"""
|
||||
result, source_oper, target_oper = _transfer_without_episode(file_name)
|
||||
|
||||
assert result.success is True
|
||||
assert result.need_notify is False
|
||||
assert result.fileitem.path == f"/downloads/{file_name}"
|
||||
assert source_oper.mock_calls == []
|
||||
assert target_oper.mock_calls == []
|
||||
|
||||
|
||||
def test_regular_tv_file_without_episode_remains_failure() -> None:
|
||||
"""普通剧集视频未识别出集数时仍应返回失败并保留通知"""
|
||||
result, source_oper, target_oper = _transfer_without_episode("Test.Show.Bonus.mkv")
|
||||
|
||||
assert result.success is False
|
||||
assert result.message == "未识别到文件集数"
|
||||
assert result.need_notify is True
|
||||
assert result.fail_list == ["/downloads/Test.Show.Bonus.mkv"]
|
||||
assert source_oper.mock_calls == []
|
||||
assert target_oper.mock_calls == []
|
||||
Reference in New Issue
Block a user