张宗平
100 lines
2.9 KiB
Python
100 lines
2.9 KiB
Python
"""connector 模块测试（mock taosws.connect）。"""
|
|
|
|
from unittest.mock import patch, MagicMock
|
|
import pytest
|
|
from ts_anomaly_td.connector import TDConnection, ConnectionError, _MAX_RETRY_DURATION
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_connect():
|
|
"""mock taosws.connect 返回假连接。"""
|
|
with patch("ts_anomaly_td.connector.taosws.connect") as m:
|
|
mock_conn = MagicMock()
|
|
mock_conn.execute.return_value = [[("3.3.6.0",)]]
|
|
m.return_value = mock_conn
|
|
yield m
|
|
|
|
|
|
def test_connect_success(mock_connect):
|
|
"""首次连接成功。"""
|
|
conn = TDConnection()
|
|
assert conn.is_healthy()
|
|
mock_connect.assert_called_once()
|
|
|
|
|
|
def test_execute_query(mock_connect):
|
|
"""execute 返回行列表。"""
|
|
mock_conn = mock_connect.return_value
|
|
# taosws 返回可迭代的行对象；每行 row 是 (1, 'a') / (2, 'b')
|
|
mock_conn.execute.return_value = [(1, "a"), (2, "b")]
|
|
|
|
conn = TDConnection()
|
|
rows = conn.execute("SELECT 1")
|
|
assert rows == [(1, "a"), (2, "b")]
|
|
|
|
|
|
def test_execute_no_result(mock_connect):
|
|
"""execute_no_result 不返回值。"""
|
|
conn = TDConnection()
|
|
result = conn.execute_no_result("CREATE DATABASE test")
|
|
assert result is None
|
|
|
|
|
|
def test_retry_logic(mock_connect):
|
|
"""前 2 次失败、第 3 次成功。"""
|
|
import ts_anomaly_td.connector as mod
|
|
|
|
# 缩短退避以加速测试
|
|
mod._MAX_BACKOFF = 0.01
|
|
try:
|
|
mock_connect.side_effect = [
|
|
Exception("connection refused"),
|
|
Exception("connection refused"),
|
|
MagicMock(execute=MagicMock(return_value=[[("3.3.6.0",)]])),
|
|
]
|
|
|
|
conn = TDConnection()
|
|
assert conn.is_healthy()
|
|
assert mock_connect.call_count == 3
|
|
finally:
|
|
mod._MAX_BACKOFF = 300
|
|
|
|
|
|
def test_retry_timeout(mock_connect):
|
|
"""持续失败超过 30 分钟应抛出 ConnectionError。"""
|
|
import ts_anomaly_td.connector as mod
|
|
|
|
mod._MAX_RETRY_DURATION = 0.1
|
|
mod._MAX_BACKOFF = 0.01
|
|
try:
|
|
mock_connect.side_effect = Exception("always fails")
|
|
|
|
with pytest.raises(ConnectionError, match="30"):
|
|
conn = TDConnection()
|
|
# 触发懒加载连接：访问 .conn 属性（非 is_healthy，后者会吞掉异常）
|
|
_ = conn.conn
|
|
finally:
|
|
mod._MAX_RETRY_DURATION = 1800
|
|
mod._MAX_BACKOFF = 300
|
|
|
|
|
|
def test_close(mock_connect):
|
|
"""close 关闭底层连接。"""
|
|
conn = TDConnection()
|
|
# 触发懒加载连接
|
|
conn.is_healthy()
|
|
conn.close()
|
|
mock_connect.return_value.close.assert_called_once()
|
|
|
|
|
|
def test_batch_insert(mock_connect):
|
|
"""batch_insert 拼接 INSERT SQL。"""
|
|
conn = TDConnection()
|
|
rows = [(1000, 1.0, 0), (2000, 2.0, 1)]
|
|
conn.batch_insert("test", rows)
|
|
|
|
mock_conn = mock_connect.return_value
|
|
# 验证 execute 被调用
|
|
calls = [str(c[0][0]) for c in mock_conn.execute.call_args_list]
|
|
assert any("INSERT INTO s_test VALUES" in c for c in calls)
|