diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 25f11996e..3322e6988 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -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 热点文件 diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index 3228b66a7..afa9f84ca 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -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 }, diff --git a/tests/test_chain_rate_limit.py b/tests/test_chain_rate_limit.py index 7c722d8d7..62e7d8162 100644 --- a/tests/test_chain_rate_limit.py +++ b/tests/test_chain_rate_limit.py @@ -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() diff --git a/tests/test_system_notification_dispatch.py b/tests/test_system_notification_dispatch.py index f327892f5..68825c748 100644 --- a/tests/test_system_notification_dispatch.py +++ b/tests/test_system_notification_dispatch.py @@ -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"]