Files
easy-tdx/tests/unit/test_failover.py
T
Justin Gu 6b98715a14 fix(client): get_index_bars/get_security_bars 加空数据故障转移
指数/板块指数(880xxx 等)并非所有服务器都提供,延迟最低的不一定返回数据。
实测约 6/47 台可达主机对所有指数返回空 body 且不报错,from_best_host() 按纯
延迟选主会选中这种"快但空"的主机,导致 get_index_bars 返回空 DataFrame。

get_market_stat 早有 _find_host_returning_quotes 兜底,但 K 线查询缺失同类逻辑。
本提交补齐对称设计:

- 新增 _find_host_returning_bars(sync + async):照搬 _find_host_returning_quotes
  模式,复用 find_working_host_sync/_async(最多试 5 台、带持久化、异常吞掉、
  全失败回退原 host)
- get_security_bars / get_index_bars(sync + async 共 4 处)接入:空数据且
  auto_reconnect 开启时触发逐台实测,找首台返回数据的 host
- 新增 TestIndexBarsEmptyFailover(5 个测试):覆盖空数据切 host、全候选空、
  非空不触发、auto_reconnect=False 不触发、个股 K 线同链路

验证:23 个 failover 测试全绿,941 个单元测试无回归。
2026-07-10 19:37:31 +08:00

