guke
40 lines
1.5 KiB
Python
40 lines
1.5 KiB
Python
"""飞书群机器人发送:消息体格式 / 关键词失败 / 网络失败,均 monkeypatch httpx 不真发。"""
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.integrations import feishu_notifier
|
|
|
|
|
|
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)
|