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 _slot_alerts(monkeypatch, rows):
    """只取判定结果，facts 另有用例覆盖。"""
    _stub_pg(monkeypatch, rows)
    return MON.check_slots('postgresql/x', LAG_CRIT)[1]


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):
    assert _slot_alerts(monkeypatch, [('cdc_large', False, 0)]) == [
        ('槽 cdc_large', '空闲，采集作业可能已停')]


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


def test_slot_lag_over_threshold_alerts(monkeypatch):
    assert _slot_alerts(monkeypatch, [('cdc_small', True, 6 * GB)]) == [
        ('槽 cdc_small', '滞后 6.0 GB')]


def test_slot_lag_at_threshold_not_alerted(monkeypatch):
    """阈值是 >，正好等于不报。"""
    assert _slot_alerts(monkeypatch, [('cdc_small', True, LAG_CRIT)]) == []


def test_lag_applies_to_non_cdc_slot_too(monkeypatch):
    """active 只对 cdc_ 判，滞后对所有槽都判。"""
    assert _slot_alerts(monkeypatch, [('finance_pg2pg', True, 9 * GB)]) == [
        ('槽 finance_pg2pg', '滞后 9.0 GB')]


def test_null_lag_does_not_crash(monkeypatch):
    """confirmed_flush_lsn 为 NULL 时 lag 是 None，当 0 处理。"""
    assert _slot_alerts(monkeypatch, [('cdc_large', True, None)]) == []


def test_inactive_and_lagging_yields_two_alerts(monkeypatch):
    assert _slot_alerts(monkeypatch, [('cdc_large', False, 7 * GB)]) == [
        ('槽 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)
    facts, alerts = MON.check_slots('postgresql/x', LAG_CRIT)
    assert facts == []
    assert len(alerts) == 1
    assert alerts[0][0] == '主库'
    assert '数据源 ini 不存在' in alerts[0][1]


# ---------- 复制槽：事实 ----------

def test_slot_facts_cover_every_slot(monkeypatch):
    """健康的槽也要出现在 facts 里，-report 才有东西可列。"""
    _stub_pg(monkeypatch, [('cdc_large', True, 0),
                           ('finance_pg2pg', True, 3 * GB)])
    facts, alerts = MON.check_slots('postgresql/x', LAG_CRIT)
    assert facts == [('cdc_large', True, 0), ('finance_pg2pg', True, 3 * GB)]
    assert alerts == []


def test_slot_facts_normalize_null_lag(monkeypatch):
    _stub_pg(monkeypatch, [('cdc_large', True, None)])
    assert MON.check_slots('postgresql/x', LAG_CRIT)[0] == [('cdc_large', True, 0)]


# ---------- 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'},
    ])
    facts, alerts = MON.check_flink(JM, ['st-cdc-large', 'st-cdc-small'])
    assert alerts == []
    assert facts == [('st-cdc-large', 'RUNNING'), ('st-cdc-small', 'RUNNING')]
    assert captured['url'] == 'http://cdhmaster02:8081/jobs/overview'


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


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'])[1] == [
        ('作业 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'])[1] == [
        ('作业 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']) == ([('st-cdc-large', 'RUNNING')], [])


def test_untracked_job_not_reported(monkeypatch):
    """集群上的其他作业不进 facts——这是 CDC 链路状态，不是集群状态。"""
    _stub_flink(monkeypatch, [
        {'name': 'st-finaltest', 'state': 'CANCELED'},
        {'name': 'st-cdc-large', 'state': 'RUNNING'},
    ])
    assert MON.check_flink(JM, ['st-cdc-large']) == ([('st-cdc-large', 'RUNNING')], [])


def test_flink_failure_becomes_single_alert(monkeypatch):
    def _boom(url, timeout=None):
        raise OSError('Connection refused')
    monkeypatch.setattr(MON.urllib.request, 'urlopen', _boom)
    facts, alerts = MON.check_flink(JM, ['st-cdc-large', 'st-cdc-small'])
    assert facts == []
    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)


# ---------- 状态报告渲染 ----------

HEALTHY_SLOTS = [('cdc_large', True, 0), ('cdc_small', True, 0),
                 ('finance_pg2pg', True, 107374182)]
HEALTHY_JOBS = [('st-cdc-large', 'RUNNING'), ('st-cdc-small', 'RUNNING')]


