From caee0ea37468721b4bded588bbd46e1ea3e90c58 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 29 Aug 2026 19:25:30 +0800 Subject: [PATCH] refactor(chain): unify notification routing decisions --- app/chain/_messaging.py | 397 ++++++++++-------- docs/architecture-optimization-checklist.md | 4 +- docs/architecture-overview.md | 8 +- .../fixtures/architecture/mypy-baseline.json | 4 +- .../runtime-contract-baseline.json | 46 +- tests/test_architecture_contract_baseline.py | 8 +- tests/test_architecture_event_facts.py | 8 +- tests/test_notification_routing_parity.py | 177 ++++++++ 8 files changed, 429 insertions(+), 223 deletions(-) create mode 100644 tests/test_notification_routing_parity.py diff --git a/app/chain/_messaging.py b/app/chain/_messaging.py index 263d083ac..ad9bb4514 100644 --- a/app/chain/_messaging.py +++ b/app/chain/_messaging.py @@ -5,6 +5,8 @@ eventmanager、messageoper、messagequeue 等协作对象。 """ import copy +from collections.abc import Generator, Mapping +from dataclasses import dataclass from datetime import datetime from typing import Any, Dict, List, Optional, Union, cast @@ -23,10 +25,137 @@ from app.schemas.transfer import TransferInfo from app.schemas.types import EventType, NotificationChannel +@dataclass(frozen=True, slots=True) +class _NotificationRouteLookup: + """通知路由状态机请求读取指定用户的渠道设置。""" + + username: str + log_message: str + + +@dataclass(frozen=True, slots=True) +class _NotificationRouteDelivery: + """通知路由状态机产出的单次投递决策。""" + + message: Message + immediately: bool + + +@dataclass(frozen=True, slots=True) +class _NotificationRouteLog: + """通知路由状态机产出的无投递日志决策。""" + + message: str + + +_NotificationRouteStep = Union[ + _NotificationRouteLookup, + _NotificationRouteDelivery, + _NotificationRouteLog, +] + + +def _notification_route_steps( + message: Message, + *, + action: Optional[str], + superuser: str, +) -> Generator[ + _NotificationRouteStep, + Optional[Mapping[str, object]], + None, +]: + """ + 生成同步和异步通知共用的路由状态转换。 + + 生成器只表达用户设置查询、日志和投递决策，调用方分别执行真实同步或 + 异步 I/O，避免两套外壳各自维护管理员回退和原消息投递规则。 + """ + admin_sent = False + send_original = not action + for route_action in action.split(",") if action else (): + 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=f"用户 {username} 不存在，消息无法发送到对应用户" + ) + continue + elif username == superuser: + admin_sent = True + else: + send_original = not admin_sent + break + yield _NotificationRouteDelivery( + message=routed_message, + immediately=False, + ) + + if send_original: + yield _NotificationRouteDelivery( + message=message, + immediately=bool(message.userid), + ) + + +def _render_notification_message( + *, + message: Optional[Message], + meta: Optional[MetaBase], + mediainfo: Optional[Union[MediaInfo, MusicInfo]], + torrentinfo: Optional[TorrentInfo], + transferinfo: Optional[TransferInfo], + kwargs: dict[str, Any], +) -> Optional[Message]: + """使用同步和异步入口共用的上下文规范化并渲染通知。""" + kwargs.setdefault("current_time", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) + return MessageTemplateHelper.render( + message=message, + meta=meta, + mediainfo=mediainfo, + torrentinfo=torrentinfo, + transferinfo=transferinfo, + **kwargs, + ) + + +def _notification_route_action(message: Message) -> Optional[str]: + """仅为未绑定真实用户的业务通知读取隔离路由配置。""" + if message.userid or not message.mtype: + return None + return get_notification_switch(message.mtype) + + class MessageProcessingMixin: - __mixin_host_protocol__ = ChainRuntimeMixinHost """消息输入/处理状态机与通知派发规范化。""" + __mixin_host_protocol__ = ChainRuntimeMixinHost + def start_message_processing_status( self, channel: NotificationChannel, @@ -119,9 +248,10 @@ class MessageProcessingMixin: class NotificationMixin: + """通知消息发送域：渲染、隔离路由、队列发送与消息编辑。""" + __mixin_host_protocol__ = ChainRuntimeMixinHost user_repository: ChainUserRepository - """通知消息发送域：渲染、隔离路由、队列发送与消息编辑。""" def post_message( self, @@ -144,18 +274,14 @@ class NotificationMixin: """ strict_delivery = bool(kwargs.pop("_strict_delivery", False)) strict_source = kwargs.pop("_strict_source", None) - # 添加格式化的时间参数 - kwargs.setdefault("current_time", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) - # 渲染消息 - message = MessageTemplateHelper.render( + message = _render_notification_message( message=message, meta=meta, mediainfo=mediainfo, torrentinfo=torrentinfo, transferinfo=transferinfo, - **kwargs, + kwargs=kwargs, ) - # 检查消息是否有效 if not message: logger.warning("消息为空，跳过发送") return @@ -167,87 +293,9 @@ class NotificationMixin: dispatch_message = self._normalize_notification_for_dispatch(message) if strict_source and dispatch_message.source == strict_source: dispatch_message.source = None - # 发送消息按设置隔离 - if not dispatch_message.userid and dispatch_message.mtype: - # 消息隔离设置 - notify_action = get_notification_switch( - dispatch_message.mtype - ) - if notify_action: - # 'admin' 'user,admin' 'user' 'all' - actions = notify_action.split(",") - # 是否已发送管理员标志 - admin_sended = False - send_orignal = False - useroper = self.user_repository - for action in actions: - send_message = copy.deepcopy(dispatch_message) - if action == "admin" and not admin_sended: - # 仅发送管理员 - logger.info(f"{send_message.mtype} 的消息已设置发送给管理员") - # 读取管理员消息IDS - send_message.targets = useroper.get_notification_settings( - self.runtime_config.superuser - ) - admin_sended = True - elif action == "user" and send_message.username: - # 发送对应用户 - logger.info( - f"{send_message.mtype} 的消息已设置发送给用户 {send_message.username}" - ) - # 读取用户消息IDS - send_message.targets = useroper.get_notification_settings( - send_message.username - ) - if send_message.targets is None: - # 没有找到用户 - if not admin_sended: - # 回滚发送管理员 - logger.info( - f"用户 {send_message.username} 不存在，消息将发送给管理员" - ) - # 读取管理员消息IDS - send_message.targets = useroper.get_notification_settings( - self.runtime_config.superuser - ) - admin_sended = True - else: - # 管理员发过了，此消息不发了 - logger.info( - f"用户 {send_message.username} 不存在，消息无法发送到对应用户" - ) - continue - elif send_message.username == self.runtime_config.superuser: - # 管理员同名已发送 - admin_sended = True - else: - # 按原消息发送全体 - if not admin_sended: - send_orignal = True - break - # 按设定发送 - self.eventmanager.send_event( - etype=EventType.NoticeMessage, - data=self._build_notice_message_data(send_message), - ) - self._deliver_notification( - send_message, - strict_delivery=strict_delivery, - immediately=False, - kwargs=kwargs, - ) - if not send_orignal: - return - # 发送消息事件 - self.eventmanager.send_event( - etype=EventType.NoticeMessage, - data=self._build_notice_message_data(dispatch_message), - ) - # 按原消息发送 - self._deliver_notification( + self._dispatch_notification_steps( dispatch_message, strict_delivery=strict_delivery, - immediately=bool(dispatch_message.userid), kwargs=kwargs, ) @@ -288,6 +336,49 @@ class NotificationMixin: **kwargs, ) + def _dispatch_notification_steps( + self, + message: Message, + *, + strict_delivery: bool, + kwargs: dict[str, object], + ) -> None: + """执行通知路由状态机，并在同步边界完成查询、事件和投递。""" + steps = _notification_route_steps( + message, + action=_notification_route_action(message), + superuser=self.runtime_config.superuser, + ) + lookup_result: Optional[Mapping[str, object]] = None + try: + step = next(steps) + except StopIteration: + return + while True: + lookup_result = None + if isinstance(step, _NotificationRouteLookup): + logger.info(step.log_message) + lookup_result = self.user_repository.get_notification_settings( + step.username + ) + elif isinstance(step, _NotificationRouteLog): + logger.info(step.message) + else: + self.eventmanager.send_event( + etype=EventType.NoticeMessage, + data=self._build_notice_message_data(step.message), + ) + self._deliver_notification( + step.message, + strict_delivery=strict_delivery, + immediately=step.immediately, + kwargs=kwargs, + ) + try: + step = steps.send(lookup_result) + except StopIteration: + return + async def async_post_message( self, message: Optional[Message] = None, @@ -307,104 +398,68 @@ class NotificationMixin: :param kwargs: 其他参数(覆盖业务对象属性值) :return: 成功或失败 """ - # 添加格式化的时间参数 - kwargs.setdefault("current_time", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) - # 渲染消息 - message = MessageTemplateHelper.render( + message = _render_notification_message( message=message, meta=meta, mediainfo=mediainfo, torrentinfo=torrentinfo, transferinfo=transferinfo, - **kwargs, + kwargs=kwargs, ) - # 检查消息是否有效 if not message: logger.warning("消息为空，跳过发送") return if message.save_history: await self.messageoper.async_add(**message.model_dump()) dispatch_message = self._normalize_notification_for_dispatch(message) - # 发送消息按设置隔离 - if not dispatch_message.userid and dispatch_message.mtype: - # 消息隔离设置 - notify_action = get_notification_switch( - dispatch_message.mtype - ) - if notify_action: - # 'admin' 'user,admin' 'user' 'all' - actions = notify_action.split(",") - # 是否已发送管理员标志 - admin_sended = False - send_orignal = False - useroper = self.user_repository - for action in actions: - send_message = copy.deepcopy(dispatch_message) - if action == "admin" and not admin_sended: - # 仅发送管理员 - logger.info(f"{send_message.mtype} 的消息已设置发送给管理员") - # 读取管理员消息IDS - send_message.targets = await useroper.async_get_notification_settings( - self.runtime_config.superuser - ) - admin_sended = True - elif action == "user" and send_message.username: - # 发送对应用户 - logger.info( - f"{send_message.mtype} 的消息已设置发送给用户 {send_message.username}" - ) - # 读取用户消息IDS - send_message.targets = await useroper.async_get_notification_settings( - send_message.username - ) - if send_message.targets is None: - # 没有找到用户 - if not admin_sended: - # 回滚发送管理员 - logger.info( - f"用户 {send_message.username} 不存在，消息将发送给管理员" - ) - # 读取管理员消息IDS - send_message.targets = await useroper.async_get_notification_settings( - self.runtime_config.superuser - ) - admin_sended = True - else: - # 管理员发过了，此消息不发了 - logger.info( - f"用户 {send_message.username} 不存在，消息无法发送到对应用户" - ) - continue - elif send_message.username == self.runtime_config.superuser: - # 管理员同名已发送 - admin_sended = True - else: - # 按原消息发送全体 - if not admin_sended: - send_orignal = True - break - # 按设定发送 - await self.eventmanager.async_send_event( - etype=EventType.NoticeMessage, - data=self._build_notice_message_data(send_message), - ) - await self.messagequeue.async_send_message( - "post_message", message=send_message, **kwargs - ) - if not send_orignal: - return - # 发送消息事件 - await self.eventmanager.async_send_event( - etype=EventType.NoticeMessage, - data=self._build_notice_message_data(dispatch_message), + await self._async_dispatch_notification_steps( + dispatch_message, + kwargs=kwargs, ) - # 按原消息发送 - await self.messagequeue.async_send_message( - "post_message", - message=dispatch_message, - immediately=True if dispatch_message.userid else False, - **kwargs, + + async def _async_dispatch_notification_steps( + self, + message: Message, + *, + kwargs: dict[str, object], + ) -> None: + """执行通知路由状态机，并在异步边界完成查询、事件和投递。""" + steps = _notification_route_steps( + message, + action=_notification_route_action(message), + superuser=self.runtime_config.superuser, ) + lookup_result: Optional[Mapping[str, object]] = None + try: + step = next(steps) + except StopIteration: + return + while True: + lookup_result = None + if isinstance(step, _NotificationRouteLookup): + logger.info(step.log_message) + lookup_result = ( + await self.user_repository.async_get_notification_settings( + step.username + ) + ) + elif isinstance(step, _NotificationRouteLog): + logger.info(step.message) + else: + await self.eventmanager.async_send_event( + etype=EventType.NoticeMessage, + data=self._build_notice_message_data(step.message), + ) + await self.messagequeue.async_send_message( + "post_message", + message=step.message, + immediately=step.immediately, + **kwargs, + ) + try: + step = steps.send(lookup_result) + except StopIteration: + return def post_medias_message( self, message: Message, medias: List[MediaInfo] diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md index 56c28dbb2..a1e298cf9 100644 --- a/docs/architecture-optimization-checklist.md +++ b/docs/architecture-optimization-checklist.md @@ -78,7 +78,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement | | Python 源码量 | 约 298,700 行 | 60 个文件超过 1,000 行，10 个超过 2,000 行 | | 长方法 | 299 个超过 80 行 | 70 个超过 150 行，23 个超过 250 行；大量是私有方法 | -| 全量 mypy 历史债务 | 10,255 / 592 文件 | 识别链 typed plan/result 与 TMDB 双 ABI 共用计划继续清除参数、返回值和泛型注解债；strict frontier 当前覆盖 41 个文件 | +| 全量 mypy 历史债务 | 10,245 / 592 文件 | 识别链 typed plan/result、TMDB 与消息双 ABI 共用计划继续清除参数、返回值和泛型注解债；strict frontier 当前覆盖 41 个文件 | | Ruff 历史诊断 | 642 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率低水位 | Application 81.00%，Domain 80.67% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | @@ -171,7 +171,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` - Event consumer 扫描曾把任意同名 `.register()` 调用当成事件注册；S0-L2.5 已改为证明 canonical EventManager receiver，10 个动态误报归零并保留唯一 workflow 动态注册。 - S0-L2.6 已将 producer/consumer 合并为逐调用事实源；本轮统一 Transfer 事件发送点后为 - 95 个 producer（94 静态、1 动态）与 + 93 个 producer（92 静态、1 动态）与 17 个 consumer（16 静态、1 动态）；consumer 由不可自动写入的精确人工 policy 管理。 **目标与步骤** diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 19023dc52..0a5c9131e 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -400,8 +400,8 @@ collector 只接受 canonical `eventmanager`、`EventManager()` 及其有限别 `register`/`add_event_listener`；当前宿主有 16 个静态注册点，另保留 1 个由工作流配置驱动的 真实动态注册。`app/plugins/**` 插件副本不进入宿主事实。 -生产者与消费者共用 `scripts/architecture/event_facts.py` 这一份逐调用事实源。当前宿主有 95 个 -生产调用，其中 94 个静态解析为 96 个事件引用，只有 `Command.send_plugin_event` 的插件事件类型 +生产者与消费者共用 `scripts/architecture/event_facts.py` 这一份逐调用事实源。当前宿主有 93 个 +生产调用，其中 92 个静态解析为 94 个事件引用，只有 `Command.send_plugin_event` 的插件事件类型 保持动态；17 个消费注册中 16 个静态、1 个动态。生成的 `runtime-contract-baseline.json` 保存 line-free 事实、数量和枚举索引；人工维护的 `runtime-contract-policy.json` 只批准 consumer 的精确 fingerprint、owner 和理由，任何新增、替换、 @@ -740,7 +740,7 @@ flowchart LR stream/vendor/diagnostic/control-plane 事实是精确 containment。每条初始边的指纹由测试独立冻结， bindings/uses 变化、分类互换、通配导入和初始边增长都会失败；债务删除时同步删除冻结项以禁止恢复， `--write-host` 不会改写人工 policy 或冻结上界。 -- `event_facts` 是生产者/消费者唯一收集源；运行快照记录 95 个生产调用和 17 个消费注册， +- `event_facts` 是生产者/消费者唯一收集源；运行快照记录 93 个生产调用和 17 个消费注册， consumer 的 17 个唯一 fingerprint 另由只读人工 policy 精确准入。CI 将语义 policy 与生成快照 分成独立步骤，前者不能通过刷新后者绕过。 - 任何所有权迁移必须同步更新：canonical 导入、`app/runtime/compat/manifest.py`、 @@ -761,7 +761,7 @@ flowchart LR | Direct egress | 53（债务已清零，53 条精确 containment） | | Module Contract V2 spec | 217（其中 215 个进入 `run_module` 观察面） | | Event Contract | 53 | -| Event producer / consumer | 95（94 静态、1 动态）/ 17（16 静态、1 动态） | +| Event producer / consumer | 93（92 静态、1 动态）/ 17（16 静态、1 动态） | | Model/Oper 自动事务与自建 Session | 0 | | 组合根外 `SystemConfigOper()` | 0 | diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index bfa38f03b..fca0c534c 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -1248,8 +1248,8 @@ "no-untyped-def": 2 }, "app/chain/_messaging.py": { - "arg-type": 2, - "attr-defined": 35, + "arg-type": 1, + "attr-defined": 26, "no-any-return": 6, "no-untyped-call": 2, "no-untyped-def": 2, diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index 452e07e43..c204b9f7a 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -1829,10 +1829,8 @@ "043b1451d0995070b782b4c0e78186cbb1c019c9cfd23500cd9c84c0d1992218" ], "producer_fingerprints": [ - "233eaefcad5d7635f25d05f7962774aa30890cce9bb2850186d9ac951b20e9be", - "233eaefcad5d7635f25d05f7962774aa30890cce9bb2850186d9ac951b20e9be", - "f69f4e169daad9a6e1793edf090aa773cc69459043810e1e0d739c8de144c866", - "f69f4e169daad9a6e1793edf090aa773cc69459043810e1e0d739c8de144c866" + "268702e19e01537b7e5c13981506cd3d75696d5da749ca0bd4c32eea1d7905a8", + "42f78d867c3555fe7ce92874c84226a6cd420535cb8addaf7d9f6c97d8b58a1d" ] }, "EventType.PluginAction": { @@ -1968,11 +1966,11 @@ ] } }, - "fact_count": 112, + "fact_count": 110, "invalid_consumer_count": 0, "invalid_producer_count": 0, - "producer_call_count": 95, - "producer_event_reference_count": 96, + "producer_call_count": 93, + "producer_event_reference_count": 94, "producers": [ { "caller": "app.agent.orchestrator", @@ -2232,10 +2230,10 @@ "events": [ "EventType.NoticeMessage" ], - "fingerprint": "233eaefcad5d7635f25d05f7962774aa30890cce9bb2850186d9ac951b20e9be", + "fingerprint": "42f78d867c3555fe7ce92874c84226a6cd420535cb8addaf7d9f6c97d8b58a1d", "invalid": false, "method": "async_send_event", - "qualname": "NotificationMixin.async_post_message", + "qualname": "NotificationMixin._async_dispatch_notification_steps", "receiver_kind": "injected_event_manager" }, { @@ -2244,34 +2242,10 @@ "events": [ "EventType.NoticeMessage" ], - "fingerprint": "233eaefcad5d7635f25d05f7962774aa30890cce9bb2850186d9ac951b20e9be", - "invalid": false, - "method": "async_send_event", - "qualname": "NotificationMixin.async_post_message", - "receiver_kind": "injected_event_manager" - }, - { - "caller": "app.chain._messaging", - "dynamic": false, - "events": [ - "EventType.NoticeMessage" - ], - "fingerprint": "f69f4e169daad9a6e1793edf090aa773cc69459043810e1e0d739c8de144c866", + "fingerprint": "268702e19e01537b7e5c13981506cd3d75696d5da749ca0bd4c32eea1d7905a8", "invalid": false, "method": "send_event", - "qualname": "NotificationMixin.post_message", - "receiver_kind": "injected_event_manager" - }, - { - "caller": "app.chain._messaging", - "dynamic": false, - "events": [ - "EventType.NoticeMessage" - ], - "fingerprint": "f69f4e169daad9a6e1793edf090aa773cc69459043810e1e0d739c8de144c866", - "invalid": false, - "method": "send_event", - "qualname": "NotificationMixin.post_message", + "qualname": "NotificationMixin._dispatch_notification_steps", "receiver_kind": "injected_event_manager" }, { @@ -3116,7 +3090,7 @@ } ], "static_consumer_count": 16, - "static_producer_call_count": 94 + "static_producer_call_count": 92 }, "event_specs": { "ChainEventType.AgentLLMProvider": { diff --git a/tests/test_architecture_contract_baseline.py b/tests/test_architecture_contract_baseline.py index 507e87dcd..62febecfd 100644 --- a/tests/test_architecture_contract_baseline.py +++ b/tests/test_architecture_contract_baseline.py @@ -463,17 +463,17 @@ def test_event_contract_baseline_covers_every_public_event_enum() -> None: assert set(events["event_index"]) == expected assert events["event_count"] == len(expected) - assert events["producer_call_count"] == 95 - assert events["static_producer_call_count"] == 94 + assert events["producer_call_count"] == 93 + assert events["static_producer_call_count"] == 92 assert events["dynamic_producer_count"] == 1 assert events["invalid_producer_count"] == 0 - assert events["producer_event_reference_count"] == 96 + assert events["producer_event_reference_count"] == 94 assert events["consumer_registration_count"] == 17 assert events["static_consumer_count"] == 16 assert events["dynamic_consumer_count"] == 1 assert events["invalid_consumer_count"] == 0 assert events["consumer_event_reference_count"] == 16 - assert events["fact_count"] == 112 + assert events["fact_count"] == 110 assert len({fact["fingerprint"] for fact in events["consumers"]}) == 17 assert all( not fact["caller"].startswith("app.plugins") diff --git a/tests/test_architecture_event_facts.py b/tests/test_architecture_event_facts.py index d343bce21..22ec725e1 100644 --- a/tests/test_architecture_event_facts.py +++ b/tests/test_architecture_event_facts.py @@ -1142,7 +1142,7 @@ def handler(event): def test_collect_event_facts_matches_current_host_inventory() -> None: - """统一事实覆盖当前 95 个 producer 调用与 17 个 consumer 调用。""" + """统一事实覆盖当前 93 个 producer 调用与 17 个 consumer 调用。""" from scripts.architecture.baseline import ( _event_enum_members, discover_modules, @@ -1158,11 +1158,11 @@ def test_collect_event_facts_matches_current_host_inventory() -> None: producers = facts["producers"] consumers = facts["consumers"] - assert len(producers) == 95 - assert sum(not fact["dynamic"] and not fact["invalid"] for fact in producers) == 94 + assert len(producers) == 93 + assert sum(not fact["dynamic"] and not fact["invalid"] for fact in producers) == 92 assert sum(fact["dynamic"] for fact in producers) == 1 assert sum(fact["invalid"] for fact in producers) == 0 - assert sum(len(fact["events"]) for fact in producers) == 96 + assert sum(len(fact["events"]) for fact in producers) == 94 assert len(consumers) == 17 assert sum(not fact["dynamic"] and not fact["invalid"] for fact in consumers) == 16 assert sum(fact["dynamic"] for fact in consumers) == 1 diff --git a/tests/test_notification_routing_parity.py b/tests/test_notification_routing_parity.py new file mode 100644 index 000000000..73559ce23 --- /dev/null +++ b/tests/test_notification_routing_parity.py @@ -0,0 +1,177 @@ +import asyncio +from types import SimpleNamespace +from typing import Optional, Union +from unittest.mock import AsyncMock, Mock + +import pytest + +from app.chain._messaging import MessageProcessingMixin, NotificationMixin +from app.schemas.message import Message +from app.schemas.types import MessageType + + +class _NotificationSettingsRepository: + """记录同步和异步通知设置查询，并返回同一份测试快照。""" + + def __init__( + self, + settings: dict[str, Optional[dict[str, object]]], + *, + failure: Optional[str] = None, + ) -> None: + """保存测试设置、失败用户名和两种查询调用记录。""" + self.settings = settings + self.failure = failure + self.sync_calls: list[str] = [] + self.async_calls: list[str] = [] + + def get_notification_settings( + self, username: str + ) -> Optional[dict[str, object]]: + """同步返回指定用户的通知设置。""" + self.sync_calls.append(username) + if username == self.failure: + raise RuntimeError(f"无法读取 {username} 的通知设置") + return self.settings.get(username) + + async def async_get_notification_settings( + self, username: str + ) -> Optional[dict[str, object]]: + """异步返回指定用户的通知设置。""" + self.async_calls.append(username) + if username == self.failure: + raise RuntimeError(f"无法读取 {username} 的通知设置") + return self.settings.get(username) + + +class _NotificationHarness(MessageProcessingMixin, NotificationMixin): + """提供通知 mixin 所需的最小可观测同步和异步边界。""" + + def __init__(self, repository: _NotificationSettingsRepository) -> None: + """注入用户设置仓库及可断言的历史、事件和队列替身。""" + self.user_repository = repository + self.runtime_config = SimpleNamespace(superuser="admin") + self.messageoper = SimpleNamespace( + add=Mock(), + async_add=AsyncMock(), + exists_by_source=Mock(return_value=False), + ) + self.eventmanager = SimpleNamespace( + send_event=Mock(), + async_send_event=AsyncMock(), + ) + self.messagequeue = SimpleNamespace( + send_message=Mock(), + async_send_message=AsyncMock(), + ) + + +def _delivery_snapshot(mock: Union[Mock, AsyncMock]) -> list[tuple[dict, bool]]: + """把队列调用投影为与同步方式无关的消息和立即投递标志。""" + return [ + ( + call.kwargs["message"].model_dump(mode="json"), + bool(call.kwargs.get("immediately", False)), + ) + for call in mock.call_args_list + ] + + +def _event_snapshot(mock: Union[Mock, AsyncMock]) -> list[dict]: + """返回通知事件调用中的稳定数据载荷。""" + return [call.kwargs["data"] for call in mock.call_args_list] + + +@pytest.mark.parametrize( + ("action", "settings", "expected_users", "expected_targets"), + [ + ( + "user,all", + {"alice": {"telegram": "1001"}}, + ["alice"], + [{"telegram": "1001"}, None], + ), + ( + "user,admin", + {"alice": None, "admin": {"telegram": "9001"}}, + ["alice", "admin"], + [{"telegram": "9001"}], + ), + ( + "admin,user", + {"alice": None, "admin": {"telegram": "9001"}}, + ["admin", "alice"], + [{"telegram": "9001"}], + ), + ], +) +def test_notification_routing_sync_async_parity( + monkeypatch, + action: str, + settings: dict[str, Optional[dict[str, object]]], + expected_users: list[str], + expected_targets: list[Optional[dict[str, object]]], +) -> None: + """同步与异步入口应共享用户路由、管理员回退和原消息投递决策。""" + monkeypatch.setattr( + "app.chain._messaging.get_notification_switch", + lambda _mtype: action, + ) + sync_repository = _NotificationSettingsRepository(settings) + async_repository = _NotificationSettingsRepository(settings) + sync_chain = _NotificationHarness(sync_repository) + async_chain = _NotificationHarness(async_repository) + message = Message( + mtype=MessageType.Download, + title="下载完成", + text="影片已加入下载器", + username="alice", + ) + + sync_chain.post_message(message.model_copy(deep=True)) + asyncio.run(async_chain.async_post_message(message.model_copy(deep=True))) + + sync_deliveries = _delivery_snapshot(sync_chain.messagequeue.send_message) + async_deliveries = _delivery_snapshot( + async_chain.messagequeue.async_send_message + ) + assert sync_repository.sync_calls == expected_users + assert async_repository.async_calls == expected_users + assert [item[0]["targets"] for item in sync_deliveries] == expected_targets + assert sync_deliveries == async_deliveries + assert _event_snapshot(sync_chain.eventmanager.send_event) == _event_snapshot( + async_chain.eventmanager.async_send_event + ) + assert sync_chain.messageoper.add.call_count == 1 + assert async_chain.messageoper.async_add.await_count == 1 + + +def test_notification_routing_lookup_failure_sync_async_parity(monkeypatch) -> None: + """用户设置查询失败时两种入口都应传播错误且不得产生错误路由投递。""" + monkeypatch.setattr( + "app.chain._messaging.get_notification_switch", + lambda _mtype: "user", + ) + sync_chain = _NotificationHarness( + _NotificationSettingsRepository({}, failure="alice") + ) + async_chain = _NotificationHarness( + _NotificationSettingsRepository({}, failure="alice") + ) + message = Message( + mtype=MessageType.Download, + title="下载完成", + username="alice", + ) + + with pytest.raises(RuntimeError, match="无法读取 alice"): + sync_chain.post_message(message.model_copy(deep=True)) + with pytest.raises(RuntimeError, match="无法读取 alice"): + asyncio.run(async_chain.async_post_message(message.model_copy(deep=True))) + + sync_chain.eventmanager.send_event.assert_not_called() + async_chain.eventmanager.async_send_event.assert_not_awaited() + sync_chain.messagequeue.send_message.assert_not_called() + 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