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

# -*- coding:utf-8 -*-
"""
bin/cdc-monitor.py 单测：复制槽判定 / YARN 存活 / 消息渲染 / main 串联。

不连真 PG（pgdb.connect + query 走替身）、不跑真 yarn（subprocess 走替身）、
不发真消息（Alerter 走替身）。
脚本路径含连字符，用 importlib.util 动态加载为模块。
"""
import importlib.util
import os
import re
import sys
from datetime import datetime
from unittest.mock import MagicMock

import pytest

PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
SCRIPT_PATH = os.path.join(PROJECT_ROOT, 'bin', 'cdc-monitor.py')

GB = 1024 ** 3
LAG_CRIT = 5 * GB


def _load_script():
    spec = importlib.util.spec_from_file_location('cdc_monitor', SCRIPT_PATH)
    mod = importlib.util.module_from_spec(spec)
    sys.modules['cdc_monitor'] = mod
    spec.loader.exec_module(mod)
    return mod


MON = _load_script()


def _stub_pg(monkeypatch, rows, conn=None):
    """替身 PG：connect 返回 conn，query 返回 rows。返回该 conn 供断言 close。"""
    conn = conn or MagicMock()
    monkeypatch.setattr(MON.pgdb, 'connect', lambda ds_ref: conn)
    monkeypatch.setattr(MON.pgdb, 'query', lambda c, sql: rows)
    return conn


# ---------- 复制槽 ----------

def test_cdc_slot_inactive_alerts(monkeypatch):
    _stub_pg(monkeypatch, [('cdc_large', False, 0)])
    assert MON.check_slots('postgresql/x', LAG_CRIT) == [
        ('槽 cdc_large', '空闲，采集作业可能已停')]


def test_non_cdc_slot_inactive_is_ignored(monkeypatch):
    """PG→PG 那条槽 apply 卡住时会反复重连，active 抖动不算告警。"""
    _stub_pg(monkeypatch, [('finance_pg2pg', False, 0)])
    assert MON.check_slots('postgresql/x', LAG_CRIT) == []


def test_slot_lag_over_threshold_alerts(monkeypatch):
    _stub_pg(monkeypatch, [('cdc_small', True, 6 * GB)])
    alerts = MON.check_slots('postgresql/x', LAG_CRIT)
    assert alerts == [('槽 cdc_small', '滞后 6.0 GB')]


def test_slot_lag_at_threshold_not_alerted(monkeypatch):
    """阈值是 >，正好等于不报。"""
    _stub_pg(monkeypatch, [('cdc_small', True, LAG_CRIT)])
    assert MON.check_slots('postgresql/x', LAG_CRIT) == []


def test_lag_applies_to_non_cdc_slot_too(monkeypatch):
    """active 只对 cdc_ 判，滞后对所有槽都判。"""
    _stub_pg(monkeypatch, [('finance_pg2pg', True, 9 * GB)])
    assert MON.check_slots('postgresql/x', LAG_CRIT) == [
        ('槽 finance_pg2pg', '滞后 9.0 GB')]


def test_null_lag_does_not_crash(monkeypatch):
    """confirmed_flush_lsn 为 NULL 时 lag 是 None，当 0 处理。"""
    _stub_pg(monkeypatch, [('cdc_large', True, None)])
    assert MON.check_slots('postgresql/x', LAG_CRIT) == []


def test_inactive_and_lagging_yields_two_alerts(monkeypatch):
    _stub_pg(monkeypatch, [('cdc_large', False, 7 * GB)])
    assert MON.check_slots('postgresql/x', LAG_CRIT) == [
        ('槽 cdc_large', '空闲，采集作业可能已停'),
        ('槽 cdc_large', '滞后 7.0 GB'),
    ]


def test_connection_closed_after_query(monkeypatch):
    conn = _stub_pg(monkeypatch, [])
    MON.check_slots('postgresql/x', LAG_CRIT)
    conn.close.assert_called_once()


def test_db_failure_becomes_alert(monkeypatch):
    def _boom(ds_ref):
        raise RuntimeError('数据源 ini 不存在')
    monkeypatch.setattr(MON.pgdb, 'connect', _boom)
    alerts = MON.check_slots('postgresql/x', LAG_CRIT)
    assert len(alerts) == 1
    assert alerts[0][0] == '主库'
    assert '数据源 ini 不存在' in alerts[0][1]


# ---------- YARN ----------

def test_all_jobs_running_no_alert(monkeypatch):
    monkeypatch.setattr(MON.subprocess, 'check_output',
                        lambda *a, **kw: b'app-1 st-cdc-large RUNNING\napp-2 st-cdc-small RUNNING')
    assert MON.check_yarn(['st-cdc-large', 'st-cdc-small']) == []


def test_missing_job_alerts(monkeypatch):
    monkeypatch.setattr(MON.subprocess, 'check_output',
                        lambda *a, **kw: b'app-1 st-cdc-large RUNNING')
    assert MON.check_yarn(['st-cdc-large', 'st-cdc-small']) == [
        ('作业 st-cdc-small', '不在 YARN RUNNING 列表')]


