213 lines
8.3 KiB
Python
213 lines
8.3 KiB
Python
"""工具层测试：notifier 推送逻辑（mock外部HTTP）"""
|
||
import json
|
||
from unittest.mock import patch, MagicMock
|
||
import pytest
|
||
from app.utils.notifier import (
|
||
send_wecom_robot,
|
||
send_wecom_app,
|
||
send_mail,
|
||
push_alert,
|
||
_build_content,
|
||
push_pending_alerts,
|
||
)
|
||
|
||
|
||
class TestWecomRobot:
|
||
"""企业微信群机器人推送测试（mock HTTP）"""
|
||
|
||
@patch("urllib.request.urlopen")
|
||
def test_send_success(self, mock_urlopen):
|
||
"""推送成功"""
|
||
mock_resp = MagicMock()
|
||
mock_resp.read.return_value = json.dumps({"errcode": 0}).encode()
|
||
mock_urlopen.return_value.__enter__.return_value = mock_resp
|
||
|
||
result = send_wecom_robot("https://qyapi.weixin.qq.com/webhook/test",
|
||
"测试标题", "测试内容", "red")
|
||
assert result["success"] is True
|
||
assert "已推送" in result["message"]
|
||
|
||
@patch("urllib.request.urlopen")
|
||
def test_send_api_error(self, mock_urlopen):
|
||
"""企微API返回错误"""
|
||
mock_resp = MagicMock()
|
||
mock_resp.read.return_value = json.dumps({"errcode": 40001, "errmsg": "invalid webhook"}).encode()
|
||
mock_urlopen.return_value.__enter__.return_value = mock_resp
|
||
|
||
result = send_wecom_robot("https://qyapi.weixin.qq.com/webhook/test",
|
||
"测试", "内容", "yellow")
|
||
assert result["success"] is False
|
||
assert "推送失败" in result["message"]
|
||
|
||
@patch("urllib.request.urlopen")
|
||
def test_send_network_error(self, mock_urlopen):
|
||
"""网络异常"""
|
||
mock_urlopen.side_effect = Exception("Connection refused")
|
||
|
||
result = send_wecom_robot("https://qyapi.weixin.qq.com/webhook/test",
|
||
"测试", "内容", "green")
|
||
assert result["success"] is False
|
||
assert "推送异常" in result["message"]
|
||
|
||
def test_build_content(self):
|
||
"""_build_content 内容构建"""
|
||
alert = {
|
||
"kpi_name": "销售总额",
|
||
"period": "2026-06",
|
||
"actual_value": "800,000",
|
||
"target_value": "1,000,000",
|
||
"alert_level": "red",
|
||
"resolution": "加强销售推广",
|
||
}
|
||
content = _build_content(alert)
|
||
assert "销售总额" in content
|
||
assert "2026-06" in content
|
||
assert "800,000" in content
|
||
assert "加强销售推广" in content
|
||
assert "紧急" in content
|
||
|
||
|
||
class TestPushAlert:
|
||
"""主推送函数测试"""
|
||
|
||
@patch("app.utils.notifier.send_wecom_robot")
|
||
def test_push_to_wecom_robot(self, mock_send):
|
||
"""推送到企微机器人渠道"""
|
||
mock_send.return_value = {"success": True, "message": "ok"}
|
||
channels = [{
|
||
"name": "测试群", "channel_type": "wecom", "enabled": True,
|
||
"config": {"webhook_url": "https://qyapi.weixin.qq.com/webhook/abc"}
|
||
}]
|
||
results = push_alert({"alert_level": "yellow", "alert_message": "测试"}, channels)
|
||
assert len(results) == 1
|
||
assert results[0]["channel"] == "wecom"
|
||
assert results[0]["success"] is True
|
||
|
||
@patch("app.utils.notifier.send_wecom_app")
|
||
def test_push_to_wecom_app(self, mock_send):
|
||
"""推送到企微应用消息"""
|
||
mock_send.return_value = {"success": True, "message": "ok"}
|
||
channels = [{
|
||
"name": "应用通知", "channel_type": "wecom_app", "enabled": True,
|
||
"config": {"corp_id": "xxx", "corp_secret": "yyy", "agent_id": "1000002", "touser": "@all"}
|
||
}]
|
||
results = push_alert({"alert_level": "yellow", "alert_message": "测试"}, channels)
|
||
assert len(results) == 1
|
||
assert results[0]["channel"] == "wecom_app"
|
||
|
||
@patch("app.utils.notifier.send_mail")
|
||
def test_push_to_mail(self, mock_send):
|
||
"""推送到邮件"""
|
||
mock_send.return_value = {"success": True, "message": "ok"}
|
||
channels = [{
|
||
"name": "通知邮箱", "channel_type": "mail", "enabled": True,
|
||
"config": {"to": ["admin@example.com"]}
|
||
}]
|
||
results = push_alert({"alert_level": "red", "alert_message": "测试"}, channels)
|
||
assert len(results) == 1
|
||
assert results[0]["channel"] == "mail"
|
||
|
||
def test_disabled_channel_skipped(self):
|
||
"""已禁用的渠道跳过"""
|
||
channels = [{
|
||
"name": "禁用渠道", "channel_type": "wecom", "enabled": False,
|
||
"config": {}
|
||
}]
|
||
results = push_alert({"alert_level": "yellow", "alert_message": "测试"}, channels)
|
||
assert len(results) == 0
|
||
|
||
def test_empty_channels(self):
|
||
"""无渠道 → 空结果"""
|
||
results = push_alert({"alert_level": "yellow", "alert_message": "测试"}, [])
|
||
assert results == []
|
||
|
||
|
||
class TestSendWecomApp:
|
||
"""企微应用消息测试"""
|
||
|
||
@patch("requests.get")
|
||
@patch("requests.post")
|
||
def test_app_send_success(self, mock_post, mock_get):
|
||
"""企微应用消息发送成功"""
|
||
mock_get.return_value.json.return_value = {"errcode": 0, "access_token": "fake_token"}
|
||
mock_post.return_value.json.return_value = {"errcode": 0}
|
||
|
||
result = send_wecom_app("corp123", "secret456", "1000001",
|
||
"@all", "标题", "内容", "yellow")
|
||
assert result["success"] is True
|
||
|
||
@patch("requests.get")
|
||
def test_app_token_fail(self, mock_get):
|
||
"""获取token失败"""
|
||
mock_get.return_value.json.return_value = {"errcode": 40013, "errmsg": "invalid corpid"}
|
||
|
||
result = send_wecom_app("bad_corp", "bad_secret", "1000001",
|
||
"@all", "标题", "内容", "yellow")
|
||
assert result["success"] is False
|
||
assert "获取token失败" in result["message"]
|
||
|
||
|
||
class TestSendMail:
|
||
"""邮件推送测试（mock SMTP）"""
|
||
|
||
@patch("smtplib.SMTP_SSL")
|
||
def test_mail_send_success(self, mock_smtp):
|
||
"""邮件发送成功"""
|
||
mock_server = MagicMock()
|
||
mock_smtp.return_value = mock_server
|
||
|
||
config = {"host": "smtp.example.com", "port": 465, "user": "user@example.com",
|
||
"password": "pass", "use_ssl": True, "from_addr": "user@example.com"}
|
||
result = send_mail(config, ["admin@example.com"], "测试标题", "测试内容")
|
||
assert result["success"] is True
|
||
mock_server.login.assert_called_once()
|
||
mock_server.sendmail.assert_called_once()
|
||
|
||
@patch("smtplib.SMTP_SSL")
|
||
def test_mail_send_failure(self, mock_smtp):
|
||
"""邮件发送异常"""
|
||
mock_smtp.side_effect = Exception("SMTP server error")
|
||
|
||
config = {"host": "smtp.example.com", "port": 465, "user": "user@example.com",
|
||
"password": "pass", "use_ssl": True, "from_addr": "user@example.com"}
|
||
result = send_mail(config, ["admin@example.com"], "测试标题", "测试内容")
|
||
assert result["success"] is False
|
||
assert "邮件发送失败" in result["message"]
|
||
|
||
|
||
class TestPushPendingAlerts:
|
||
"""push_pending_alerts 测试"""
|
||
|
||
def test_no_channels(self, db):
|
||
"""没有渠道 → 返回0"""
|
||
count = push_pending_alerts(db)
|
||
assert count == 0
|
||
|
||
def test_no_pending_alerts(self, db):
|
||
"""没有待处理预警 → 返回0"""
|
||
from app.models import NotificationChannel
|
||
db.add(NotificationChannel(name="测试群", channel_type="wecom", config={}, enabled=True))
|
||
db.commit()
|
||
|
||
count = push_pending_alerts(db)
|
||
assert count == 0
|
||
|
||
@patch("app.utils.notifier.push_alert")
|
||
def test_push_pending_with_data(self, mock_push, db):
|
||
"""有渠道+待处理预警 → 尝试推送"""
|
||
from app.models import NotificationChannel, KPIAlert, KPIDefinition
|
||
kpi = KPIDefinition(kpi_code="TEST_PUSH", kpi_name="推送测试", dimension="finance", status="active")
|
||
db.add(kpi)
|
||
db.commit()
|
||
|
||
db.add(NotificationChannel(name="测试群", channel_type="wecom",
|
||
config={"webhook_url": "http://test"}, enabled=True))
|
||
db.add(KPIAlert(kpi_id=kpi.id, alert_level="red", alert_message="测试预警", status="pending"))
|
||
db.commit()
|
||
|
||
mock_push.return_value = [{"channel": "wecom", "channel_name": "测试群", "success": True, "message": "ok"}]
|
||
|
||
count = push_pending_alerts(db)
|
||
mock_push.assert_called_once()
|
||
assert count == 1
|