diff --git a/app/application/notification.py b/app/application/notification.py index bdd4876d3..3a92fc13b 100644 --- a/app/application/notification.py +++ b/app/application/notification.py @@ -44,12 +44,12 @@ def get_notification_configs( ) -def get_notification_switch(mtype: MessageType) -> Optional[str]: - """返回指定通知场景的目标范围。""" +def get_notification_switch(mtype: MessageType) -> str: + """返回指定通知场景的目标范围，缺少配置时默认仅管理员。""" for switch in get_service_configs( SystemConfigKey.NotificationSwitchs, NotificationSwitchConf, ): if switch.type == mtype.value: - return switch.action - return None + return switch.action or "admin" + return "admin" diff --git a/app/chain/_messaging.py b/app/chain/_messaging.py index 9ad3f095c..4a50a7f2d 100644 --- a/app/chain/_messaging.py +++ b/app/chain/_messaging.py @@ -69,48 +69,100 @@ def _notification_route_steps( 生成同步和异步通知共用的路由状态转换。 生成器只表达用户设置查询、日志和投递决策，调用方分别执行真实同步或 - 异步 I/O，避免两套外壳各自维护管理员回退和原消息投递规则。 + 异步 I/O，避免两套外壳各自维护管理员回退和原消息投递规则。没有可用的 + 用户通知设置时只记录日志，不产出无目标消息，避免渠道默认广播。 """ admin_sent = False - send_original = not action - for route_action in action.split(",") if action else (): + send_original = action is None + route_actions: tuple[str, ...] = () + if action is not None: + route_actions = tuple( + route_action.strip() or "admin" + for route_action in action.split(",") + ) + for route_action in route_actions: routed_message = copy.deepcopy(message) if route_action == "admin" and not admin_sent: settings = yield _NotificationRouteLookup( username=superuser, log_message=f"{routed_message.mtype} 的消息已设置发送给管理员", ) - routed_message.targets = cast(dict[str, Any], settings) admin_sent = True - elif route_action == "user" and routed_message.username: - username = cast(str, routed_message.username) - settings = yield _NotificationRouteLookup( - username=username, - log_message=( - f"{routed_message.mtype} 的消息已设置发送给用户 {username}" - ), - ) - routed_message.targets = cast(dict[str, Any], settings) if settings is None: - if not admin_sent: - settings = yield _NotificationRouteLookup( - username=superuser, - log_message=( - f"用户 {username} 不存在，消息将发送给管理员" - ), - ) - routed_message.targets = cast(dict[str, Any], settings) - admin_sent = True - else: + yield _NotificationRouteLog( + message="管理员未配置通知渠道，消息不会广播发送" + ) + continue + routed_message.targets = cast(dict[str, Any], settings) + elif route_action == "user": + if not routed_message.username: + if admin_sent: yield _NotificationRouteLog( - message=f"用户 {username} 不存在，消息无法发送到对应用户" + message="消息未关联用户，且管理员已收到消息" ) continue - elif username == superuser: + settings = yield _NotificationRouteLookup( + username=superuser, + log_message="消息未关联用户，消息将发送给管理员", + ) admin_sent = True - else: + if settings is None: + yield _NotificationRouteLog( + message="管理员未配置通知渠道，消息不会广播发送" + ) + continue + routed_message.targets = cast(dict[str, Any], settings) + else: + username = cast(str, routed_message.username) + settings = yield _NotificationRouteLookup( + username=username, + log_message=( + f"{routed_message.mtype} 的消息已设置发送给用户 {username}" + ), + ) + routed_message.targets = cast(dict[str, Any], settings) + if settings is None: + if not admin_sent: + settings = yield _NotificationRouteLookup( + username=superuser, + log_message=( + f"用户 {username} 不存在，消息将发送给管理员" + ), + ) + routed_message.targets = cast(dict[str, Any], settings) + admin_sent = True + if settings is None: + yield _NotificationRouteLog( + message="管理员未配置通知渠道，消息不会广播发送" + ) + continue + else: + yield _NotificationRouteLog( + message=f"用户 {username} 不存在，消息无法发送到对应用户" + ) + continue + elif username == superuser: + admin_sent = True + elif route_action == "all": send_original = not admin_sent break + else: + yield _NotificationRouteLog( + message=f"未知通知路由 {route_action}，消息将按管理员范围发送" + ) + if admin_sent: + continue + settings = yield _NotificationRouteLookup( + username=superuser, + log_message="通知路由无效，消息将发送给管理员", + ) + admin_sent = True + if settings is None: + yield _NotificationRouteLog( + message="管理员未配置通知渠道，消息不会广播发送" + ) + continue + routed_message.targets = cast(dict[str, Any], settings) yield _NotificationRouteDelivery( message=routed_message, immediately=False, @@ -145,10 +197,12 @@ def _render_notification_message( def _notification_route_action(message: Message) -> Optional[str]: - """仅为未绑定真实用户的业务通知读取隔离路由配置。""" - if message.userid or not message.mtype: + """为未绑定真实用户的通知确定隔离路由，缺少类型时默认仅管理员。""" + if message.userid: return None - return get_notification_switch(message.mtype) + if not message.mtype: + return "admin" + return get_notification_switch(message.mtype) or "admin" class MessageProcessingMixin: diff --git a/app/modules/__init__.py b/app/modules/__init__.py index 2312e9625..cd16fdd81 100644 --- a/app/modules/__init__.py +++ b/app/modules/__init__.py @@ -262,6 +262,9 @@ class _MessageBase(ServiceBase[TService, NotificationConf]): # 检查消息来源 if message.source and message.source != source: return False + # 没有类型且没有显式目标时不得进入渠道的默认广播路径。 + if not message.userid and not message.mtype: + return bool(message.targets) # 不是定向发送时，检查消息类型开关 if not message.userid and message.mtype: conf = self.get_config(source) diff --git a/app/runtime/extensions/service.py b/app/runtime/extensions/service.py index aae816b7a..b3ecc40ad 100644 --- a/app/runtime/extensions/service.py +++ b/app/runtime/extensions/service.py @@ -87,9 +87,9 @@ class ServiceConfigHelper: ) @staticmethod - def get_notification_switch(mtype: MessageType) -> Optional[str]: - """返回指定通知场景的目标范围。""" + def get_notification_switch(mtype: MessageType) -> str: + """返回指定通知场景的目标范围，缺少配置时默认仅管理员。""" for switch in ServiceConfigHelper.get_notification_switches(): if switch.type == mtype.value: - return switch.action - return None + return switch.action or "admin" + return "admin" diff --git a/tests/test_application_service_directory.py b/tests/test_application_service_directory.py index 6becf7040..68e757394 100644 --- a/tests/test_application_service_directory.py +++ b/tests/test_application_service_directory.py @@ -47,3 +47,4 @@ def test_named_service_config_helpers_preserve_enabled_policy(monkeypatch) -> No item.name for item in get_notification_configs(include_disabled=True) ] == ["enabled-channel", "disabled-channel"] assert get_notification_switch(MessageType.Download) == "admin" + assert get_notification_switch(MessageType.Agent) == "admin" diff --git a/tests/test_message_notifications.py b/tests/test_message_notifications.py index cc68bf540..d04762048 100644 --- a/tests/test_message_notifications.py +++ b/tests/test_message_notifications.py @@ -182,6 +182,11 @@ def test_notification_post_message_is_persisted_without_sse_queue(monkeypatch) - chain = ChainBase() # messagequeue 是全局单例，用 monkeypatch 避免用例间污染 + monkeypatch.setattr( + chain.user_repository, + "get_notification_settings", + lambda _username: {"telegram_userid": "admin-1"}, + ) send_message = Mock() monkeypatch.setattr(chain.messagequeue, "send_message", send_message) monkeypatch.setattr(chain.eventmanager, "send_event", Mock()) @@ -212,6 +217,11 @@ def test_agent_notification_post_message_is_persisted_without_sse_queue(monkeypa chain = ChainBase() # messagequeue 是全局单例，用 monkeypatch 避免用例间污染 + monkeypatch.setattr( + chain.user_repository, + "get_notification_settings", + lambda _username: {"telegram_userid": "admin-1"}, + ) send_message = Mock() monkeypatch.setattr(chain.messagequeue, "send_message", send_message) monkeypatch.setattr(chain.eventmanager, "send_event", Mock()) @@ -240,6 +250,11 @@ def test_transient_notification_post_message_skips_history_but_dispatches(monkey chain = ChainBase() # messagequeue 是全局单例，用 monkeypatch 避免用例间污染 + monkeypatch.setattr( + chain.user_repository, + "get_notification_settings", + lambda _username: {"telegram_userid": "admin-1"}, + ) send_message = Mock() monkeypatch.setattr(chain.messagequeue, "send_message", send_message) send_event = Mock() diff --git a/tests/test_notification_routing_parity.py b/tests/test_notification_routing_parity.py index 73559ce23..647dfb390 100644 --- a/tests/test_notification_routing_parity.py +++ b/tests/test_notification_routing_parity.py @@ -6,7 +6,9 @@ from unittest.mock import AsyncMock, Mock import pytest from app.chain._messaging import MessageProcessingMixin, NotificationMixin +from app.modules.wechat import WechatModule from app.schemas.message import Message +from app.schemas.system import NotificationConf from app.schemas.types import MessageType @@ -175,3 +177,79 @@ def test_notification_routing_lookup_failure_sync_async_parity(monkeypatch) -> N async_chain.messagequeue.async_send_message.assert_not_awaited() assert sync_chain.messageoper.add.call_count == 1 assert async_chain.messageoper.async_add.await_count == 1 + + +def test_untyped_notification_defaults_to_admin_sync_async(monkeypatch) -> None: + """未设置消息类型的通知应同步、异步都只查询管理员并按其设置投递。""" + monkeypatch.setattr( + "app.chain._messaging.get_notification_switch", + lambda _mtype: None, + ) + settings = {"admin": {"wechat_userid": "admin-1"}} + sync_repository = _NotificationSettingsRepository(settings) + async_repository = _NotificationSettingsRepository(settings) + sync_chain = _NotificationHarness(sync_repository) + async_chain = _NotificationHarness(async_repository) + message = Message(title="工作流结果", text="执行完成") + + sync_chain.post_message(message.model_copy(deep=True)) + asyncio.run(async_chain.async_post_message(message.model_copy(deep=True))) + + assert sync_repository.sync_calls == ["admin"] + assert async_repository.async_calls == ["admin"] + expected_targets = [{"wechat_userid": "admin-1"}] + assert [ + item[0]["targets"] + for item in _delivery_snapshot(sync_chain.messagequeue.send_message) + ] == expected_targets + assert [ + item[0]["targets"] + for item in _delivery_snapshot(async_chain.messagequeue.async_send_message) + ] == expected_targets + + +def test_user_route_without_username_falls_back_to_admin(monkeypatch) -> None: + """用户范围通知缺少责任人时应回退管理员，不得恢复为全体广播。""" + monkeypatch.setattr( + "app.chain._messaging.get_notification_switch", + lambda _mtype: "user,all", + ) + repository = _NotificationSettingsRepository( + {"admin": {"telegram_userid": "admin-1"}} + ) + chain = _NotificationHarness(repository) + + chain.post_message(Message(mtype=MessageType.Download, title="下载完成")) + + assert repository.sync_calls == ["admin"] + deliveries = _delivery_snapshot(chain.messagequeue.send_message) + assert [item[0]["targets"] for item in deliveries] == [ + {"telegram_userid": "admin-1"} + ] + assert [item[1] for item in deliveries] == [False] + + +def test_wechat_rejects_untyped_broadcast_without_explicit_targets(monkeypatch) -> None: + """企业微信渠道不得把无类型且无目标的消息交给客户端默认广播。""" + module = WechatModule() + config = NotificationConf(name="wechat-test", type="wechat", enabled=True) + module._configs = {config.name: config} + client = Mock() + monkeypatch.setattr(module, "get_configs", lambda: {config.name: config}) + monkeypatch.setattr(module, "get_instance", lambda _name: client) + + module.post_message( + Message(source=config.name, title="工作流结果", text="执行完成") + ) + client.send_msg.assert_not_called() + + module.post_message( + Message( + source=config.name, + title="工作流结果", + text="执行完成", + targets={"wechat_userid": "admin-1"}, + ) + ) + client.send_msg.assert_called_once() + assert client.send_msg.call_args.kwargs["userid"] == "admin-1" diff --git a/tests/test_outbox.py b/tests/test_outbox.py index d2c3c6947..2eacc4517 100644 --- a/tests/test_outbox.py +++ b/tests/test_outbox.py @@ -458,6 +458,11 @@ def test_strict_notification_preserves_legacy_provider_signature(monkeypatch) -> chain = CommandChain() received = [] + monkeypatch.setattr( + chain.user_repository, + "get_notification_settings", + lambda _username: {"telegram_userid": "admin-1"}, + ) def legacy_provider(message) -> None: """模拟只接受旧式单参数签名的第三方通知 provider。""" received.append((message, get_correlation_id())) @@ -488,6 +493,11 @@ def test_strict_notification_retry_writes_history_once(monkeypatch) -> None: history_sources = set() provider_sources = [] + monkeypatch.setattr( + chain.user_repository, + "get_notification_settings", + lambda _username: {"telegram_userid": "admin-1"}, + ) monkeypatch.setattr(chain.eventmanager, "send_event", lambda *_args, **_kwargs: None) monkeypatch.setattr( chain.messageoper, diff --git a/tests/test_system_notification_dispatch.py b/tests/test_system_notification_dispatch.py index df3475b6e..f327892f5 100644 --- a/tests/test_system_notification_dispatch.py +++ b/tests/test_system_notification_dispatch.py @@ -42,6 +42,10 @@ class TestSystemNotificationDispatch(unittest.TestCase): ), 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)