张宗平
80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
"""detection 模块测试（mock TDengine execute 返回预设窗口）。"""
|
||
|
||
from unittest.mock import MagicMock
|
||
import pytest
|
||
from ts_anomaly_td.detection import run_anomaly_window, detect_all_algos, ALL_ALGOS
|
||
|
||
|
||
def make_mock_conn(rows=None):
|
||
"""构造 mock TDConnection，execute 返回预设行。"""
|
||
conn = MagicMock()
|
||
conn.execute.return_value = rows or []
|
||
return conn
|
||
|
||
|
||
def test_run_anomaly_window_returns_windows():
|
||
"""正常返回两个异常窗口。"""
|
||
conn = make_mock_conn([
|
||
(1000000, 2000000),
|
||
(3000000, 4000000),
|
||
])
|
||
windows = run_anomaly_window(conn, "finance_001", "ksigma")
|
||
assert windows == [(1000000, 2000000), (3000000, 4000000)]
|
||
|
||
|
||
def test_run_anomaly_window_empty():
|
||
"""无异常窗口时返回空列表。"""
|
||
conn = make_mock_conn([])
|
||
windows = run_anomaly_window(conn, "finance_001", "iqr")
|
||
assert windows == []
|
||
|
||
|
||
def test_run_anomaly_window_invalid_algo():
|
||
"""不支持的算法抛 ValueError。"""
|
||
conn = make_mock_conn()
|
||
with pytest.raises(ValueError, match="不支持的算法"):
|
||
run_anomaly_window(conn, "test", "nonexistent")
|
||
|
||
|
||
def test_run_anomaly_window_sql_error():
|
||
"""SQL 执行失败传播异常。"""
|
||
conn = make_mock_conn()
|
||
conn.execute.side_effect = Exception("TDengine error")
|
||
|
||
with pytest.raises(Exception, match="TDengine error"):
|
||
run_anomaly_window(conn, "test", "ksigma")
|
||
|
||
|
||
def test_detect_all_algos_independent_fault_tolerance():
|
||
"""单个算法失败不影响其他算法。"""
|
||
conn = MagicMock()
|
||
|
||
def side_effect(sql):
|
||
if "iqr" in sql:
|
||
raise Exception("iqr failed")
|
||
return [(1000, 2000)]
|
||
|
||
conn.execute.side_effect = side_effect
|
||
|
||
results = detect_all_algos(conn, "test")
|
||
|
||
# 所有 6 个算法都在结果中
|
||
assert set(results.keys()) == set(ALL_ALGOS)
|
||
|
||
# iqr 记录了 error
|
||
assert results["iqr"]["error"] == "iqr failed"
|
||
assert results["iqr"]["windows"] == []
|
||
|
||
# 其他算法正常
|
||
for algo in ALL_ALGOS:
|
||
if algo != "iqr":
|
||
assert results[algo]["error"] is None
|
||
assert len(results[algo]["windows"]) == 1
|
||
|
||
|
||
def test_detect_all_algos_custom_algo_list():
|
||
"""指定算法子集。"""
|
||
conn = make_mock_conn([(1000, 2000)])
|
||
results = detect_all_algos(conn, "test", algos=["ksigma", "iqr"])
|
||
assert set(results.keys()) == {"ksigma", "iqr"}
|