def test_status_all_healthy():
    assert MON.render_status(HEALTHY_SLOTS, HEALTHY_JOBS, [], now=FIXED_NOW) == (
        '### <font color="info">CDC 链路状态</font>\n'
        '> 检测时间：2026-08-26 15:21:27\n'
        '> 槽 cdc_large：<font color="info">活跃，滞后 0.0 GB</font>\n'
        '> 槽 cdc_small：<font color="info">活跃，滞后 0.0 GB</font>\n'
        '> 槽 finance_pg2pg：<font color="info">活跃，滞后 0.1 GB</font>\n'
        '> 作业 st-cdc-large：<font color="info">RUNNING</font>\n'
        '> 作业 st-cdc-small：<font color="info">RUNNING</font>')


def test_status_marks_only_bad_items():
    """异常项转 warning，其余保持 info，标题跟着转 warning。"""
    slots = [('cdc_large', False, 0), ('cdc_small', True, 0)]
    jobs = [('st-cdc-large', 'RESTARTING'), ('st-cdc-small', 'RUNNING')]
    alerts = [('槽 cdc_large', '空闲，采集作业可能已停'),
              ('作业 st-cdc-large', '状态 RESTARTING')]
    lines = MON.render_status(slots, jobs, alerts, now=FIXED_NOW).split('\n')
    assert lines[0] == '### <font color="warning">CDC 链路状态</font>'
    assert lines[2] == '> 槽 cdc_large：<font color="warning">空闲，滞后 0.0 GB</font>'
    assert lines[3] == '> 槽 cdc_small：<font color="info">活跃，滞后 0.0 GB</font>'
    assert lines[4] == '> 作业 st-cdc-large：<font color="warning">RESTARTING</font>'
    assert lines[5] == '> 作业 st-cdc-small：<font color="info">RUNNING</font>'


def test_status_shows_missing_job():
    lines = MON.render_status([], [('st-cdc-small', None)],
                              [('作业 st-cdc-small', '不在 Flink 集群上')],
                              now=FIXED_NOW).split('\n')
    assert lines[2] == '> 作业 st-cdc-small：<font color="warning">不在集群上</font>'


def test_status_falls_back_to_failure_reason():
    """查询失败时没有事实可列，报告不能是空的。"""
    alerts = [('主库', '复制槽查询失败：connection refused'),
              ('Flink 集群', '查询失败：timeout')]
    lines = MON.render_status([], [], alerts, now=FIXED_NOW).split('\n')
    assert lines[0] == '### <font color="warning">CDC 链路状态</font>'
    assert lines[2] == '> 主库：<font color="warning">复制槽查询失败：connection refused</font>'
    assert lines[3] == '> Flink 集群：<font color="warning">查询失败：timeout</font>'
    assert len(lines) == 4


def test_status_does_not_repeat_items_already_listed():
    """槽/作业的告警已经体现在它自己那行上，末尾不再重复一遍。"""
    slots = [('cdc_large', False, 0)]
    alerts = [('槽 cdc_large', '空闲，采集作业可能已停')]
    assert len(MON.render_status(slots, [], alerts, now=FIXED_NOW).split('\n')) == 3


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

def _run_main(monkeypatch, slot_alerts, flink_alerts, argv=None,
              slot_facts=(), flink_facts=()):
    """跑 main，返回替身 alerter 实例供断言。"""
    monkeypatch.setattr(sys, 'argv', argv or ['cdc-monitor.py'])
    monkeypatch.setattr(MON, 'check_slots',
                        lambda ds, lag: (list(slot_facts), list(slot_alerts)))
    monkeypatch.setattr(MON, 'check_flink',
                        lambda jm, jobs: (list(flink_facts), 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_report_sends_even_when_healthy(monkeypatch):
    """-report 是上线核对用的，正常也要发。"""
    instance = _run_main(monkeypatch, [], [],
                         argv=['cdc-monitor.py', '-report'],
                         slot_facts=HEALTHY_SLOTS, flink_facts=HEALTHY_JOBS)
    instance.send_markdown.assert_called_once()
    content = instance.send_markdown.call_args[0][0]
    assert content.startswith('### <font color="info">CDC 链路状态</font>')
    assert '> 作业 st-cdc-large：<font color="info">RUNNING</font>' in content


def test_main_report_shows_abnormal_state(monkeypatch):
    """-report 遇到异常发的仍是状态报告，不是告警消息。"""
    instance = _run_main(monkeypatch, [], [('作业 st-cdc-large', '状态 RESTARTING')],
                         argv=['cdc-monitor.py', '-report'],
                         flink_facts=[('st-cdc-large', 'RESTARTING')])
    content = instance.send_markdown.call_args[0][0]
    assert content.startswith('### <font color="warning">CDC 链路状态</font>')
    assert '> 作业 st-cdc-large：<font color="warning">RESTARTING</font>' 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')