mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 17:54:15 +08:00
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 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user