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

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

不连真 PG（pgdb.connect + query 走替身）、不打真 JobManager（urlopen 走替身）、
不发真消息（Alerter 走替身）。
脚本路径含连字符，用 importlib.util 动态加载为模块。
"""
import importlib.util
import json
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
JM = 'cdhmaster02:8081'


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 _stub_flink(monkeypatch, jobs):
    """替身 JobManager：/jobs/overview 返回给定作业列表。返回捕获的请求参数。"""
    captured = {}

    def _open(url, timeout=None):
        captured['url'] = url
        captured['timeout'] = timeout
        resp = MagicMock()
        resp.read.return_value = json.dumps({'jobs': jobs}).encode('utf-8')
        resp.__enter__.return_value = resp
        return resp

    monkeypatch.setattr(MON.urllib.request, 'urlopen', _open)
    return captured


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

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]


# ---------- Flink 作业存活 ----------

def test_all_jobs_running_no_alert(monkeypatch):
    captured = _stub_flink(monkeypatch, [
        {'name': 'st-cdc-large', 'state': 'RUNNING'},
        {'name': 'st-cdc-small', 'state': 'RUNNING'},
    ])
    assert MON.check_flink(JM, ['st-cdc-large', 'st-cdc-small']) == []
    assert captured['url'] == 'http://cdhmaster02:8081/jobs/overview'


def test_missing_job_alerts(monkeypatch):
    _stub_flink(monkeypatch, [{'name': 'st-cdc-large', 'state': 'RUNNING'}])
    assert MON.check_flink(JM, ['st-cdc-large', 'st-cdc-small']) == [
        ('作业 st-cdc-small', '不在 Flink 集群上')]


def test_restarting_job_alerts(monkeypatch):
    """2026-09-07 的故障形态：作业名一直在列表里，只判存在会漏报 7 小时。"""
    _stub_flink(monkeypatch, [
        {'name': 'st-cdc-large', 'state': 'RESTARTING'},
        {'name': 'st-cdc-small', 'state': 'RESTARTING'},
    ])
    assert MON.check_flink(JM, ['st-cdc-large', 'st-cdc-small']) == [
        ('作业 st-cdc-large', '状态 RESTARTING'),
        ('作业 st-cdc-small', '状态 RESTARTING'),
    ]


def test_canceled_job_alerts(monkeypatch):
    """取消的作业仍留在 /jobs/overview 里。"""
    _stub_flink(monkeypatch, [{'name': 'st-cdc-large', 'state': 'CANCELED'}])
    assert MON.check_flink(JM, ['st-cdc-large']) == [
        ('作业 st-cdc-large', '状态 CANCELED')]


def test_stale_record_does_not_mask_running(monkeypatch):
    """同名作业的历史记录不能盖掉在跑的那条，RUNNING 优先。"""
    _stub_flink(monkeypatch, [
        {'name': 'st-cdc-large', 'state': 'CANCELED'},
        {'name': 'st-cdc-large', 'state': 'RUNNING'},
    ])
    assert MON.check_flink(JM, ['st-cdc-large']) == []


def test_flink_failure_becomes_single_alert(monkeypatch):
    def _boom(url, timeout=None):
        raise OSError('Connection refused')
    monkeypatch.setattr(MON.urllib.request, 'urlopen', _boom)
    alerts = MON.check_flink(JM, ['st-cdc-large', 'st-cdc-small'])
    assert len(alerts) == 1          # 查不到就是一条，不按作业数量刷屏
    assert alerts[0][0] == 'Flink 集群'
    assert 'Connection refused' 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', '不在 Flink 集群上'),
        ('作业 st-cdc-small', '状态 RESTARTING'),
        ('主库', '复制槽查询失败：connection refused'),
        ('Flink 集群', '查询失败：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) == 8
    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, flink_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_flink', lambda jm, jobs: list(flink_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', '状态 RESTARTING')])
    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_flink', lambda jm, jobs: [])
    fake_cls = MagicMock()
    monkeypatch.setattr(MON, 'Alerter', fake_cls)
    MON.main()
    fake_cls.assert_called_once_with(channel='default')


def test_main_defaults_cover_both_jobs(monkeypatch):
    """默认值必须带上小表组，否则它挂了不报警。"""
    seen = {}
    monkeypatch.setattr(sys, 'argv', ['cdc-monitor.py'])
    monkeypatch.setattr(MON, 'check_slots', lambda ds, lag: [])
    monkeypatch.setattr(MON, 'check_flink',
                        lambda jm, jobs: seen.update(jm=jm, jobs=jobs) or [])
    monkeypatch.setattr(MON, 'Alerter', MagicMock())

    MON.main()

    assert seen['jm'] == 'cdhmaster02:8081'
    assert seen['jobs'] == ['st-cdc-large', 'st-cdc-small']


def test_main_passes_cli_overrides(monkeypatch):
    seen = {}
    monkeypatch.setattr(sys, 'argv', [
        'cdc-monitor.py', '-ds', 'postgresql/other', '-channel', 'realtime',
        '-jm', 'other-host:9081', '-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_flink',
                        lambda jm, jobs: seen.update(jm=jm, 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['jm'] == 'other-host:9081'
    assert seen['jobs'] == ['a', 'b']      # 逗号分隔去空白、丢空项
    fake_cls.assert_called_once_with(channel='realtime')