guke
104 lines
4.0 KiB
Python
104 lines
4.0 KiB
Python
"""飞书群机器人发送:text/post 消息体格式 / 关键词失败 / 网络失败,均 monkeypatch httpx 不真发。"""
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.integrations import feishu_notifier
|
|
|
|
# ---- send_feishu_text ----
|
|
|
|
def test_send_posts_text_payload(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_post(url, json, timeout):
|
|
captured["url"] = url
|
|
captured["json"] = json
|
|
return httpx.Response(200, json={"code": 0, "msg": "success"})
|
|
|
|
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
|
|
feishu_notifier.send_feishu_text("https://open.feishu.cn/hook/xxx", "比价失败报警 · test", timeout=5.0)
|
|
assert captured["url"] == "https://open.feishu.cn/hook/xxx"
|
|
assert captured["json"] == {"msg_type": "text", "content": {"text": "比价失败报警 · test"}}
|
|
|
|
|
|
def test_send_raises_on_keyword_rejection(monkeypatch):
|
|
def fake_post(url, json, timeout):
|
|
return httpx.Response(200, json={"code": 19024, "msg": "Key Words Not Found"})
|
|
|
|
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
|
|
with pytest.raises(feishu_notifier.FeishuNotifyError):
|
|
feishu_notifier.send_feishu_text("https://hook", "无关键词文本", timeout=5.0)
|
|
|
|
|
|
def test_send_raises_on_http_error(monkeypatch):
|
|
def fake_post(url, json, timeout):
|
|
return httpx.Response(500, text="boom")
|
|
|
|
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
|
|
with pytest.raises(feishu_notifier.FeishuNotifyError):
|
|
feishu_notifier.send_feishu_text("https://hook", "比价失败报警", timeout=5.0)
|
|
|
|
|
|
# ---- send_feishu_post ----
|
|
|
|
def test_send_post_payload_structure(monkeypatch):
|
|
"""send_feishu_post 发出 msg_type=post 的 payload,结构符合飞书 post 格式。"""
|
|
captured = {}
|
|
|
|
def fake_post(url, json, timeout):
|
|
captured["url"] = url
|
|
captured["json"] = json
|
|
return httpx.Response(200, json={"code": 0, "msg": "success"})
|
|
|
|
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
|
|
content = [[{"tag": "text", "text": "合计 3 条"}]]
|
|
feishu_notifier.send_feishu_post(
|
|
"https://open.feishu.cn/hook/yyy",
|
|
"🚨 比价失败报警 · 2026-08-04",
|
|
content,
|
|
timeout=5.0,
|
|
)
|
|
assert captured["url"] == "https://open.feishu.cn/hook/yyy"
|
|
payload = captured["json"]
|
|
assert payload["msg_type"] == "post"
|
|
zh_cn = payload["content"]["post"]["zh_cn"]
|
|
assert zh_cn["title"] == "🚨 比价失败报警 · 2026-08-04"
|
|
assert zh_cn["content"] == content
|
|
|
|
|
|
def test_send_post_raises_on_code_nonzero(monkeypatch):
|
|
"""send_feishu_post code!=0 时抛 FeishuNotifyError。"""
|
|
def fake_post(url, json, timeout):
|
|
return httpx.Response(200, json={"code": 19024, "msg": "Key Words Not Found"})
|
|
|
|
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
|
|
with pytest.raises(feishu_notifier.FeishuNotifyError):
|
|
feishu_notifier.send_feishu_post(
|
|
"https://hook", "🚨 比价失败报警", [[{"tag": "text", "text": "test"}]], timeout=5.0
|
|
)
|
|
|
|
|
|
def test_send_post_raises_on_http_error(monkeypatch):
|
|
"""send_feishu_post 非 2xx 抛 FeishuNotifyError。"""
|
|
def fake_post(url, json, timeout):
|
|
return httpx.Response(500, text="internal server error")
|
|
|
|
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
|
|
with pytest.raises(feishu_notifier.FeishuNotifyError):
|
|
feishu_notifier.send_feishu_post(
|
|
"https://hook", "🚨 比价失败报警", [[{"tag": "text", "text": "test"}]], timeout=5.0
|
|
)
|
|
|
|
|
|
def test_send_post_raises_on_network_error(monkeypatch):
|
|
"""send_feishu_post 网络异常抛 FeishuNotifyError。"""
|
|
def fake_post(url, json, timeout):
|
|
raise httpx.ConnectError("connection refused")
|
|
|
|
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
|
|
with pytest.raises(feishu_notifier.FeishuNotifyError):
|
|
feishu_notifier.send_feishu_post(
|
|
"https://hook", "🚨 比价失败报警", [[{"tag": "text", "text": "test"}]], timeout=5.0
|
|
)
|