mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-02 21:34:32 +08:00
test: stub optional downloader modules only when they are missing
test_chain_rate_limit.py and test_system_notification_dispatch.py installed a bare qbittorrentapi/transmission_rpc/psutil ModuleType into sys.modules at import time with sys.modules.setdefault. When either file is collected before a test that imports the real package in the same shard (for example tests/test_downloader_add_failure_message.py and its TorrentDictionary import), the placeholder wins and collection fails with "cannot import name ... (unknown location)". Any change to the test file set reshuffles the six CI shards and can surface this. Use app.testing.bootstrap.ensure_optional_stub, which keeps the real module when it is installed, and convert both unittest.TestCase files to pytest-native functions as the testing guide requires when touching a TestCase file. Record the resulting lower Ruff watermark. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
freeman
@@ -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