From 5b0c631f8022a4050f70ff00b3588e0cb8e37935 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 16 Aug 2026 06:35:46 +0800 Subject: [PATCH] =?UTF-8?q?refactor(notification):=20=E5=BB=BA=E7=AB=8B?= =?UTF-8?q?=E9=80=9A=E7=9F=A5=E6=B8=A0=E9=81=93=E9=80=9A=E7=94=A8=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E5=A5=91=E7=BA=A6=20channel=5Fmanage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - schemas 层新增 NotificationAction 公共动作词汇表（状态/二维码/退出/连通性/缓存迁移） - 模块实现统一 channel_manage(channel, action, **params)，按渠道名路由， 非本渠道返回 None；动作语义、表单参数解释与临时参数初始化封闭在模块内 - NotificationChain 收敛为单一 manage_channel 纯透明转发， chain 层不再出现任何渠道名与模块特色 - endpoint 路由路径保持前端兼容，内部统一走通用接口 - 新增 6 项契约守护测试，文档记录通用模式：新渠道接入无需改 chain --- app/api/endpoints/notification.py | 27 ++- app/chain/notification.py | 122 +++---------- app/modules/wechatclawbot/__init__.py | 198 +++++++--------------- app/schemas/types.py | 18 ++ docs/rules/05-architecture.md | 9 + tests/test_notification_channel_manage.py | 120 +++++++++++++ 6 files changed, 242 insertions(+), 252 deletions(-) create mode 100644 tests/test_notification_channel_manage.py diff --git a/app/api/endpoints/notification.py b/app/api/endpoints/notification.py index c1ac448de..fbcd9bf28 100644 --- a/app/api/endpoints/notification.py +++ b/app/api/endpoints/notification.py @@ -7,6 +7,7 @@ from app.api.response import ResponseAPIRouter from app.chain.notification import NotificationChain from app.db.models import User from app.api.deps import get_current_active_superuser +from app.schemas.types import MessageChannel, NotificationAction router = ResponseAPIRouter() @@ -27,15 +28,17 @@ def wechatclawbot_status( _: User = Depends(get_current_active_superuser), ): """查询微信 ClawBot 登录状态和二维码。""" - result = NotificationChain().get_wechatclawbot_status( + result = NotificationChain().manage_channel( + channel=MessageChannel.WechatClawBot, + action=NotificationAction.STATUS, source=source, fallback_source=fallback_source, + refresh_remote=refresh_remote, + auto_generate_qrcode=auto_generate_qrcode, WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - refresh_remote=refresh_remote, - auto_generate_qrcode=auto_generate_qrcode, ) return schemas.Response( success=bool(result.get("success")), @@ -59,7 +62,9 @@ def refresh_wechatclawbot_qrcode( _: User = Depends(get_current_active_superuser), ): """刷新微信 ClawBot 二维码。""" - result = NotificationChain().refresh_wechatclawbot_qrcode( + result = NotificationChain().manage_channel( + channel=MessageChannel.WechatClawBot, + action=NotificationAction.REFRESH_QRCODE, source=source, fallback_source=fallback_source, WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, @@ -89,7 +94,9 @@ def logout_wechatclawbot( _: User = Depends(get_current_active_superuser), ): """退出微信 ClawBot 登录。""" - result = NotificationChain().logout_wechatclawbot( + result = NotificationChain().manage_channel( + channel=MessageChannel.WechatClawBot, + action=NotificationAction.LOGOUT, source=source, fallback_source=fallback_source, WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, @@ -119,7 +126,9 @@ def test_wechatclawbot( _: User = Depends(get_current_active_superuser), ): """测试微信 ClawBot 当前登录态是否可用。""" - result = NotificationChain().test_wechatclawbot_connection( + result = NotificationChain().manage_channel( + channel=MessageChannel.WechatClawBot, + action=NotificationAction.TEST_CONNECTION, source=source, fallback_source=fallback_source, WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, @@ -143,10 +152,12 @@ def migrate_wechatclawbot_cache( _: User = Depends(get_current_active_superuser), ): """在通知名称变更时迁移对应的微信 ClawBot 登录缓存。""" - success, message = NotificationChain().migrate_wechatclawbot_cache( + result = NotificationChain().manage_channel( + channel=MessageChannel.WechatClawBot, + action=NotificationAction.MIGRATE_CACHE, old_name=old_source, new_name=new_source, cleanup_old=cleanup_old, overwrite=overwrite, ) - return schemas.Response(success=success, message=message) + return schemas.Response(success=bool(result.get("success")), message=result.get("message")) diff --git a/app/chain/notification.py b/app/chain/notification.py index 39c671a07..2b0056464 100644 --- a/app/chain/notification.py +++ b/app/chain/notification.py @@ -1,114 +1,30 @@ -from typing import Any, Dict, Tuple +from typing import Any, Dict from app.chain import ChainBase +from app.schemas.types import MessageChannel, NotificationAction class NotificationChain(ChainBase): """ - 通知渠道管理链，仅做模块方法名契约的薄分发，渠道连接与能力全部封闭在模块内部 + 通知渠道管理链，仅按渠道名透明转发管理动作到模块 + + 不包含任何渠道特定逻辑：动作语义、表单参数解释、客户端实例解析与 + 临时参数初始化全部封闭在实现 channel_manage 契约的模块内部 """ - def get_wechatclawbot_status( + def manage_channel( self, - source=None, - fallback_source=None, - WECHATCLAWBOT_BASE_URL=None, - WECHATCLAWBOT_DEFAULT_TARGET=None, - WECHATCLAWBOT_ADMINS=None, - WECHATCLAWBOT_POLL_TIMEOUT=None, - refresh_remote: bool = True, - auto_generate_qrcode: bool = True, + channel: MessageChannel, + action: NotificationAction, + **params: Any, ) -> Dict[str, Any]: - """查询微信 ClawBot 登录状态与二维码。""" - result = self.run_module( - "wechatclawbot_status", - source=source, - fallback_source=fallback_source, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - refresh_remote=refresh_remote, - auto_generate_qrcode=auto_generate_qrcode, - ) - return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"} + """ + 对指定通知渠道执行管理动作 - def refresh_wechatclawbot_qrcode( - self, - source=None, - fallback_source=None, - WECHATCLAWBOT_BASE_URL=None, - WECHATCLAWBOT_DEFAULT_TARGET=None, - WECHATCLAWBOT_ADMINS=None, - WECHATCLAWBOT_POLL_TIMEOUT=None, - ) -> Dict[str, Any]: - """刷新微信 ClawBot 登录二维码。""" - result = self.run_module( - "wechatclawbot_refresh_qrcode", - source=source, - fallback_source=fallback_source, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - ) - return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"} - - def logout_wechatclawbot( - self, - source=None, - fallback_source=None, - WECHATCLAWBOT_BASE_URL=None, - WECHATCLAWBOT_DEFAULT_TARGET=None, - WECHATCLAWBOT_ADMINS=None, - WECHATCLAWBOT_POLL_TIMEOUT=None, - ) -> Dict[str, Any]: - """退出微信 ClawBot 登录。""" - result = self.run_module( - "wechatclawbot_logout", - source=source, - fallback_source=fallback_source, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - ) - return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"} - - def test_wechatclawbot_connection( - self, - source=None, - fallback_source=None, - WECHATCLAWBOT_BASE_URL=None, - WECHATCLAWBOT_DEFAULT_TARGET=None, - WECHATCLAWBOT_ADMINS=None, - WECHATCLAWBOT_POLL_TIMEOUT=None, - ) -> Dict[str, Any]: - """测试微信 ClawBot 当前登录态是否可用。""" - result = self.run_module( - "wechatclawbot_test_connection", - source=source, - fallback_source=fallback_source, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - ) - return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"} - - def migrate_wechatclawbot_cache( - self, - old_name: str, - new_name: str, - cleanup_old: bool = False, - overwrite: bool = False, - ) -> Tuple[bool, str]: - """在通知名称变更时迁移对应的微信 ClawBot 登录缓存。""" - result = self.run_module( - "wechatclawbot_migrate_cache", - old_name=old_name, - new_name=new_name, - cleanup_old=cleanup_old, - overwrite=overwrite, - ) - return result or (False, "微信 ClawBot 通知未启用") + :param channel: 渠道标识，用于模块路由 + :param action: 通用管理动作，具体语义由渠道模块解释 + :param params: 表单与动作参数，原样透传给模块 + :return: 统一结构 {"success": bool, "message": ..., ...} + """ + result = self.run_module("channel_manage", channel=channel, action=action, **params) + return result or {"success": False, "message": "该通知渠道未启用或不支持此管理动作"} diff --git a/app/modules/wechatclawbot/__init__.py b/app/modules/wechatclawbot/__init__.py index 17ba77de2..20e0f32d0 100644 --- a/app/modules/wechatclawbot/__init__.py +++ b/app/modules/wechatclawbot/__init__.py @@ -12,7 +12,7 @@ from app.runtime.log import logger from app.modules import _MessageBase, _ModuleBase from app.modules.wechatclawbot.wechatclawbot import WechatClawBot from app.schemas import CommingMessage, Notification -from app.schemas.types import MessageChannel, ModuleType +from app.schemas.types import MessageChannel, ModuleType, NotificationAction register_channel_admin_resolver( @@ -83,22 +83,60 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]): """初始化模块设置。""" pass - def _resolve_client( + def channel_manage( self, - source: Optional[str] = None, - fallback_source: Optional[str] = None, - WECHATCLAWBOT_BASE_URL: Optional[str] = None, - WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None, - WECHATCLAWBOT_ADMINS: Optional[str] = None, - WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None, - ) -> Tuple[Optional[Any], Optional[str]]: + channel: MessageChannel, + action: NotificationAction, + **params: Any, + ) -> Optional[Dict[str, Any]]: + """通知渠道通用管理入口，按渠道名路由，仅处理本渠道。 + + 动作语义与表单参数全部由模块自行解释：优先使用已保存配置实例， + 无匹配配置时可基于表单参数构造临时实例（未保存配置的扫码预览）。 + 统一返回 {"success": bool, "message": ..., ...} 结构。 + """ + if channel != self.get_subtype(): + return None + try: + action = NotificationAction(action) + except ValueError: + return {"success": False, "message": f"不支持的渠道管理动作：{action}"} + + if action == NotificationAction.MIGRATE_CACHE: + success, message = WechatClawBot.migrate_cached_state( + old_name=params.get("old_name"), + new_name=params.get("new_name"), + cleanup_old=bool(params.get("cleanup_old")), + overwrite=bool(params.get("overwrite")), + ) + return {"success": success, "message": message} + + client, errmsg = self._resolve_client(params) + if not client: + return {"success": False, "message": errmsg} + + if action == NotificationAction.STATUS: + return client.get_status( + refresh_remote=bool(params.get("refresh_remote", True)), + auto_generate_qrcode=bool(params.get("auto_generate_qrcode", True)), + ) + if action == NotificationAction.REFRESH_QRCODE: + return client.refresh_qrcode() + if action == NotificationAction.LOGOUT: + return client.logout() + if action == NotificationAction.TEST_CONNECTION: + state, message = client.test_connection() + return {"success": state, "message": message} + return {"success": False, "message": f"不支持的渠道管理动作：{action.value}"} + + def _resolve_client(self, params: Dict[str, Any]) -> Tuple[Optional[Any], Optional[str]]: """解析微信 ClawBot 客户端实例，返回 (客户端, 错误信息)。 优先使用已加载的配置实例，均无配置时退回到基于表单参数的临时客户端， 用于未保存配置的扫码状态预览。 """ - source_name = str(source or "").strip() or None - fallback_name = str(fallback_source or "").strip() or None + source_name = str(params.get("source") or "").strip() or None + fallback_name = str(params.get("fallback_source") or "").strip() or None candidate_names = [] for candidate in (fallback_name, source_name): @@ -117,13 +155,7 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]): if client: return client, None - temp_client = self._build_temp_client( - source=source_name or fallback_name, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - ) + temp_client = self._build_temp_client(params) if temp_client: return temp_client, None @@ -131,136 +163,20 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]): return None, f"未找到名为 {source_name} 的微信 ClawBot 通知配置" return None, "微信 ClawBot 通知未启用或配置尚未保存，请先保存并启用当前渠道" - def wechatclawbot_status( - self, - source: Optional[str] = None, - fallback_source: Optional[str] = None, - WECHATCLAWBOT_BASE_URL: Optional[str] = None, - WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None, - WECHATCLAWBOT_ADMINS: Optional[str] = None, - WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None, - refresh_remote: bool = True, - auto_generate_qrcode: bool = True, - ) -> Dict[str, Any]: - """查询微信 ClawBot 登录状态与二维码，实例解析全部封闭在模块内部。""" - client, errmsg = self._resolve_client( - source=source, - fallback_source=fallback_source, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - ) - if not client: - return {"success": False, "message": errmsg} - return client.get_status( - refresh_remote=refresh_remote, - auto_generate_qrcode=auto_generate_qrcode, - ) - - def wechatclawbot_refresh_qrcode( - self, - source: Optional[str] = None, - fallback_source: Optional[str] = None, - WECHATCLAWBOT_BASE_URL: Optional[str] = None, - WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None, - WECHATCLAWBOT_ADMINS: Optional[str] = None, - WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None, - ) -> Dict[str, Any]: - """刷新微信 ClawBot 登录二维码。""" - client, errmsg = self._resolve_client( - source=source, - fallback_source=fallback_source, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - ) - if not client: - return {"success": False, "message": errmsg} - return client.refresh_qrcode() - - def wechatclawbot_logout( - self, - source: Optional[str] = None, - fallback_source: Optional[str] = None, - WECHATCLAWBOT_BASE_URL: Optional[str] = None, - WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None, - WECHATCLAWBOT_ADMINS: Optional[str] = None, - WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None, - ) -> Dict[str, Any]: - """退出微信 ClawBot 登录。""" - client, errmsg = self._resolve_client( - source=source, - fallback_source=fallback_source, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - ) - if not client: - return {"success": False, "message": errmsg} - return client.logout() - - def wechatclawbot_test_connection( - self, - source: Optional[str] = None, - fallback_source: Optional[str] = None, - WECHATCLAWBOT_BASE_URL: Optional[str] = None, - WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None, - WECHATCLAWBOT_ADMINS: Optional[str] = None, - WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None, - ) -> Dict[str, Any]: - """测试微信 ClawBot 当前登录态是否可用。""" - client, errmsg = self._resolve_client( - source=source, - fallback_source=fallback_source, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, - ) - if not client: - return {"success": False, "message": errmsg} - state, message = client.test_connection() - return {"success": state, "message": message} - - def _build_temp_client( - self, - source: Optional[str] = None, - WECHATCLAWBOT_BASE_URL: Optional[str] = None, - WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None, - WECHATCLAWBOT_ADMINS: Optional[str] = None, - WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None, - ): - """基于当前表单配置创建一个临时客户端，用于未保存时的扫码状态预览。""" - source_name = str(source or "").strip() + def _build_temp_client(self, params: Dict[str, Any]) -> Optional[Any]: + """基于表单参数创建临时客户端，用于未保存配置时的扫码状态预览。""" + source_name = str(params.get("source") or params.get("fallback_source") or "").strip() if not source_name: return None return WechatClawBot( name=source_name, - WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL, - WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET, - WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS, - WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT, + WECHATCLAWBOT_BASE_URL=params.get("WECHATCLAWBOT_BASE_URL"), + WECHATCLAWBOT_DEFAULT_TARGET=params.get("WECHATCLAWBOT_DEFAULT_TARGET"), + WECHATCLAWBOT_ADMINS=params.get("WECHATCLAWBOT_ADMINS"), + WECHATCLAWBOT_POLL_TIMEOUT=params.get("WECHATCLAWBOT_POLL_TIMEOUT"), auto_start_polling=False, ) - def wechatclawbot_migrate_cache( - self, - old_name: str, - new_name: str, - cleanup_old: bool = False, - overwrite: bool = False, - ): - """在通知名称变更时迁移对应的微信 ClawBot 登录缓存。""" - return WechatClawBot.migrate_cached_state( - old_name=old_name, - new_name=new_name, - cleanup_old=cleanup_old, - overwrite=overwrite, - ) - @staticmethod def _load_json(body: Any) -> Optional[dict]: """将内容解析为 JSON 字典。""" diff --git a/app/schemas/types.py b/app/schemas/types.py index 26fbfd7f4..dfa05dd49 100644 --- a/app/schemas/types.py +++ b/app/schemas/types.py @@ -491,6 +491,24 @@ class MessageChannel(Enum): QQ = "QQ" +class NotificationAction(str, Enum): + """ + 通知渠道通用管理动作 + + 作为渠道管理契约的公共词汇表，具体动作的支持范围与参数语义由渠道模块自行解释 + """ + # 查询登录状态与二维码 + STATUS = "status" + # 刷新登录二维码 + REFRESH_QRCODE = "refresh_qrcode" + # 退出登录 + LOGOUT = "logout" + # 测试连通性 + TEST_CONNECTION = "test_connection" + # 迁移渠道名变更前的登录缓存 + MIGRATE_CACHE = "migrate_cache" + + # 下载器类型 class DownloaderType(Enum): # Qbittorrent diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index f60a190d3..31aa20763 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -212,6 +212,15 @@ exceptions and value domains used by both modules and upper layers live in method names. The directory remains unchanged because discovery and plugin code depend on this established runtime root. +Channels that need login management or temporary-parameter initialization +follow one generic contract instead of per-channel APIs: modules implement +`channel_manage(channel, action, **params)`, route by the requested +`MessageChannel` (returning `None` for other channels), and interpret actions +from the shared `schemas.types.NotificationAction` vocabulary plus opaque form +parameters themselves. `NotificationChain.manage_channel` forwards transparently +and must stay free of any channel-specific names or logic; new channels adopt +the same contract without touching the chain. + ### DB / Oper layer SQLAlchemy models stay under `app/db/models/`; the data access classes live in diff --git a/tests/test_notification_channel_manage.py b/tests/test_notification_channel_manage.py new file mode 100644 index 000000000..e92f8004f --- /dev/null +++ b/tests/test_notification_channel_manage.py @@ -0,0 +1,120 @@ +""" +通知渠道通用管理契约（channel_manage）守护测试 + +验证通用模式的三条核心性质： +1. 按渠道名路由：非本渠道直接返回 None，交由分发机制继续执行其它模块 +2. 动作词汇表由 schemas 契约层统一定义，未支持动作返回统一错误结构 +3. 临时参数初始化封闭在模块内：无已保存配置时可基于表单参数构造临时客户端 +""" +from types import SimpleNamespace + +import pytest + +from app.modules.wechatclawbot import WechatClawBotModule +from app.schemas.types import MessageChannel, NotificationAction + + +@pytest.fixture +def module(): + return WechatClawBotModule() + + +def test_channel_manage_routes_only_matching_channel(module): + """非本渠道的管理请求返回 None，run_module 分发将继续执行其它模块。""" + result = module.channel_manage( + channel=MessageChannel.Telegram, + action=NotificationAction.STATUS, + ) + assert result is None + + +def test_channel_manage_rejects_unknown_action(module): + """动作词汇表之外的请求返回统一错误结构。""" + result = module.channel_manage( + channel=MessageChannel.WechatClawBot, + action="not_an_action", + ) + assert result["success"] is False + assert "不支持" in result["message"] + + +def test_channel_manage_requires_saved_config_without_form_params(module, monkeypatch): + """无任何配置且未提供表单参数时，返回提示保存配置的错误。""" + monkeypatch.setattr(module, "get_instance", lambda name=None: None) + result = module.channel_manage( + channel=MessageChannel.WechatClawBot, + action=NotificationAction.TEST_CONNECTION, + ) + assert result["success"] is False + assert "保存" in result["message"] + + +def test_channel_manage_builds_temporary_client_from_form_params(module, monkeypatch): + """无已保存配置时基于表单参数构造临时客户端，实现未保存即预览。""" + monkeypatch.setattr(module, "get_instance", lambda name=None: None) + + captured = {} + + class _FakeClient: + def test_connection(self): + captured["called"] = True + return False, "未登录，请先扫码完成绑定" + + monkeypatch.setattr( + "app.modules.wechatclawbot.WechatClawBot", + lambda **kwargs: (captured.update(kwargs=kwargs), _FakeClient())[1], + ) + + result = module.channel_manage( + channel=MessageChannel.WechatClawBot, + action=NotificationAction.TEST_CONNECTION, + source="预览渠道", + WECHATCLAWBOT_BASE_URL="http://127.0.0.1:1", + ) + + assert captured["called"] is True + assert captured["kwargs"]["name"] == "预览渠道" + assert captured["kwargs"]["auto_start_polling"] is False + assert result["success"] is False + assert "未登录" in result["message"] + + +def test_channel_manage_prefers_saved_instance(module, monkeypatch): + """存在已保存配置实例时优先使用，不构造临时客户端。""" + saved = SimpleNamespace() + saved.get_status = lambda refresh_remote, auto_generate_qrcode: { + "success": True, + "connected": True, + } + monkeypatch.setattr(module, "get_config", lambda name=None: SimpleNamespace(name="已保存")) + monkeypatch.setattr(module, "get_instance", lambda name=None: saved) + + result = module.channel_manage( + channel=MessageChannel.WechatClawBot, + action=NotificationAction.STATUS, + source="已保存", + ) + assert result["success"] is True + assert result["connected"] is True + + +def test_channel_manage_migrate_cache_dispatches_without_client(module, monkeypatch): + """缓存迁移动作不依赖客户端实例，直接走静态迁移逻辑。""" + + def _fake_migrate(old_name, new_name, cleanup_old, overwrite): + assert old_name == "旧名" + assert new_name == "新名" + return True, "迁移成功" + + monkeypatch.setattr( + "app.modules.wechatclawbot.WechatClawBot.migrate_cached_state", + staticmethod(_fake_migrate), + ) + + result = module.channel_manage( + channel=MessageChannel.WechatClawBot, + action=NotificationAction.MIGRATE_CACHE, + old_name="旧名", + new_name="新名", + ) + assert result == {"success": True, "message": "迁移成功"}