张宗平
56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
"""tmq_watch 模块测试（mock TMQ consumer）。"""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
from ts_anomaly_td.tmq_watch import WatchRunner
|
|
|
|
|
|
def make_mock_conn():
|
|
conn = MagicMock()
|
|
conn.execute.return_value = [(1000, 1.0), (2000, 2.0)]
|
|
return conn
|
|
|
|
|
|
def test_watch_runner_polling_fallback(monkeypatch):
|
|
"""TMQ 不可用时自动降级为轮询。"""
|
|
conn = make_mock_conn()
|
|
runner = WatchRunner(conn, "test", window_sec=0.1)
|
|
|
|
# 强制 TMQ 不可用
|
|
runner._run_tmq = MagicMock(side_effect=ImportError("no TMQ"))
|
|
|
|
# 跑一个轮询周期就停
|
|
poll_count = [0]
|
|
|
|
def mock_poll():
|
|
poll_count[0] += 1
|
|
runner._mode = "polling"
|
|
runner.stop()
|
|
|
|
runner._run_polling = mock_poll
|
|
|
|
runner.run()
|
|
assert runner._mode == "polling"
|
|
assert poll_count[0] == 1
|
|
|
|
|
|
def test_watch_runner_stop_on_signal():
|
|
"""收到 stop 后退出循环。"""
|
|
conn = make_mock_conn()
|
|
runner = WatchRunner(conn, "test", window_sec=0.1)
|
|
runner.stop()
|
|
assert runner._stop is True
|
|
|
|
|
|
def test_watch_runner_window_check():
|
|
"""_check_window 调用 detection。"""
|
|
conn = make_mock_conn()
|
|
runner = WatchRunner(conn, "test")
|
|
|
|
with patch("ts_anomaly_td.detection.detect_all_algos") as mock_det:
|
|
mock_det.return_value = {
|
|
"ksigma": {"windows": [(1000, 2000)], "error": None},
|
|
}
|
|
runner._check_window([(1000, 1.0)])
|
|
mock_det.assert_called_once_with(conn, "test")
|