mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 19:14:19 +08:00
GitHubandClaude Opus 4.8
4dfd18050e
- pyproject.toml: add mypy overrides for pandas/tabulate/matplotlib stubs, disable strict checking for vendored MyTT library - config.py: use cast() for dict[str, Any] .get() returns - beichi.py: widen _calc_bi_force param to BI | XD, import XD - backtest/cli.py: split combo/single strategy into separate typed variables - backtest/combo.py: add bool_array() helper for numpy return types - chanlun/analyser.py: type ignore for pandas row access, fix dict type arg - unified.py: change fields param from object to Any - ex/mac_client.py: add type args to list literals - cli/cmd_offline.py: wrap int market as Market enum before API call - cli/cmd_chanlun.py: fix dict type arg - offline/write_*.py: explicit int() cast for struct.unpack returns - MyTT.py: fix line-too-long comments, UP038 isinstance syntax - tests: fix E712 (==False → ~mask), E741 (noqa), F841, import sorting - ruff format applied across codebase Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
48 lines
1.3 KiB
Python
48 lines
1.3 KiB
Python
"""同步 transport 回归测试。"""
|
|
|
|
from unittest.mock import patch
|
|
|
|
from easy_tdx.exceptions import TdxConnectionError
|
|
from easy_tdx.transport.sync import TdxConnection
|
|
|
|
|
|
class _FakeSocket:
|
|
def __init__(self) -> None:
|
|
self.timeout: float | None = None
|
|
self.connected_to: tuple[str, int] | None = None
|
|
self.closed = False
|
|
|
|
def settimeout(self, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def connect(self, address: tuple[str, int]) -> None:
|
|
self.connected_to = address
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def test_sync_connection_closes_socket_when_setup_fails() -> None:
|
|
sock = _FakeSocket()
|
|
conn = TdxConnection("127.0.0.1", port=7709, timeout=0.2)
|
|
|
|
with (
|
|
patch("easy_tdx.transport.sync.socket.socket", return_value=sock),
|
|
patch.object(
|
|
TdxConnection,
|
|
"_send_setup",
|
|
side_effect=TdxConnectionError("setup failed"),
|
|
),
|
|
):
|
|
try:
|
|
conn.connect()
|
|
except TdxConnectionError as exc:
|
|
assert "setup failed" in str(exc)
|
|
else: # pragma: no cover - 防御性断言
|
|
raise AssertionError("expected setup failure")
|
|
|
|
assert sock.timeout == 0.2
|
|
assert sock.connected_to == ("127.0.0.1", 7709)
|
|
assert sock.closed is True
|
|
assert conn._sock is None
|