From 40fea3236ebcec68d18fa4654f50e3e9ebb272db Mon Sep 17 00:00:00 2001 From: richard Date: Sun, 16 Aug 2026 13:40:42 +0800 Subject: [PATCH] feat(monitor): add webhook test message button Add a test button to the push-notification settings so users can verify Feishu and WeCom group-webhook configs after saving. Backend: POST /api/settings/preferences/webhook-test reads the saved webhook URL/secret and sends a one-shot test message (max_attempts=1 to skip production retry backoff). All failure modes return HTTP 200 + {ok:false, detail} so the UI renders a single path. Frontend: api.sendTestWebhook(); TestSendButton/TestResult components shared by both channels; success feedback auto-dismisses after 2s. Tests: 7 cases covering both channels, failure modes, and single-attempt behavior. Co-Authored-By: Claude --- backend/app/api/settings.py | 40 +++++++++ backend/app/services/webhook_adapter.py | 12 ++- backend/tests/test_webhook_test.py | 94 ++++++++++++++++++++++ frontend/src/lib/api.ts | 5 ++ frontend/src/pages/settings/Monitoring.tsx | 59 +++++++++++++- 5 files changed, 203 insertions(+), 7 deletions(-) create mode 100644 backend/tests/test_webhook_test.py diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 76deafd..88209e2 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -1232,6 +1232,46 @@ def update_wecom_webhook(req: WecomWebhookPrefsIn) -> dict: return {"wecom_webhook_url": saved_url} +class WebhookTestIn(BaseModel): + channel: Literal["feishu", "wecom"] + + +@router.post("/preferences/webhook-test") +def test_webhook(req: WebhookTestIn) -> dict: + """向已保存的 Webhook 地址发送一条测试消息，验证配置是否正确。 + + 只测试已保存的配置（与生产推送同源），不测试未保存草稿。 + 未配置 / 地址非法 / 发送失败均返回 HTTP 200 + {ok: False}， + 前端统一读 detail 渲染绿/红，不抛 400。 + """ + from app.services import preferences + from app.services import webhook_adapter + + title = "TickFlow Stock Panel 推送测试" + body = "如果你看到这条消息，说明推送配置正确 🎉" + + if req.channel == "feishu": + url = preferences.get_feishu_webhook_url() + if not url: + return {"ok": False, "detail": "尚未配置飞书 Webhook，请先保存"} + if not webhook_adapter.is_valid_feishu_url(url): + return {"ok": False, "detail": "已保存的飞书 Webhook 地址非法，请重新保存"} + secret = preferences.get_feishu_webhook_secret() + # 诊断用途单次尝试: 失败即返回, 不等生产退避重试 (~17s) + ok = webhook_adapter.send_feishu(url, title, body, secret, max_attempts=1) + else: # wecom + url = preferences.get_wecom_webhook_url() + if not url: + return {"ok": False, "detail": "尚未配置企业微信 Webhook，请先保存"} + if not webhook_adapter.is_valid_wecom_url(url): + return {"ok": False, "detail": "已保存的企业微信 Webhook 地址非法，请重新保存"} + ok = webhook_adapter.send_wecom(url, title, body) + + if ok: + return {"ok": True, "detail": "测试消息已发送，请到群内查收"} + return {"ok": False, "detail": "推送失败：网络不可达或地址/密钥不正确，详情见后端日志"} + + class WecomBotPrefsIn(BaseModel): bot_id: str secret: str diff --git a/backend/app/services/webhook_adapter.py b/backend/app/services/webhook_adapter.py index 8a73698..77cbfd8 100644 --- a/backend/app/services/webhook_adapter.py +++ b/backend/app/services/webhook_adapter.py @@ -93,7 +93,7 @@ def _truncate_to_bytes(text: str, max_bytes: int, suffix: str = "…") -> str: _FEISHU_MAX_ATTEMPTS = 3 -def _post_feishu(webhook_url: str, payload: dict, secret: str) -> bool: +def _post_feishu(webhook_url: str, payload: dict, secret: str, max_attempts: int = _FEISHU_MAX_ATTEMPTS) -> bool: """发送飞书 webhook 请求并判定成败 (供 text / card 共用)。 成功响应: HTTP 200 且业务 code=0 (或非 JSON/非 dict 的 200)。 @@ -102,11 +102,14 @@ def _post_feishu(webhook_url: str, payload: dict, secret: str) -> bool: 一次瞬时 5xx/timeout 若不重试, 该告警会被冷却窗口(默认 1h)压掉, 离屏用户彻底 收不到推送。永久失败 (4xx / 业务 code≠0, 如签名错、URL 失效) 不重试。最终失败 记 WARNING (而非之前的 debug), 保证「推送丢了」在日志里可见。 + + max_attempts: 尝试次数, 默认 3 (生产推送语义)。诊断用途(如手动测试配置)可传 1, + 避免失败时等满退避重试。 """ import httpx last_err = "" - for attempt in range(1, _FEISHU_MAX_ATTEMPTS + 1): + for attempt in range(1, max_attempts + 1): try: # 启用签名校验时, 请求体须带 timestamp + sign (每次重试都重算, 防时间戳过期) if secret: @@ -143,7 +146,7 @@ def _post_feishu(webhook_url: str, payload: dict, secret: str) -> bool: return False -def send_feishu(webhook_url: str, title: str, body: str, secret: str = "") -> bool: +def send_feishu(webhook_url: str, title: str, body: str, secret: str = "", max_attempts: int = _FEISHU_MAX_ATTEMPTS) -> bool: """推送一条文本消息到飞书群推送 Webhook。 Args: @@ -151,6 +154,7 @@ def send_feishu(webhook_url: str, title: str, body: str, secret: str = "") -> bo title: 消息标题 (与正文拼接为一条文本) body: 消息正文 secret: 签名密钥 (机器人启用了「签名校验」时必填; 留空则不带签名) + max_attempts: 尝试次数 (诊断用途可传 1, 默认保持生产重试语义) Returns: True=成功送达, False=失败或 URL 非法。 @@ -164,7 +168,7 @@ def send_feishu(webhook_url: str, title: str, body: str, secret: str = "") -> bo return False payload: dict = {"msg_type": "text", "content": {"text": text}} - return _post_feishu(webhook_url, payload, secret) + return _post_feishu(webhook_url, payload, secret, max_attempts) def send_feishu_card(webhook_url: str, title: str, subtitle: str, body_md: str, secret: str = "") -> bool: diff --git a/backend/tests/test_webhook_test.py b/backend/tests/test_webhook_test.py new file mode 100644 index 0000000..2b4d4e9 --- /dev/null +++ b/backend/tests/test_webhook_test.py @@ -0,0 +1,94 @@ +"""Webhook 测试消息功能 — 向已保存的 Webhook 地址发送测试消息验证配置。 + +纯逻辑，不触网（monkeypatch webhook_adapter.send_*），直接调用 settings 端点函数。 +""" +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from app.api.settings import ( + WebhookTestIn, +) +from app.api.settings import ( + test_webhook as run_webhook_test, +) + +FEISHU_URL = "https://open.feishu.cn/open-apis/bot/v2/hook/test" +WECOM_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=test-key" + + +def test_feishu_sends_saved_url_and_secret(monkeypatch): + calls = {} + monkeypatch.setattr("app.services.preferences.get_feishu_webhook_url", lambda: FEISHU_URL) + monkeypatch.setattr("app.services.preferences.get_feishu_webhook_secret", lambda: "my-secret") + monkeypatch.setattr( + "app.services.webhook_adapter.send_feishu", + lambda url, title, body, secret="", max_attempts=3: calls.update( + url=url, title=title, body=body, secret=secret, max_attempts=max_attempts, + ) or True, + ) + + result = run_webhook_test(WebhookTestIn(channel="feishu")) + + assert result["ok"] is True + assert calls["url"] == FEISHU_URL + assert calls["secret"] == "my-secret" + # 诊断路径单次尝试, 不等生产退避重试 + assert calls["max_attempts"] == 1 + assert calls["title"] == "TickFlow Stock Panel 推送测试" + assert "推送配置正确" in calls["body"] + + +def test_feishu_send_failure_returns_ok_false(monkeypatch): + monkeypatch.setattr("app.services.preferences.get_feishu_webhook_url", lambda: FEISHU_URL) + monkeypatch.setattr("app.services.preferences.get_feishu_webhook_secret", lambda: "") + monkeypatch.setattr("app.services.webhook_adapter.send_feishu", lambda *a, **k: False) + + result = run_webhook_test(WebhookTestIn(channel="feishu")) + + assert result["ok"] is False + assert "推送失败" in result["detail"] + + +@pytest.mark.parametrize("channel,url_getter", [ + ("feishu", "app.services.preferences.get_feishu_webhook_url"), + ("wecom", "app.services.preferences.get_wecom_webhook_url"), +]) +def test_not_configured_returns_ok_false(monkeypatch, channel, url_getter): + monkeypatch.setattr(url_getter, lambda: "") + + result = run_webhook_test(WebhookTestIn(channel=channel)) + + assert result["ok"] is False + assert "尚未配置" in result["detail"] + + +def test_invalid_saved_feishu_url_returns_ok_false(monkeypatch): + monkeypatch.setattr("app.services.preferences.get_feishu_webhook_url", lambda: "https://evil.example/hook/xx") + + result = run_webhook_test(WebhookTestIn(channel="feishu")) + + assert result["ok"] is False + assert "地址非法" in result["detail"] + + +def test_wecom_sends_saved_url(monkeypatch): + calls = {} + monkeypatch.setattr("app.services.preferences.get_wecom_webhook_url", lambda: WECOM_URL) + monkeypatch.setattr( + "app.services.webhook_adapter.send_wecom", + lambda url, title, body: calls.update(url=url, title=title, body=body) or True, + ) + + result = run_webhook_test(WebhookTestIn(channel="wecom")) + + assert result["ok"] is True + assert calls["url"] == WECOM_URL + assert calls["title"] == "TickFlow Stock Panel 推送测试" + assert "推送配置正确" in calls["body"] + + +def test_unknown_channel_rejected_by_pydantic(): + with pytest.raises(ValidationError): + WebhookTestIn(channel="wecom-bot") diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 5533b70..a4d00a7 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1950,6 +1950,11 @@ export const api = { method: 'PUT', body: JSON.stringify({ url }), }), + sendTestWebhook: (channel: 'feishu' | 'wecom') => + request<{ ok: boolean; detail: string }>('/api/settings/preferences/webhook-test', { + method: 'POST', + body: JSON.stringify({ channel }), + }), updateWecomBot: (botId: string, secret: string, enabled: boolean = true) => request<{ wecom_bot_id: string diff --git a/frontend/src/pages/settings/Monitoring.tsx b/frontend/src/pages/settings/Monitoring.tsx index 392d017..dd0afc9 100644 --- a/frontend/src/pages/settings/Monitoring.tsx +++ b/frontend/src/pages/settings/Monitoring.tsx @@ -177,6 +177,13 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = saveWecomWebhook.mutate(url) }, [wecomDraft, saveWecomWebhook]) + const testFeishu = useMutation({ + mutationFn: () => api.sendTestWebhook('feishu'), + }) + const testWecom = useMutation({ + mutationFn: () => api.sendTestWebhook('wecom'), + }) + // 智能机器人 (BotID + Secret) 保存 → 后端立即重建连接 const saveWecomBot = useMutation({ mutationFn: ({ botId, secret }: { botId: string; secret: string }) => @@ -503,7 +510,7 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = Webhook 地址 setFeishuDraft(e.target.value)} + onChange={e => { setFeishuDraft(e.target.value); if (!testFeishu.isPending) testFeishu.reset() }} placeholder={FEISHU_PREFIX + 'xxxxxxxx'} className="h-9 w-full rounded-btn border border-border bg-base px-3 text-xs font-mono text-foreground focus:outline-none focus:border-accent/50" /> @@ -514,7 +521,7 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = setFeishuSecretDraft(e.target.value)} + onChange={e => { setFeishuSecretDraft(e.target.value); if (!testFeishu.isPending) testFeishu.reset() }} placeholder="机器人未启用签名校验则留空" className="h-9 w-full rounded-btn border border-border bg-base px-3 text-xs font-mono text-foreground focus:outline-none focus:border-accent/50" /> @@ -532,9 +539,11 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = > {saveFeishuWebhook.isPending ? '保存中…' : '保存'} + {feishuWebhookUrl && ( ● 已配置 )} +
@@ -588,7 +597,7 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = Webhook 地址 或 Key setWecomDraft(e.target.value)} + onChange={e => { setWecomDraft(e.target.value); if (!testWecom.isPending) testWecom.reset() }} placeholder={WECOM_PREFIX + '?key=xxxxxxxx' + ' 或直接填 key'} className="h-9 w-full rounded-btn border border-border bg-base px-3 text-xs font-mono text-foreground focus:outline-none focus:border-accent/50" /> @@ -606,9 +615,11 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = > {saveWecomWebhook.isPending ? '保存中…' : '保存'} + {wecomWebhookUrl && ( ● 已配置 )} +
@@ -734,6 +745,48 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = } +// ===== 推送测试按钮 + 内联结果 ===== + +function TestSendButton({ test, configured }: { + test: { isPending: boolean; mutate: () => void } + configured: boolean +}) { + return ( + + ) +} + +function TestResult({ test }: { + test: { data?: { ok: boolean; detail: string } | null; isError: boolean; error?: Error | null; reset: () => void } +}) { + // 成功结果 2 秒后自动消失; 失败保留, 便于阅读 + useEffect(() => { + if (test.data?.ok) { + const t = window.setTimeout(test.reset, 2000) + return () => window.clearTimeout(t) + } + }, [test.data, test.reset]) + + let text: string | null = null + let tone = '' + if (test.isError) { + text = String(test.error?.message ?? '发送失败') + tone = 'text-danger' + } else if (test.data) { + text = (test.data.ok ? '✓ ' : '✗ ') + test.data.detail + tone = test.data.ok ? 'text-emerald-500' : 'text-danger' + } + if (!text) return null + return {text} +} + // ===== ToggleRow ===== function ToggleRow({