mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
"""
|
|
阶段 4 六个重点 Chain 的纵向切片数量守卫。
|
|
"""
|
|
|
|
import inspect
|
|
|
|
import pytest
|
|
|
|
from app.chain.download.facade import DownloadChain
|
|
from app.chain.media.facade import MediaChain
|
|
from app.chain.message import MessageChain
|
|
from app.chain.search.facade import SearchChain
|
|
from app.chain.subscribe.facade import SubscribeChain
|
|
from app.chain.transfer.facade import TransferChain
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("chain_type", "method_tokens"),
|
|
[
|
|
(
|
|
SubscribeChain,
|
|
{
|
|
"exists": "_subscription_query",
|
|
"get_subscribe_by_source": "_subscription_query",
|
|
"has_music_subscribe": "_subscription_query",
|
|
},
|
|
),
|
|
(
|
|
SearchChain,
|
|
{
|
|
"save_last_search_params": "_search_state",
|
|
"last_search_params": "SearchCacheOwner",
|
|
"last_search_results": "SearchCacheOwner",
|
|
},
|
|
),
|
|
(
|
|
TransferChain,
|
|
{
|
|
"put_to_queue": "_transfer_queue_service",
|
|
"remove_from_queue": "_transfer_queue_service",
|
|
"get_queue_tasks": "_transfer_queue_service",
|
|
},
|
|
),
|
|
(
|
|
DownloadChain,
|
|
{
|
|
"downloading": "_download_task_service",
|
|
"set_downloading": "_download_task_service",
|
|
"remove_downloading": "_download_task_service",
|
|
},
|
|
),
|
|
(
|
|
MediaChain,
|
|
{
|
|
"normalize_music_candidates": "MusicCatalogService",
|
|
"search": "_music_catalog",
|
|
"async_search": "_music_catalog",
|
|
"search_music": "self.search",
|
|
"async_search_music": "self.async_search",
|
|
},
|
|
),
|
|
(
|
|
MessageChain,
|
|
{
|
|
"remote_clear_session": "_message_session_service",
|
|
"remote_stop_agent": "_message_session_service",
|
|
"remote_session_status": "_message_session_service",
|
|
},
|
|
),
|
|
],
|
|
)
|
|
def test_key_chain_keeps_three_application_service_slices(
|
|
chain_type: type,
|
|
method_tokens: dict[str, str],
|
|
) -> None:
|
|
"""
|
|
每个重点 Chain 至少三个公开方法必须继续委托窄应用服务。
|
|
"""
|
|
assert len(method_tokens) >= 3
|
|
missing = []
|
|
for method_name, service_token in method_tokens.items():
|
|
method = getattr(chain_type, method_name)
|
|
if service_token not in inspect.getsource(method):
|
|
missing.append(f"{chain_type.__name__}.{method_name}->{service_token}")
|
|
assert missing == []
|