# -*- 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')