claude_dev
38f5635b59
- test_bridge_client: to_bridge_code转换(sh/sz/前缀) + HTTP mock(成功/失败返回None不抛) - test_shadow_orders: 幂等persistence + enabled跳过 + 正向记录+symbol转换 + 幂等不重复 + bridge失败不阻断 - NAS容器真实环境(Python3.10/pytest) 11 passed - 补 D-3 测试空缺(Sub Agent临时自测未沉淀成持久测试)
64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
"""D-3 bridge_client 单元测试（symbol 转换 + HTTP mock）。
|
||
|
||
不依赖 Windows/miniQMT——bridge_client 是纯 HTTP 客户端，mock urllib 即可。
|
||
覆盖：to_bridge_code 转换规则、place_order 成功/失败返回、失败不抛异常。
|
||
"""
|
||
import urllib.error
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from sanguo_trader.bridge_client import BridgeClient, to_bridge_code
|
||
|
||
|
||
class TestToBridgeCode:
|
||
"""sanguo symbol（纯数字码）→ bridge code（sh/sz 前缀）转换。
|
||
|
||
规则与 sanguo_data.datareader.guess_exchange 一致。
|
||
"""
|
||
|
||
def test_sh_codes(self):
|
||
"""沪市：60/68/51/56/58 开头。"""
|
||
assert to_bridge_code("600000") == "sh600000"
|
||
assert to_bridge_code("688981") == "sh688981"
|
||
assert to_bridge_code("510300") == "sh510300"
|
||
|
||
def test_sz_codes(self):
|
||
"""深市：00/30/15 开头。"""
|
||
assert to_bridge_code("000001") == "sz000001"
|
||
assert to_bridge_code("300750") == "sz300750"
|
||
assert to_bridge_code("159915") == "sz159915"
|
||
|
||
def test_already_prefixed(self):
|
||
"""已含 sh/sz 前缀 → 小写直返。"""
|
||
assert to_bridge_code("sh600000") == "sh600000"
|
||
assert to_bridge_code("SZ000001") == "sz000001"
|
||
|
||
|
||
class TestBridgeClientHttp:
|
||
"""BridgeClient HTTP 调用 mock（不真发请求）。"""
|
||
|
||
def test_place_order_ok(self):
|
||
"""成功返回 {ok, order_id}。"""
|
||
with patch("sanguo_trader.bridge_client.urllib.request.urlopen") as mu:
|
||
resp = MagicMock()
|
||
resp.read.return_value = b'{"ok": true, "order_id": 12345}'
|
||
mu.return_value.__enter__.return_value = resp
|
||
client = BridgeClient("http://b.test", "tok")
|
||
r = client.place_order("sh600000", "buy", 10.5, 100)
|
||
assert r == {"ok": True, "order_id": 12345}
|
||
mu.assert_called_once() # 确实发了请求
|
||
|
||
def test_place_order_network_failure_returns_none(self):
|
||
"""网络失败 → 返回 None，不抛异常（影子下单旁路原则）。"""
|
||
with patch("sanguo_trader.bridge_client.urllib.request.urlopen") as mu:
|
||
mu.side_effect = urllib.error.URLError("conn refused")
|
||
client = BridgeClient("http://b.test", "tok")
|
||
r = client.place_order("sh600000", "buy", 10.5, 100)
|
||
assert r is None
|
||
|
||
def test_get_account_failure_returns_none(self):
|
||
with patch("sanguo_trader.bridge_client.urllib.request.urlopen") as mu:
|
||
mu.side_effect = TimeoutError("slow")
|
||
client = BridgeClient("http://b.test", "tok")
|
||
r = client.get_account()
|
||
assert r is None
|