520 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""跨主机故障转移(failover)测试。
验证 8 个 client 的 ``_execute`` 在同主机重试耗尽(``_RETRY_DELAYS`` 走完仍
``TdxConnectionError``)后,会通过 ``select_best_host_sync`` / ``_async`` 重新
测速、切到延迟最低的另一台服务器再试一轮。同时覆盖:
- ``select_best_host_sync`` 的节流(30s 窗口内不重复测速)与"跳过当前 host"语义。
- ``auto_reconnect=False`` 时 failover 不触发。
- ``get_market_stat`` 空数据时触发 failover 再试。
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from easy_tdx._reconnect import (
_FAILOVER_PING_THROTTLE_SEC,
_WORKING_HOST_MAX_ATTEMPTS,
find_working_host_sync,
select_best_host_sync,
)
from easy_tdx.client import TdxClient
from easy_tdx.commands.security_count import GetSecurityCountCmd
from easy_tdx.exceptions import TdxConnectionError
from easy_tdx.models.bar import SecurityBar
from easy_tdx.models.enums import KlineCategory, Market
# --------------------------------------------------------------------------- #
# select_best_host_sync 单元逻辑
# --------------------------------------------------------------------------- #
class TestSelectBestHostSync:
def setup_method(self) -> None:
# 每个测试前重置节流时间戳,避免上一个测试的节流窗口泄漏
import easy_tdx._reconnect as r
r._last_failover_ts = 0.0
def test_returns_lowest_latency_host_excluding_current(self) -> None:
"""返回延迟最低且与 current_host 不同的主机。"""
ping_fn = MagicMock(
return_value=[("fast", 0.01), ("slow", 0.5)] # 已按延迟升序
)
save_fn = MagicMock()
result = select_best_host_sync(["fast", "slow", "cur"], ping_fn, save_fn, 7709, 1.0, "cur")
assert result == "fast"
ping_fn.assert_called_once_with(["fast", "slow", "cur"], 7709, 1.0)
save_fn.assert_called_once_with("fast")
def test_skips_current_host_even_if_it_is_fastest(self) -> None:
"""当前主机恰好延迟最低时,应跳过它取次优。"""
ping_fn = MagicMock(return_value=[("cur", 0.01), ("other", 0.02)])
save_fn = MagicMock()
result = select_best_host_sync(["cur", "other"], ping_fn, save_fn, 7709, 1.0, "cur")
assert result == "other"
save_fn.assert_called_once_with("other")
def test_returns_none_when_only_current_reachable(self) -> None:
"""只有当前主机可达时返回 None(不切换、不持久化)。"""
ping_fn = MagicMock(return_value=[("cur", 0.01)])
save_fn = MagicMock()
result = select_best_host_sync(["cur"], ping_fn, save_fn, 7709, 1.0, "cur")
assert result is None
save_fn.assert_not_called()
def test_returns_none_when_no_host_reachable(self) -> None:
"""所有候选都不可达时返回 None。"""
ping_fn = MagicMock(return_value=[])
save_fn = MagicMock()
result = select_best_host_sync(["a", "b"], ping_fn, save_fn, 7709, 1.0, "cur")
assert result is None
save_fn.assert_not_called()
def test_throttle_skips_ping_within_window(self) -> None:
"""节流窗口内(30s)第二次调用直接返回 None,不触发测速。"""
ping_fn = MagicMock(return_value=[("other", 0.01)])
save_fn = MagicMock()
# 第一次:正常测速,返回 other
first = select_best_host_sync(["cur", "other"], ping_fn, save_fn, 7709, 1.0, "cur")
assert first == "other"
assert ping_fn.call_count == 1
# 第二次(立即):应被节流,跳过测速
second = select_best_host_sync(["cur", "other"], ping_fn, save_fn, 7709, 1.0, "cur")
assert second is None
# 测速调用次数不应增加
assert ping_fn.call_count == 1
def test_throttle_window_is_configurable_constant(self) -> None:
"""节流窗口常量存在且为正(防回归:误改成 0 会关闭节流)。"""
assert _FAILOVER_PING_THROTTLE_SEC > 0
# --------------------------------------------------------------------------- #
# find_working_host_sync 单元逻辑(多 host 轮询直到验证通过)
# --------------------------------------------------------------------------- #
class TestFindWorkingHostSync:
def test_returns_first_host_passing_validation(self) -> None:
"""按延迟顺序逐台测试,返回第一台通过验证的 host。"""
ranked = [("fast", 0.01), ("mid", 0.05), ("slow", 0.5)]
# fast 验证失败,mid 通过
try_fn = MagicMock(side_effect=[False, True, True])
save_fn = MagicMock()
result = find_working_host_sync(ranked, try_fn, save_fn, "cur")
assert result == "mid"
save_fn.assert_called_once_with("mid")
# 只测到通过那台为止(slow 未被测试)
assert try_fn.call_count == 2
def test_skips_current_host(self) -> None:
"""跳过 current_host,不对其调用验证函数。"""
ranked = [("cur", 0.01), ("other", 0.02)]
try_fn = MagicMock(return_value=True)
save_fn = MagicMock()
result = find_working_host_sync(ranked, try_fn, save_fn, "cur")
assert result == "other"
# cur 被跳过,只验证了 other
try_fn.assert_called_once_with("other")
def test_returns_none_when_all_fail_validation(self) -> None:
"""所有候选验证都失败时返回 None。"""
ranked = [("a", 0.01), ("b", 0.02)]
try_fn = MagicMock(return_value=False)
save_fn = MagicMock()
result = find_working_host_sync(ranked, try_fn, save_fn, "cur")
assert result is None
save_fn.assert_not_called()
def test_respects_max_attempts(self) -> None:
"""max_attempts 限制最多测试的候选数。"""
ranked = [("a", 0.01), ("b", 0.02), ("c", 0.03)]
try_fn = MagicMock(return_value=False)
save_fn = MagicMock()
result = find_working_host_sync(ranked, try_fn, save_fn, "cur", max_attempts=2)
assert result is None
# 只测了前 2 台(受 max_attempts 限制),c 未测
assert try_fn.call_count == 2
def test_validation_exception_skips_host_not_aborts(self) -> None:
"""单台验证抛异常只跳过该台,继续尝试下一台。"""
ranked = [("boom", 0.01), ("good", 0.02)]
save_fn = MagicMock()
def _try(host: str) -> bool:
if host == "boom":
raise RuntimeError("connection refused")
return True
result = find_working_host_sync(ranked, _try, save_fn, "cur")
assert result == "good"
save_fn.assert_called_once_with("good")
def test_default_max_attempts_constant(self) -> None:
"""默认 max_attempts 常量存在且合理(防回归)。"""
assert _WORKING_HOST_MAX_ATTEMPTS == 5
# --------------------------------------------------------------------------- #
# TdxClient._execute 跨主机故障转移
# --------------------------------------------------------------------------- #
class TestTdxClientFailover:
def setup_method(self) -> None:
import easy_tdx._reconnect as r
r._last_failover_ts = 0.0
def test_failover_switches_host_after_retries_exhausted(self) -> None:
"""同主机 4 次重试全失败后,应跨主机切到新 host 并成功。"""
with (
patch("easy_tdx.client.TdxConnection") as mock_conn_cls,
patch("easy_tdx.client.time.sleep"),
patch("easy_tdx.client.select_best_host_sync", return_value="new-host") as mock_select,
):
mock_conn = MagicMock()
# 首次 + 4 次重试全失败,第 6 次(failover 后)成功
mock_conn.execute.side_effect = [
TdxConnectionError("down"), # 首次
TdxConnectionError("down"), # 重试1
TdxConnectionError("down"), # 重试2
TdxConnectionError("down"), # 重试3
TdxConnectionError("down"), # 重试4
1234, # failover 到新 host 后成功
]
mock_conn_cls.return_value = mock_conn
client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0)
result = client._execute(GetSecurityCountCmd(Market.SH))
assert result == 1234
# failover 被调用,且传入的 current_host 是坏主机
mock_select.assert_called_once()
args = mock_select.call_args
assert args.args[-1] == "bad-host" # current_host
# client 的 host 已切换到新主机
assert client._host == "new-host"
def test_failover_returns_none_keeps_host_and_raises(self) -> None:
"""failover 未找到更优 host(返回 None)时,保持原 host 并抛出。"""
with (
patch("easy_tdx.client.TdxConnection") as mock_conn_cls,
patch("easy_tdx.client.time.sleep"),
patch("easy_tdx.client.select_best_host_sync", return_value=None),
):
mock_conn = MagicMock()
mock_conn.execute.side_effect = TdxConnectionError("always down")
mock_conn_cls.return_value = mock_conn
client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0)
with pytest.raises(TdxConnectionError):
client._execute(GetSecurityCountCmd(Market.SH))
# host 未被切换
assert client._host == "bad-host"
def test_no_failover_when_auto_reconnect_disabled(self) -> None:
"""auto_reconnect=False 时首次失败立即抛出,不进入 failover。"""
with (
patch("easy_tdx.client.TdxConnection") as mock_conn_cls,
patch("easy_tdx.client.time.sleep"),
patch("easy_tdx.client.select_best_host_sync") as mock_select,
):
mock_conn = MagicMock()
mock_conn.execute.side_effect = TdxConnectionError("down")
mock_conn_cls.return_value = mock_conn
client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=False, heartbeat_interval=0)
with pytest.raises(TdxConnectionError):
client._execute(GetSecurityCountCmd(Market.SH))
# failover 完全未被调用
mock_select.assert_not_called()
# --------------------------------------------------------------------------- #
# MacClient 跨主机故障转移(v1.19.4 兼容性:不污染标准 best_host)
# --------------------------------------------------------------------------- #
class TestMacClientFailover:
"""锁定 v1.19.4 修复:MacClient 的 failover 必须用 save_best_mac_host
而非 save_best_host,否则会把 MAC 服务器写进标准 best_host 配置项造成污染。
该 bug 曾在将 failover 改动从旧分支 cherry-pick 到含 v1.19.4 修复的 main 时
复现(_execute 的 failover 沿用了旧的 save_best_host)。本测试防止再次倒退。
"""
def setup_method(self) -> None:
import easy_tdx._reconnect as r
r._last_failover_ts = 0.0
def test_failover_uses_save_best_mac_host_not_save_best_host(self) -> None:
"""MacClient failover 持久化时必须调 save_best_mac_host。"""
from easy_tdx.mac.client import MacClient
from easy_tdx.mac.commands.kline_offset import KlineOffsetCmd
with (
patch("easy_tdx.mac.client.TdxConnection") as mock_conn_cls,
patch("easy_tdx.mac.client.time.sleep"),
patch(
"easy_tdx.mac.client.select_best_host_sync", return_value="new-mac-host"
) as mock_select,
):
mock_conn = MagicMock()
mock_conn.execute.side_effect = [
TdxConnectionError("down"),
TdxConnectionError("down"),
TdxConnectionError("down"),
TdxConnectionError("down"),
TdxConnectionError("down"),
999, # failover 后成功
]
mock_conn_cls.return_value = mock_conn
client = MacClient("bad-mac-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0)
client._execute(KlineOffsetCmd(0, 1))
mock_select.assert_called_once()
# 第 3 个位置参数是 save_fn,必须是 save_best_mac_host(防 v1.19.4 回归)
from easy_tdx.config import save_best_mac_host
save_fn = mock_select.call_args.args[2]
assert save_fn is save_best_mac_host, (
"MacClient failover 必须用 save_best_mac_host"
"否则污染标准 best_hostv1.19.4 修复)"
)
# --------------------------------------------------------------------------- #
# get_market_stat 空数据故障转移
# --------------------------------------------------------------------------- #
class TestMarketStatEmptyFailover:
def setup_method(self) -> None:
import easy_tdx._reconnect as r
r._last_failover_ts = 0.0
def _make_quote(self) -> object:
"""构造一个字段合法的统计指数 quote,让 get_market_stat 计算路径走通。"""
from easy_tdx.models.quote import SecurityQuote
# 880005price=涨家数/10, open=跌家数/10, low=平/10, high=总数/10
return SecurityQuote(
market=Market.SH,
code="880005",
price=159.3, # → up=1593
pre_close=0.0,
open=379.0, # → down=3790
high=552.8, # → total=5528
low=13.5, # → neutral=135
vol=0.0,
cur_vol=0.0,
amount=2.58e12,
s_vol=0.0,
b_vol=0.0,
active1=0,
active2=0,
bid1=0.0,
bid_vol1=0.0,
bid2=0.0,
bid_vol2=0.0,
bid3=0.0,
bid_vol3=0.0,
bid4=0.0,
bid_vol4=0.0,
bid5=0.0,
bid_vol5=0.0,
ask1=0.0,
ask_vol1=0.0,
ask2=0.0,
ask_vol2=0.0,
ask3=0.0,
ask_vol3=0.0,
ask4=0.0,
ask_vol4=0.0,
ask5=0.0,
ask_vol5=0.0,
rise_speed=0.0,
limit_up=None,
limit_down=None,
)
def test_empty_quotes_finds_working_host_and_returns_data(self) -> None:
"""空 quotes 时按延迟顺序逐台实测,找到返回数据的 host。"""
quote = self._make_quote()
client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0)
# _execute: 首次空(bad-host)→ 验证 hostA 空 → 验证 hostB 非空 → 最终再取一次
with (
patch.object(client, "_execute", side_effect=[[], [], [quote], [quote]]) as mock_exec,
patch.object(client, "_reconnect") as mock_reconnect,
patch(
"easy_tdx.client.ping_all",
return_value=[("hostA", 0.01), ("hostB", 0.02)],
),
):
df = client.get_market_stat()
# _execute 调用序列:1 首次 + 2 次 find_working_host 验证(hostA空、hostB非空) + 1 最终取值
assert mock_exec.call_count == 4
# _reconnect 切换到 hostA、hostB(逐台实测),最终停在 hostB
reconnect_hosts = [c.args[0] for c in mock_reconnect.call_args_list]
assert reconnect_hosts == ["hostA", "hostB"]
assert len(df) == 1
def test_empty_quotes_all_candidates_empty_raises(self) -> None:
"""所有候选都返回空时,抛 RuntimeError。"""
client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0)
with (
patch.object(client, "_execute", return_value=[]),
patch.object(client, "_reconnect") as mock_reconnect,
patch(
"easy_tdx.client.ping_all",
return_value=[("hostA", 0.01), ("hostB", 0.02)],
),
):
with pytest.raises(RuntimeError, match="无法获取市场统计数据"):
client.get_market_stat()
# find_working_host 逐台实测了 hostA、hostB_reconnect 被各调一次)
reconnect_hosts = [c.args[0] for c in mock_reconnect.call_args_list]
assert reconnect_hosts == ["hostA", "hostB"]
# --------------------------------------------------------------------------- #
# get_index_bars / get_security_bars 空数据故障转移
# (与 TestMarketStatEmptyFailover 对称:指数/板块指数 880xxx 并非所有服务器都提供)
# --------------------------------------------------------------------------- #
class TestIndexBarsEmptyFailover:
"""K 线空数据故障转移——验证 get_index_bars/get_security_bars 空时逐台实测切 host。"""
def setup_method(self) -> None:
import easy_tdx._reconnect as r
r._last_failover_ts = 0.0
def _make_bar(self) -> SecurityBar:
"""构造一根字段合法的日 K,让 get_index_bars 下游处理走通。"""
return SecurityBar(
open=10.0,
close=10.5,
high=10.8,
low=9.9,
vol=1000.0,
amount=10500.0,
year=2026,
month=7,
day=10,
hour=15,
minute=0,
)
def test_empty_bars_finds_working_host_and_returns_data(self) -> None:
"""空 bars 时按延迟顺序逐台实测,找到返回数据的 host。"""
bar = self._make_bar()
client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0)
# _execute: 首次空(bad-host)→ 验证 hostA 空 → 验证 hostB 非空 → 最终再取一次
with (
patch.object(client, "_execute", side_effect=[[], [], [bar], [bar]]) as mock_exec,
patch.object(client, "_reconnect") as mock_reconnect,
patch(
"easy_tdx.client.ping_all",
return_value=[("hostA", 0.01), ("hostB", 0.02)],
),
):
df = client.get_index_bars(Market.SH, "880008", KlineCategory.DAY, 0, 10)
# _execute 调用序列:1 首次 + 2 次 find_working_host 验证(hostA空、hostB非空) + 1 最终取值
assert mock_exec.call_count == 4
# _reconnect 切换到 hostA、hostB(逐台实测),最终停在 hostB
reconnect_hosts = [c.args[0] for c in mock_reconnect.call_args_list]
assert reconnect_hosts == ["hostA", "hostB"]
assert len(df) == 1
def test_empty_bars_all_candidates_empty_returns_empty_df(self) -> None:
"""所有候选都返回空时,返回空 DataFrame(不抛异常)。"""
client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0)
with (
patch.object(client, "_execute", return_value=[]),
patch.object(client, "_reconnect") as mock_reconnect,
patch(
"easy_tdx.client.ping_all",
return_value=[("hostA", 0.01), ("hostB", 0.02)],
),
):
df = client.get_index_bars(Market.SH, "880008", KlineCategory.DAY, 0, 10)
# find_working_host 逐台实测了 hostA、hostB_reconnect 被各调一次)
reconnect_hosts = [c.args[0] for c in mock_reconnect.call_args_list]
assert reconnect_hosts == ["hostA", "hostB"]
assert df.empty
def test_non_empty_bars_does_not_trigger_failover(self) -> None:
"""首次即返回数据时,不触发空数据故障转移。"""
bar = self._make_bar()
client = TdxClient("good-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0)
with (
patch.object(client, "_execute", return_value=[bar]) as mock_exec,
patch.object(client, "_find_host_returning_bars") as mock_failover,
):
df = client.get_index_bars(Market.SH, "880008", KlineCategory.DAY, 0, 10)
assert mock_exec.call_count == 1
mock_failover.assert_not_called()
assert len(df) == 1
def test_failover_disabled_when_auto_reconnect_off(self) -> None:
"""auto_reconnect=False 时空数据不触发故障转移。"""
client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=False, heartbeat_interval=0)
with (
patch.object(client, "_execute", return_value=[]) as mock_exec,
patch.object(client, "_find_host_returning_bars") as mock_failover,
):
df = client.get_index_bars(Market.SH, "880008", KlineCategory.DAY, 0, 10)
assert mock_exec.call_count == 1
mock_failover.assert_not_called()
assert df.empty
def test_security_bars_also_triggers_failover(self) -> None:
"""get_security_bars(个股 K 线)同样接入空数据故障转移。"""
bar = self._make_bar()
client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0)
with (
patch.object(client, "_execute", side_effect=[[], [], [bar], [bar]]) as mock_exec,
patch.object(client, "_reconnect"),
patch(
"easy_tdx.client.ping_all",
return_value=[("hostA", 0.01), ("hostB", 0.02)],
),
):
df = client.get_security_bars(Market.SH, "600000", KlineCategory.DAY, 0, 10)
assert mock_exec.call_count == 4
assert len(df) == 1