poyee-data-warehouse/tests/unit/alerter/test_alerter.py

# -*- coding:utf-8 -*-
"""dw_base.alerter.alerter 配置加载 + 企微推送单测（不发真消息，urlopen 走替身）。"""
import json
from unittest.mock import MagicMock

import pytest

from dw_base.alerter import alerter as alerter_mod
from dw_base.alerter.alerter import Alerter

BASE_URL = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='


def _write_conf(tmp_path, channels=None):
    """造一份临时 alerter.ini，返回路径。"""
    if channels is None:
        channels = {'default': 'real-key-0001'}
    lines = ['[wechat_work]', 'base_url = ' + BASE_URL, '', '[channels]']
    lines.extend('{0} = {1}'.format(name, key) for name, key in channels.items())
    path = tmp_path / 'alerter.ini'
    path.write_text('\n'.join(lines) + '\n', encoding='utf-8')
    return str(path)


def _stub_urlopen(captured, resp='{"errcode":0,"errmsg":"ok"}'):
    """替身 urlopen：把 Request 存进 captured，返回可 read 的响应对象。"""
    def _open(req, timeout=None):
        captured['req'] = req
        captured['timeout'] = timeout
        fake_resp = MagicMock()
        fake_resp.read.return_value = resp.encode('utf-8')
        return fake_resp
    return _open


# ---------- 构造期的四类失败 ----------

def test_conf_file_missing_raises(tmp_path):
    with pytest.raises(RuntimeError, match='告警配置文件不存在'):
        Alerter(conf_path=str(tmp_path / 'nope.ini'))


def test_channel_not_registered_raises(tmp_path):
    conf = _write_conf(tmp_path, {'default': 'real-key-0001'})
    with pytest.raises(RuntimeError, match='告警通道未注册'):
        Alerter(channel='realtime', conf_path=conf)


def test_placeholder_key_raises(tmp_path):
    conf = _write_conf(tmp_path, {'default': '<替换为企微机器人 key>'})
    with pytest.raises(RuntimeError, match='仍是占位符'):
        Alerter(conf_path=conf)


def test_webhook_composed_from_base_and_key(tmp_path):
    conf = _write_conf(tmp_path, {'default': 'abc-123'})
    assert Alerter(conf_path=conf).webhook == BASE_URL + 'abc-123'


def test_custom_channel_picks_its_own_key(tmp_path):
    conf = _write_conf(tmp_path, {'default': 'key-d', 'realtime': 'key-r'})
    assert Alerter(channel='realtime', conf_path=conf).webhook == BASE_URL + 'key-r'


def test_key_whitespace_stripped(tmp_path):
    path = tmp_path / 'alerter.ini'
    path.write_text(
        '[wechat_work]\nbase_url = {0}\n\n[channels]\ndefault =   abc-123  \n'.format(BASE_URL),
        encoding='utf-8')
    assert Alerter(conf_path=str(path)).webhook == BASE_URL + 'abc-123'


# ---------- 发送 ----------

def test_send_markdown_posts_expected_payload(tmp_path, monkeypatch):
    conf = _write_conf(tmp_path, {'default': 'abc-123'})
    captured = {}
    monkeypatch.setattr(alerter_mod.urllib.request, 'urlopen', _stub_urlopen(captured))

    Alerter(conf_path=conf).send_markdown('### 标题\n> 一行')

    req = captured['req']
    assert req.full_url == BASE_URL + 'abc-123'
    assert req.headers['Content-type'] == 'application/json'
    assert captured['timeout'] == alerter_mod._TIMEOUT_SEC
    body = json.loads(req.data.decode('utf-8'))
    assert body == {'msgtype': 'markdown', 'markdown': {'content': '### 标题\n> 一行'}}


def test_send_markdown_keeps_chinese_unescaped(tmp_path, monkeypatch):
    """ensure_ascii=False：中文按 UTF-8 原样进 body，不是 \\uXXXX。"""
    conf = _write_conf(tmp_path)
    captured = {}
    monkeypatch.setattr(alerter_mod.urllib.request, 'urlopen', _stub_urlopen(captured))

    Alerter(conf_path=conf).send_markdown('槽空闲')

    raw = captured['req'].data
    assert '槽空闲'.encode('utf-8') in raw
    assert b'\\u' not in raw


def test_send_markdown_raises_on_nonzero_errcode(tmp_path, monkeypatch):
    conf = _write_conf(tmp_path)
    monkeypatch.setattr(
        alerter_mod.urllib.request, 'urlopen',
        _stub_urlopen({}, resp='{"errcode":93000,"errmsg":"invalid webhook url"}'))

    with pytest.raises(RuntimeError, match='企微发送失败'):
        Alerter(conf_path=conf).send_markdown('x')


def test_send_markdown_ok_on_zero_errcode(tmp_path, monkeypatch):
    conf = _write_conf(tmp_path)
    monkeypatch.setattr(alerter_mod.urllib.request, 'urlopen', _stub_urlopen({}))
    Alerter(conf_path=conf).send_markdown('x')  # 不抛即通过