mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 11:27:43 +08:00
345 lines
10 KiB
Python
345 lines
10 KiB
Python
"""
|
||
验证插件目录来源查询与已安装插件信息汇总。
|
||
"""
|
||
|
||
import asyncio
|
||
from unittest.mock import Mock
|
||
|
||
import pytest
|
||
from packaging.version import Version
|
||
|
||
from app.application.plugin.catalog import PluginCatalogService
|
||
from app.runtime.extensions.plugin.metadata import PluginMetadataMapper
|
||
from app.schemas.plugin import Plugin
|
||
|
||
|
||
def _plugin(
|
||
plugin_id: str,
|
||
version: str,
|
||
repo_url: str,
|
||
package_version: str | None = "v3",
|
||
) -> Plugin:
|
||
"""
|
||
构造目录合并测试使用的最小插件 DTO。
|
||
"""
|
||
plugin = Plugin(
|
||
id=plugin_id,
|
||
plugin_version=version,
|
||
repo_url=repo_url,
|
||
)
|
||
plugin.package_version = package_version
|
||
return plugin
|
||
|
||
|
||
def _service(**overrides) -> PluginCatalogService:
|
||
"""
|
||
构造完全依赖内存假对象的插件目录应用服务。
|
||
"""
|
||
defaults = {
|
||
"market_loader": Mock(return_value={}),
|
||
"async_market_loader": Mock(),
|
||
"installed_plugins_provider": Mock(return_value=[]),
|
||
"plugin_mapper": Mock(),
|
||
"is_local_repo": lambda value: str(value).startswith("local://"),
|
||
"version_compare": (
|
||
lambda left, operator, right:
|
||
operator == ">" and Version(left) > Version(right)
|
||
),
|
||
"warning": Mock(),
|
||
"error": Mock(),
|
||
}
|
||
defaults.update(overrides)
|
||
return PluginCatalogService(**defaults)
|
||
|
||
|
||
def test_merge_prefers_higher_generation_over_same_base_entry():
|
||
"""
|
||
高代际索引出现同 ID 同版本时不再保留基础索引副本。
|
||
"""
|
||
service = _service()
|
||
higher = _plugin("Demo", "2.0.0", "https://market-a")
|
||
base = _plugin("Demo", "2.0.0", "https://market-b")
|
||
|
||
result = service.merge([higher], [base], ["https://market-a", "https://market-b"])
|
||
|
||
assert result == [higher]
|
||
|
||
|
||
def test_merge_prefers_newer_version_and_remote_source():
|
||
"""
|
||
相同插件保留最高版本，同版本时市场来源覆盖本地副本。
|
||
"""
|
||
service = _service()
|
||
old_remote = _plugin("Demo", "1.0.0", "https://market-a")
|
||
new_local = _plugin("Demo", "2.0.0", "local://Demo")
|
||
new_remote = _plugin("Demo", "2.0.0", "https://market-b")
|
||
|
||
result = service.merge(
|
||
[old_remote, new_local, new_remote],
|
||
[],
|
||
["https://market-a", "https://market-b"],
|
||
)
|
||
|
||
assert result == [new_remote]
|
||
|
||
|
||
def test_merge_treats_plugin_id_casing_as_one_physical_plugin():
|
||
"""
|
||
同一物理插件的市场与本地 ID 大小写差异不能产生两个目录条目。
|
||
"""
|
||
service = _service()
|
||
online = _plugin("DownloadCenter", "3.2.1", "https://market-a")
|
||
local = _plugin("downloadcenter", "3.3.2", "local://DownloadCenter")
|
||
|
||
result = service.merge(
|
||
[online, local],
|
||
[],
|
||
["https://market-a"],
|
||
)
|
||
|
||
assert result == [local]
|
||
|
||
|
||
def test_merge_by_source_deduplicates_plugin_id_casing_within_generation():
|
||
"""
|
||
同一仓库和代际内的 ID 大小写差异不能产生重复候选。
|
||
"""
|
||
service = _service()
|
||
old = _plugin("DownloadCenter", "3.2.1", "https://market-a")
|
||
new = _plugin("downloadcenter", "3.3.2", "https://market-a")
|
||
|
||
result = service.merge_by_source(
|
||
[old, new],
|
||
[],
|
||
["https://market-a"],
|
||
)
|
||
|
||
assert result == [new]
|
||
|
||
|
||
def test_merge_by_source_preserves_candidates_across_generations():
|
||
"""
|
||
跨代际候选必须保留，由来源准入按代际优先级继续决策。
|
||
"""
|
||
service = _service()
|
||
v3 = _plugin("DownloadCenter", "3.2.1", "https://market-a", "v3")
|
||
v2 = _plugin("downloadcenter", "9.0.0", "https://market-a", "v2")
|
||
|
||
result = service.merge_by_source(
|
||
[v3, v2],
|
||
[],
|
||
["https://market-a"],
|
||
)
|
||
|
||
assert result == [v3, v2]
|
||
|
||
|
||
def test_load_maps_market_entries_with_installed_snapshot():
|
||
"""
|
||
单市场读取只获取一次已安装快照并按索引顺序映射 DTO。
|
||
"""
|
||
mapper = Mock(side_effect=lambda plugin_id, *_args: plugin_id)
|
||
installed_provider = Mock(return_value=["Installed"])
|
||
service = _service(
|
||
market_loader=Mock(return_value={"First": {}, "Second": {}}),
|
||
installed_plugins_provider=installed_provider,
|
||
plugin_mapper=mapper,
|
||
)
|
||
|
||
result = service.load("https://market-a", "v3", True)
|
||
|
||
assert result == ["First", "Second"]
|
||
installed_provider.assert_called_once_with()
|
||
assert mapper.call_args_list[0].args[3:] == (["Installed"], 2, "v3")
|
||
assert mapper.call_args_list[1].args[3:] == (["Installed"], 1, "v3")
|
||
|
||
|
||
def test_metadata_mapper_preserves_package_generation_without_serializing_it():
|
||
"""
|
||
市场映射必须保留候选代际，同时不能扩张插件 API 响应。
|
||
"""
|
||
mapper = PluginMetadataMapper(
|
||
plugin_instance=lambda _plugin_id: None,
|
||
plugin_class=lambda _plugin_id: None,
|
||
annotate_system_version=lambda info: info,
|
||
annotate_runtime_compatibility=lambda info: info,
|
||
is_package_compatible=lambda _info, _version: True,
|
||
auth_checker=lambda _plugin, _source: True,
|
||
version_compare=lambda _left, _operator, _right: False,
|
||
log=Mock(),
|
||
)
|
||
|
||
plugin = mapper.map(
|
||
plugin_id="Demo",
|
||
plugin_info={"name": "Demo", "version": "3.0.0"},
|
||
market="https://market-a",
|
||
installed_plugins=[],
|
||
add_time=1,
|
||
package_version="v3",
|
||
)
|
||
|
||
assert plugin is not None
|
||
assert plugin.package_version == "v3"
|
||
assert "package_version" not in plugin.model_dump()
|
||
assert "package_version" not in Plugin.model_json_schema()["properties"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_collect_isolates_failure_and_completes_progress():
|
||
"""
|
||
异步市场单任务失败时保留成功结果，并把进度推进到完成态。
|
||
"""
|
||
progress = Mock()
|
||
error = Mock()
|
||
service = _service(error=error)
|
||
|
||
async def loader(market: str, package_version: str | None, _force: bool):
|
||
"""
|
||
模拟一个失败代际和其余可正常完成的市场请求。
|
||
"""
|
||
await asyncio.sleep(0)
|
||
if market == "https://market-a" and package_version == "v3":
|
||
raise RuntimeError("unavailable")
|
||
version = "2.0.0" if package_version else "1.0.0"
|
||
plugin_id = "MarketA" if market.endswith("market-a") else "MarketB"
|
||
return [_plugin(plugin_id, version, market)]
|
||
|
||
result = await service.async_collect(
|
||
markets=["https://market-a", "https://market-b"],
|
||
compatible_flags=["v3"],
|
||
force=True,
|
||
loader=loader,
|
||
progress_callback=progress,
|
||
)
|
||
|
||
assert {plugin.id for plugin in result} == {
|
||
"MarketA",
|
||
"MarketB",
|
||
}
|
||
error.assert_called_once()
|
||
assert progress.call_args_list[0].kwargs["value"] == 0
|
||
assert progress.call_args_list[-1].kwargs["value"] == 100
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_collect_preserves_each_repository_update_candidate():
|
||
"""
|
||
来源准入读取应保留每个仓库的最高版本，不能先按插件 ID 全局去重。
|
||
"""
|
||
service = _service()
|
||
|
||
async def loader(market: str, package_version: str | None, _force: bool):
|
||
"""
|
||
按仓库返回不同版本，验证来源保留模式不会提前跨仓去重。
|
||
"""
|
||
if package_version is None:
|
||
return []
|
||
version = "2.0.0" if market == "https://market-bound" else "3.0.0"
|
||
return [_plugin("Demo", version, market)]
|
||
|
||
result = await service.async_collect(
|
||
markets=["https://market-bound", "https://market-alternative"],
|
||
compatible_flags=["v3"],
|
||
force=False,
|
||
loader=loader,
|
||
preserve_sources=True,
|
||
)
|
||
|
||
assert [(plugin.repo_url, plugin.plugin_version) for plugin in result] == [
|
||
("https://market-bound", "2.0.0"),
|
||
("https://market-alternative", "3.0.0"),
|
||
]
|
||
assert service.merge(result, [], ["https://market-bound", "https://market-alternative"]) == [result[1]]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_collect_cancels_all_loaders_when_parent_is_cancelled():
|
||
"""
|
||
请求取消时必须取消并回收全部市场 loader，不能把子任务遗留在事件循环。
|
||
"""
|
||
service = _service()
|
||
blocker = asyncio.Event()
|
||
all_started = asyncio.Event()
|
||
started = 0
|
||
cancelled = 0
|
||
|
||
async def loader(_market: str, _package_version: str | None, _force: bool):
|
||
"""
|
||
记录市场 loader 的启动和取消，并等待测试释放。
|
||
"""
|
||
nonlocal started, cancelled
|
||
started += 1
|
||
if started == 2:
|
||
all_started.set()
|
||
try:
|
||
await blocker.wait()
|
||
except asyncio.CancelledError:
|
||
cancelled += 1
|
||
raise
|
||
return []
|
||
|
||
collect_task = asyncio.create_task(
|
||
service.async_collect(
|
||
markets=["https://market-a", "https://market-b"],
|
||
compatible_flags=[],
|
||
force=True,
|
||
loader=loader,
|
||
)
|
||
)
|
||
await asyncio.wait_for(all_started.wait(), timeout=1)
|
||
collect_task.cancel()
|
||
try:
|
||
with pytest.raises(asyncio.CancelledError):
|
||
await collect_task
|
||
await asyncio.sleep(0)
|
||
assert cancelled == 2
|
||
finally:
|
||
blocker.set()
|
||
await asyncio.sleep(0)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_collect_cleans_loaders_when_progress_callback_fails():
|
||
"""
|
||
进度回调异常也必须取消并回收尚未完成的市场 loader。
|
||
"""
|
||
service = _service()
|
||
blocker = asyncio.Event()
|
||
slow_started = asyncio.Event()
|
||
slow_cancelled = asyncio.Event()
|
||
|
||
async def loader(market: str, _package_version: str | None, _force: bool):
|
||
"""
|
||
让一个市场立即完成，另一个保持阻塞以验证异常清理。
|
||
"""
|
||
if market == "https://market-a":
|
||
await slow_started.wait()
|
||
return []
|
||
slow_started.set()
|
||
try:
|
||
await blocker.wait()
|
||
except asyncio.CancelledError:
|
||
slow_cancelled.set()
|
||
raise
|
||
return []
|
||
|
||
def progress(*, value: float, **_kwargs) -> None:
|
||
"""
|
||
首个市场完成时模拟进度消费者失败。
|
||
"""
|
||
if value > 0:
|
||
raise RuntimeError("progress unavailable")
|
||
|
||
with pytest.raises(RuntimeError, match="progress unavailable"):
|
||
await asyncio.wait_for(
|
||
service.async_collect(
|
||
markets=["https://market-a", "https://market-b"],
|
||
compatible_flags=[],
|
||
force=True,
|
||
loader=loader,
|
||
progress_callback=progress,
|
||
),
|
||
timeout=1,
|
||
)
|
||
assert slow_cancelled.is_set()
|