# -*- 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) == ( '### CDC 监控告警\n' '> 检测时间:2026-08-26 15:21:27\n' '> 槽 cdc_large:空闲,采集作业可能已停') 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] == '### CDC 监控告警' 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}:{1}'.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')