mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-02 21:34:32 +08:00
Merge pull request #6926 from cwd0204/test/optional-downloader-stubs
This commit is contained in:
jxxghp
@@ -107,7 +107,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
|
||||
| Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 |
|
||||
| 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 |
|
||||
| 全量 mypy 历史债务 | 9,174 / 508 文件 | 系统版本查询补齐类型后的现状基线；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
|
||||
| Ruff 历史诊断 | 505 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
|
||||
| Ruff 历史诊断 | 498 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
|
||||
| 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
|
||||
|
||||
### 3.3 热点文件
|
||||
|
||||
@@ -842,10 +842,6 @@
|
||||
"tests/test_system_database_backup_api.py": {
|
||||
"I001": 1
|
||||
},
|
||||
"tests/test_system_notification_dispatch.py": {
|
||||
"E402": 5,
|
||||
"I001": 2
|
||||
},
|
||||
"tests/test_systemconfig_oper.py": {
|
||||
"I001": 1
|
||||
},
|
||||
|
||||
@@ -2,16 +2,15 @@
|
||||
|
||||
# ruff: noqa: E402 - 可选下载器模块必须在导入 Chain 前完成隔离。
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
import unittest
|
||||
from types import ModuleType
|
||||
from unittest.mock import Mock
|
||||
|
||||
sys.modules.setdefault("qbittorrentapi", ModuleType("qbittorrentapi"))
|
||||
setattr(sys.modules["qbittorrentapi"], "TorrentFilesList", list)
|
||||
sys.modules.setdefault("transmission_rpc", ModuleType("transmission_rpc"))
|
||||
setattr(sys.modules["transmission_rpc"], "File", object)
|
||||
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
|
||||
@@ -47,85 +46,70 @@ class _LimitedModule:
|
||||
raise RateLimitExceededException("[async_limited_method] 限流期间，跳过调用")
|
||||
|
||||
|
||||
class ChainRateLimitTest(unittest.TestCase):
|
||||
"""验证模块限流异常的兼容传播和告警语义。"""
|
||||
|
||||
def _build_chain(self):
|
||||
"""
|
||||
构造隔离的 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_helper = Mock()
|
||||
event_manager = Mock()
|
||||
message_queue = Mock()
|
||||
message_queue.bind.return_value = Mock()
|
||||
chain = ChainBase(
|
||||
ChainRuntimeContext(
|
||||
module_manager=module_manager,
|
||||
plugin_manager=plugin_manager,
|
||||
event_manager=event_manager,
|
||||
message_oper=Mock(),
|
||||
message_helper=message_helper,
|
||||
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(),
|
||||
)
|
||||
@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(),
|
||||
)
|
||||
return chain
|
||||
)
|
||||
|
||||
def test_rate_limit_is_not_reported_as_system_error(self):
|
||||
"""
|
||||
本地限流跳过不应写入系统错误通知或事件。
|
||||
"""
|
||||
chain = self._build_chain()
|
||||
|
||||
result = chain.run_module("limited_method")
|
||||
def test_rate_limit_is_not_reported_as_system_error(chain: ChainBase) -> None:
|
||||
"""本地限流跳过不应写入系统错误通知或事件。"""
|
||||
result = chain.run_module("limited_method")
|
||||
|
||||
self.assertIsNone(result)
|
||||
chain.messagehelper.put.assert_not_called()
|
||||
chain.eventmanager.send_event.assert_not_called()
|
||||
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(self):
|
||||
"""
|
||||
调用方显式要求抛出异常时，限流异常应继续向上抛出。
|
||||
"""
|
||||
chain = self._build_chain()
|
||||
|
||||
with self.assertRaises(RateLimitExceededException):
|
||||
chain.run_module("limited_method", raise_exception=True)
|
||||
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()
|
||||
chain.messagehelper.put.assert_not_called()
|
||||
chain.eventmanager.send_event.assert_not_called()
|
||||
|
||||
def test_async_rate_limit_is_not_reported_as_system_error(self):
|
||||
"""
|
||||
异步模块的本地限流跳过也不应触发系统错误路径。
|
||||
"""
|
||||
chain = self._build_chain()
|
||||
|
||||
result = asyncio.run(chain.async_run_module("async_limited_method"))
|
||||
@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")
|
||||
|
||||
self.assertIsNone(result)
|
||||
chain.messagehelper.put.assert_not_called()
|
||||
chain.eventmanager.send_event.assert_not_called()
|
||||
assert result is None
|
||||
chain.messagehelper.put.assert_not_called()
|
||||
chain.eventmanager.send_event.assert_not_called()
|
||||
|
||||
@@ -1,139 +1,141 @@
|
||||
import sys
|
||||
"""系统内部用户消息分发与异步立即发送的上下文传播测试。"""
|
||||
|
||||
# ruff: noqa: E402 - 可选三方模块必须在导入消息链前完成隔离。
|
||||
|
||||
import asyncio
|
||||
import unittest
|
||||
from types import ModuleType
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.modules.setdefault("qbittorrentapi", ModuleType("qbittorrentapi"))
|
||||
setattr(sys.modules["qbittorrentapi"], "TorrentFilesList", list)
|
||||
sys.modules.setdefault("transmission_rpc", ModuleType("transmission_rpc"))
|
||||
setattr(sys.modules["transmission_rpc"], "File", object)
|
||||
sys.modules.setdefault("psutil", ModuleType("psutil"))
|
||||
import pytest
|
||||
|
||||
from app.testing.bootstrap import ensure_optional_stub
|
||||
|
||||
# 仅在可选依赖缺失时补占位；已安装时保留真实模块，避免污染同一进程里导入真实类型的用例。
|
||||
ensure_optional_stub("qbittorrentapi", TorrentFilesList=list)
|
||||
ensure_optional_stub("transmission_rpc", File=object)
|
||||
ensure_optional_stub("psutil")
|
||||
|
||||
from app.chain.message import MessageChain
|
||||
from app.application.messaging.message import MessageQueueManager
|
||||
from app.schemas.message import Message
|
||||
from app.runtime.correlation import correlation_scope, get_correlation_id
|
||||
from app.chain.message import MessageChain
|
||||
from app.foundation.identity import (
|
||||
SYSTEM_INTERNAL_USER_ID,
|
||||
is_internal_user_id,
|
||||
normalize_internal_user_id,
|
||||
)
|
||||
from app.runtime.correlation import correlation_scope, get_correlation_id
|
||||
from app.schemas.message import Message
|
||||
|
||||
|
||||
class TestSystemNotificationDispatch(unittest.TestCase):
|
||||
def test_internal_userid_identity_helpers(self):
|
||||
self.assertTrue(is_internal_user_id(SYSTEM_INTERNAL_USER_ID))
|
||||
self.assertTrue(is_internal_user_id(" System "))
|
||||
self.assertIsNone(normalize_internal_user_id(SYSTEM_INTERNAL_USER_ID))
|
||||
self.assertEqual(normalize_internal_user_id("10001"), "10001")
|
||||
def _internal_message() -> Message:
|
||||
"""构造以系统内部用户为收件人的消息。"""
|
||||
return Message(
|
||||
userid=SYSTEM_INTERNAL_USER_ID,
|
||||
username="admin",
|
||||
title="后台报告",
|
||||
text="任务完成",
|
||||
)
|
||||
|
||||
def test_post_message_normalizes_internal_userid_before_queueing(self):
|
||||
chain = MessageChain()
|
||||
message = Message(
|
||||
userid=SYSTEM_INTERNAL_USER_ID,
|
||||
username="admin",
|
||||
title="后台报告",
|
||||
text="任务完成",
|
||||
)
|
||||
|
||||
with patch("app.chain._messaging.MessageTemplateHelper.render", return_value=message), patch.object(
|
||||
chain.messagehelper, "put"
|
||||
), patch.object(chain.messageoper, "add"), patch.object(
|
||||
chain.eventmanager, "send_event"
|
||||
) as send_event, patch.object(
|
||||
chain.user_repository,
|
||||
"get_notification_settings",
|
||||
return_value={"telegram_userid": "admin-1"},
|
||||
), patch.object(
|
||||
chain.messagequeue, "send_message"
|
||||
) as send_message:
|
||||
chain.post_message(message)
|
||||
def test_internal_userid_identity_helpers() -> None:
|
||||
"""系统内部用户标识的判定与归一化。"""
|
||||
assert is_internal_user_id(SYSTEM_INTERNAL_USER_ID)
|
||||
assert is_internal_user_id(" System ")
|
||||
assert normalize_internal_user_id(SYSTEM_INTERNAL_USER_ID) is None
|
||||
assert normalize_internal_user_id("10001") == "10001"
|
||||
|
||||
event_payload = send_event.call_args.kwargs["data"]
|
||||
queued_message = send_message.call_args.kwargs["message"]
|
||||
|
||||
self.assertIsNone(event_payload["userid"])
|
||||
self.assertIsNone(queued_message.userid)
|
||||
self.assertFalse(send_message.call_args.kwargs["immediately"])
|
||||
def test_post_message_normalizes_internal_userid_before_queueing() -> None:
|
||||
"""入队前应把系统内部用户归一化为空，不进入事件与队列载荷。"""
|
||||
chain = MessageChain()
|
||||
message = _internal_message()
|
||||
|
||||
def test_send_direct_message_normalizes_internal_userid(self):
|
||||
chain = MessageChain()
|
||||
message = Message(
|
||||
userid=SYSTEM_INTERNAL_USER_ID,
|
||||
username="admin",
|
||||
title="后台报告",
|
||||
text="任务完成",
|
||||
)
|
||||
with patch("app.chain._messaging.MessageTemplateHelper.render", return_value=message), patch.object(
|
||||
chain.messagehelper, "put"
|
||||
), patch.object(chain.messageoper, "add"), patch.object(
|
||||
chain.eventmanager, "send_event"
|
||||
) as send_event, patch.object(
|
||||
chain.user_repository,
|
||||
"get_notification_settings",
|
||||
return_value={"telegram_userid": "admin-1"},
|
||||
), patch.object(
|
||||
chain.messagequeue, "send_message"
|
||||
) as send_message:
|
||||
chain.post_message(message)
|
||||
|
||||
with patch.object(chain, "run_module") as run_module:
|
||||
chain.send_direct_message(message)
|
||||
event_payload = send_event.call_args.kwargs["data"]
|
||||
queued_message = send_message.call_args.kwargs["message"]
|
||||
|
||||
sent_message = run_module.call_args.kwargs["message"]
|
||||
self.assertIsNone(sent_message.userid)
|
||||
assert event_payload["userid"] is None
|
||||
assert queued_message.userid is None
|
||||
assert send_message.call_args.kwargs["immediately"] is False
|
||||
|
||||
def test_async_send_message_uses_executor_for_immediate_send(self):
|
||||
"""异步立即发送不能在事件循环里直接执行同步渠道回调。"""
|
||||
class _FakeLoop:
|
||||
"""同步模拟 asyncio 的 executor 提交，并记录提交时的上下文。"""
|
||||
|
||||
def __init__(self, loop):
|
||||
self.called = False
|
||||
self.loop = loop
|
||||
self.submission_correlation_id = None
|
||||
def test_send_direct_message_normalizes_internal_userid() -> None:
|
||||
"""直接发送时同样归一化系统内部用户。"""
|
||||
chain = MessageChain()
|
||||
|
||||
def run_in_executor(self, _executor, func, *args):
|
||||
"""立即执行 worker 调用并返回已完成 Future。"""
|
||||
self.called = True
|
||||
self.submission_correlation_id = get_correlation_id()
|
||||
future = self.loop.create_future()
|
||||
try:
|
||||
func(*args)
|
||||
except BaseException as error:
|
||||
future.set_exception(error)
|
||||
else:
|
||||
future.set_result(None)
|
||||
return future
|
||||
with patch.object(chain, "run_module") as run_module:
|
||||
chain.send_direct_message(_internal_message())
|
||||
|
||||
async def _run():
|
||||
manager = MessageQueueManager()
|
||||
fake_loop = _FakeLoop(asyncio.get_running_loop())
|
||||
observed = []
|
||||
with patch(
|
||||
"asyncio.get_running_loop",
|
||||
return_value=fake_loop,
|
||||
), patch.object(
|
||||
manager,
|
||||
"_send",
|
||||
side_effect=lambda *_args, **_kwargs: observed.append(
|
||||
get_correlation_id()
|
||||
),
|
||||
) as send:
|
||||
with correlation_scope("message-request"):
|
||||
await manager.async_send_message("payload", immediately=True)
|
||||
self.assertTrue(fake_loop.called)
|
||||
self.assertIsNone(fake_loop.submission_correlation_id)
|
||||
self.assertEqual(observed, ["message-request"])
|
||||
send.assert_called_once_with("payload")
|
||||
sent_message = run_module.call_args.kwargs["message"]
|
||||
assert sent_message.userid is None
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
def test_async_send_message_preserves_call_context(self):
|
||||
"""异步立即发送的同步渠道回调应保留当前请求关联 ID。"""
|
||||
observed = []
|
||||
class _FakeLoop:
|
||||
"""同步模拟 asyncio 的 executor 提交，并记录提交时的上下文。"""
|
||||
|
||||
async def _run():
|
||||
manager = MessageQueueManager()
|
||||
with patch.object(
|
||||
manager,
|
||||
"_send",
|
||||
side_effect=lambda *_args, **_kwargs: observed.append(
|
||||
get_correlation_id()
|
||||
),
|
||||
):
|
||||
with correlation_scope("message-request"):
|
||||
await manager.async_send_message("payload", immediately=True)
|
||||
def __init__(self, loop):
|
||||
self.called = False
|
||||
self.loop = loop
|
||||
self.submission_correlation_id = None
|
||||
|
||||
asyncio.run(_run())
|
||||
def run_in_executor(self, _executor, func, *args):
|
||||
"""立即执行 worker 调用并返回已完成 Future。"""
|
||||
self.called = True
|
||||
self.submission_correlation_id = get_correlation_id()
|
||||
future = self.loop.create_future()
|
||||
try:
|
||||
func(*args)
|
||||
except BaseException as error: # pylint: disable=broad-exception-caught
|
||||
future.set_exception(error)
|
||||
else:
|
||||
future.set_result(None)
|
||||
return future
|
||||
|
||||
self.assertEqual(observed, ["message-request"])
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_send_message_uses_executor_for_immediate_send() -> None:
|
||||
"""异步立即发送不能在事件循环里直接执行同步渠道回调。"""
|
||||
manager = MessageQueueManager()
|
||||
fake_loop = _FakeLoop(asyncio.get_running_loop())
|
||||
observed = []
|
||||
with patch(
|
||||
"asyncio.get_running_loop",
|
||||
return_value=fake_loop,
|
||||
), patch.object(
|
||||
manager,
|
||||
"_send",
|
||||
side_effect=lambda *_args, **_kwargs: observed.append(get_correlation_id()),
|
||||
) as send:
|
||||
with correlation_scope("message-request"):
|
||||
await manager.async_send_message("payload", immediately=True)
|
||||
|
||||
assert fake_loop.called
|
||||
assert fake_loop.submission_correlation_id is None
|
||||
assert observed == ["message-request"]
|
||||
send.assert_called_once_with("payload")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_send_message_preserves_call_context() -> None:
|
||||
"""异步立即发送的同步渠道回调应保留当前请求关联 ID。"""
|
||||
observed = []
|
||||
manager = MessageQueueManager()
|
||||
with patch.object(
|
||||
manager,
|
||||
"_send",
|
||||
side_effect=lambda *_args, **_kwargs: observed.append(get_correlation_id()),
|
||||
):
|
||||
with correlation_scope("message-request"):
|
||||
await manager.async_send_message("payload", immediately=True)
|
||||
|
||||
assert observed == ["message-request"]
|
||||
|
||||
Reference in New Issue
Block a user