mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-01 21:07:01 +08:00
129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""
|
||
Chain 本地限流行为测试。
|
||
"""
|
||
|
||
# 可选下载器模块必须在导入 Chain 前完成隔离。
|
||
# ruff: noqa: E402
|
||
|
||
from unittest.mock import Mock
|
||
|
||
import pytest
|
||
|
||
from app.testing.bootstrap import ensure_optional_stub
|
||
|
||
# 仅在可选依赖缺失时补占位；已安装时保留真实模块，避免污染同一进程里导入真实类型的用例。
|
||
ensure_optional_stub("qbittorrentapi", TorrentFilesList=list)
|
||
ensure_optional_stub("transmission_rpc", File=object)
|
||
|
||
from app.application.chain.context import ChainRuntimeContext
|
||
from app.chain.base import ChainBase
|
||
from app.runtime.extensions.module.dispatcher import ModuleInvocationDispatcher
|
||
from app.schemas.exception import RateLimitExceededException
|
||
|
||
|
||
class _LimitedModule:
|
||
"""
|
||
模拟始终触发本地限流的宿主模块。
|
||
"""
|
||
|
||
def get_name(self):
|
||
"""
|
||
返回测试模块名称。
|
||
"""
|
||
return "限流测试模块"
|
||
|
||
def get_priority(self):
|
||
"""
|
||
返回测试模块优先级。
|
||
"""
|
||
return 1
|
||
|
||
def limited_method(self, raise_exception: bool = False):
|
||
"""
|
||
模拟同步模块在本地限流期间跳过调用。
|
||
"""
|
||
raise RateLimitExceededException("[limited_method] 限流期间，跳过调用")
|
||
|
||
async def async_limited_method(self, raise_exception: bool = False):
|
||
"""
|
||
模拟异步模块在本地限流期间跳过调用。
|
||
"""
|
||
raise RateLimitExceededException("[async_limited_method] 限流期间，跳过调用")
|
||
|
||
|
||
@pytest.fixture
|
||
def chain() -> ChainBase:
|
||
"""
|
||
构造隔离的 ChainBase，避免依赖真实模块和插件运行状态。
|
||
"""
|
||
limited_module = _LimitedModule()
|
||
plugin_manager = Mock()
|
||
plugin_manager.get_plugin_modules.return_value = {}
|
||
module_manager = Mock()
|
||
module_manager.get_running_modules.return_value = [limited_module]
|
||
message_queue = Mock()
|
||
message_queue.bind.return_value = Mock()
|
||
return ChainBase(
|
||
ChainRuntimeContext(
|
||
module_manager=module_manager,
|
||
plugin_manager=plugin_manager,
|
||
event_manager=Mock(),
|
||
message_oper=Mock(),
|
||
message_helper=Mock(),
|
||
file_cache=Mock(),
|
||
async_file_cache=Mock(),
|
||
message_queue=message_queue,
|
||
module_dispatcher_factory=ModuleInvocationDispatcher,
|
||
site_repository=Mock(),
|
||
subscription_repository=Mock(),
|
||
subscription_mutation_scope=Mock(),
|
||
sync_subscription_mutation_scope=Mock(),
|
||
subscription_delete_scope=Mock(),
|
||
sync_subscription_delete_scope=Mock(),
|
||
subscription_completion_scope=Mock(),
|
||
rule_group_mutation_scope=Mock(),
|
||
site_reference_mutation_scope=Mock(),
|
||
download_history_repository=Mock(),
|
||
transfer_history_repository=Mock(),
|
||
transfer_admission_repository=Mock(),
|
||
transfer_execution_repository=Mock(),
|
||
media_server_repository=Mock(),
|
||
download_failure_repository=Mock(),
|
||
user_repository=Mock(),
|
||
)
|
||
)
|
||
|
||
|
||
def test_rate_limit_is_not_reported_as_system_error(chain: ChainBase) -> None:
|
||
"""
|
||
本地限流跳过不应写入系统错误通知或事件。
|
||
"""
|
||
result = chain.run_module("limited_method")
|
||
|
||
assert result is None
|
||
chain.messagehelper.put.assert_not_called()
|
||
chain.eventmanager.send_event.assert_not_called()
|
||
|
||
|
||
def test_rate_limit_can_still_be_raised_explicitly(chain: ChainBase) -> None:
|
||
"""
|
||
调用方显式要求抛出异常时，限流异常应继续向上抛出。
|
||
"""
|
||
with pytest.raises(RateLimitExceededException):
|
||
chain.run_module("limited_method", raise_exception=True)
|
||
|
||
chain.messagehelper.put.assert_not_called()
|
||
chain.eventmanager.send_event.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_rate_limit_is_not_reported_as_system_error(chain: ChainBase) -> None:
|
||
"""
|
||
异步模块的本地限流跳过也不应触发系统错误路径。
|
||
"""
|
||
result = await chain.async_run_module("async_limited_method")
|
||
|
||
assert result is None
|
||
chain.messagehelper.put.assert_not_called()
|
||
chain.eventmanager.send_event.assert_not_called()
|