def test_yarn_failure_becomes_single_alert(monkeypatch):
    def _boom(*a, **kw):
        raise OSError('yarn: command not found')
    monkeypatch.setattr(MON.subprocess, 'check_output', _boom)
    alerts = MON.check_yarn(['st-cdc-large', 'st-cdc-small'])
    assert len(alerts) == 1          # 查不到就是一条，不按作业数量刷屏
    assert alerts[0][0] == 'YARN'
    assert 'command not found' in alerts[0][1]


# ---------- 渲染：走一圈全部五类告警 ----------

FIXED_NOW = datetime(2026, 8, 26, 15, 21, 27)


def test_render_single_alert():
    assert MON.render([('槽 cdc_large', '空闲，采集作业可能已停')], now=FIXED_NOW) == (
        '### <font color="warning">CDC 监控告警</font>\n'
        '> 检测时间：2026-08-26 15:21:27\n'
        '> 槽 cdc_large：<font color="warning">空闲，采集作业可能已停</font>')


def test_render_defaults_to_current_time(monkeypatch):
    """不传 now 时取当前时间，格式 yyyy-MM-dd HH:mm:ss。"""
    out = MON.render([('主库', '连不上')])
    assert re.match(r'^> 检测时间：\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$', out.split('\n')[1])


def test_render_all_alert_kinds():
    alerts = [
        ('槽 cdc_large', '空闲，采集作业可能已停'),
        ('槽 cdc_small', '滞后 6.0 GB'),
        ('作业 st-cdc-large', '不在 YARN RUNNING 列表'),
        ('主库', '复制槽查询失败：connection refused'),
        ('YARN', '查询失败：timeout'),
    ]
    out = MON.render(alerts, now=FIXED_NOW)
    lines = out.split('\n')
    assert lines[0] == '### <font color="warning">CDC 监控告警</font>'
    assert lines[1] == '> 检测时间：2026-08-26 15:21:27'
    assert len(lines) == 7
    for (key, desc), line in zip(alerts, lines[2:]):
        assert line == '> {0}：<font color="warning">{1}</font>'.format(key, desc)


# ---------- main 串联 ----------

def _run_main(monkeypatch, slot_alerts, yarn_alerts):
    """跑 main，返回替身 alerter 实例供断言。"""
    monkeypatch.setattr(sys, 'argv', ['cdc-monitor.py'])
    monkeypatch.setattr(MON, 'check_slots', lambda ds, lag: list(slot_alerts))
    monkeypatch.setattr(MON, 'check_yarn', lambda jobs: list(yarn_alerts))
    instance = MagicMock()
    monkeypatch.setattr(MON, 'Alerter', MagicMock(return_value=instance))
    MON.main()
    return instance


def test_main_sends_nothing_when_healthy(monkeypatch):
    instance = _run_main(monkeypatch, [], [])
    instance.send_markdown.assert_not_called()


def test_main_sends_once_with_all_alerts(monkeypatch):
    instance = _run_main(
        monkeypatch,
        [('槽 cdc_large', '空闲，采集作业可能已停')],
        [('作业 st-cdc-small', '不在 YARN RUNNING 列表')])
    instance.send_markdown.assert_called_once()
    content = instance.send_markdown.call_args[0][0]
    assert '槽 cdc_large' in content
    assert '作业 st-cdc-small' in content


def test_main_builds_alerter_before_checks(monkeypatch):
    """健康时也要构造 Alerter：配置坏了要立刻暴露，不能等真出事才发现发不出去。"""
    monkeypatch.setattr(sys, 'argv', ['cdc-monitor.py'])
    monkeypatch.setattr(MON, 'check_slots', lambda ds, lag: [])
    monkeypatch.setattr(MON, 'check_yarn', lambda jobs: [])
    fake_cls = MagicMock()
    monkeypatch.setattr(MON, 'Alerter', fake_cls)
    MON.main()
    fake_cls.assert_called_once_with(channel='default')


def test_main_passes_cli_overrides(monkeypatch):
    seen = {}
    monkeypatch.setattr(sys, 'argv', [
        'cdc-monitor.py', '-ds', 'postgresql/other', '-channel', 'realtime',
        '-jobs', 'a, b ,', '-lag-gb', '2'])
    monkeypatch.setattr(MON, 'check_slots',
                        lambda ds, lag: seen.update(ds=ds, lag=lag) or [])
    monkeypatch.setattr(MON, 'check_yarn', lambda jobs: seen.update(jobs=jobs) or [])
    fake_cls = MagicMock()
    monkeypatch.setattr(MON, 'Alerter', fake_cls)

    MON.main()

    assert seen['ds'] == 'postgresql/other'
    assert seen['lag'] == 2 * GB
    assert seen['jobs'] == ['a', 'b']      # 逗号分隔去空白、丢空项
    fake_cls.assert_called_once_with(channel='realtime')