shaguabijia-app-server/tests/test_feishu_notifier.py

"""飞书群机器人发送: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
        )


# ---- send_feishu_card ----

def test_send_card_payload_structure(monkeypatch):
    """send_feishu_card 发出 msg_type=interactive + card 字段的 payload。"""
    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)
    card = {
        "schema": "2.0",
        "header": {"title": {"tag": "plain_text", "content": "🚨 比价失败报警 · test"}, "template": "red"},
        "body": {"elements": [{"tag": "markdown", "content": "近 15 分钟"}]},
    }
    feishu_notifier.send_feishu_card("https://open.feishu.cn/hook/zzz", card, timeout=5.0)
    assert captured["url"] == "https://open.feishu.cn/hook/zzz"
    assert captured["json"]["msg_type"] == "interactive"
    assert captured["json"]["card"] is card


def test_send_card_uses_default_timeout(monkeypatch):
    """send_feishu_card 默认 timeout=10.0。"""
    captured = {}

    def fake_post(url, json, timeout):
        captured["timeout"] = timeout
        return httpx.Response(200, json={"code": 0, "msg": "success"})

    monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
    feishu_notifier.send_feishu_card("https://hook", {"schema": "2.0"})
    assert captured["timeout"] == 10.0


def test_send_card_raises_on_code_nonzero(monkeypatch):
    """send_feishu_card 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_card("https://hook", {"schema": "2.0"})


def test_send_card_raises_on_http_error(monkeypatch):
    """send_feishu_card 非 2xx 抛 FeishuNotifyError。"""
    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_card("https://hook", {"schema": "2.0"})


def test_send_card_raises_on_network_error(monkeypatch):
    """send_feishu_card 网络异常抛 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_card("https://hook", {"schema": "2.0"})