mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 23:54:21 +08:00
GitHub
d0d65d64b8
服务器连不上或返回空数据时，自动测速切到延迟最低的可用服务器再试， Python API / CLI / Web API 三入口全部自动生效。 核心改动： - _reconnect.py 新增 select_best_host_sync/async（连接失败 failover） 和 find_working_host_sync/async（空数据逐台实测） - 8 个 client 的 _execute 注入跨主机故障转移（复用 auto_reconnect 开关） - get_market_stat 空数据时按延迟顺序逐台实测找返回数据的服务器 - 新增 _reconnect/_areconnect helper 收敛重建连接副本 - MacClient failover 用 save_best_mac_host（延续 v1.19.4 不污染 best_host） - 顺手修复 test_commands_offline 未使用 import（main CI failure 根因） 测试：925 passed（新增 18 个 failover 测试），ruff/mypy 零新增错误。
399 lines
16 KiB
Python
399 lines
16 KiB
Python
"""跨主机故障转移（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.enums import 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_host（v1.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
|
||
|
||
# 880005：price=涨家数/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"]
|