claude_dev
918bbed0fc
- 修复 deps.py 中 get_vn_service 的引用错误 (vn_service.vn_service -> vn_service) - 移除登录页面上的明文密码提示 - 改进前端错误处理，避免数据加载失败导致登录显示错误
173 lines
5.4 KiB
Python
173 lines
5.4 KiB
Python
"""
|
|
Phase 2 简单测试脚本
|
|
不依赖 pytest 的手动测试
|
|
"""
|
|
import asyncio
|
|
import sys
|
|
import os
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from sanguo_web.services.main_service import VeighNaService
|
|
|
|
|
|
async def test_service():
|
|
"""测试 VeighNa 服务"""
|
|
print("=" * 60)
|
|
print("Phase 2 API 功能测试")
|
|
print("=" * 60)
|
|
|
|
service = VeighNaService()
|
|
|
|
# 1. 测试初始化
|
|
print("\n[1/7] 测试服务初始化...")
|
|
await service.initialize()
|
|
assert service.is_initialized, "服务未初始化"
|
|
print(" ✓ 服务初始化成功")
|
|
|
|
# 2. 测试获取可用网关
|
|
print("\n[2/7] 测试获取可用网关...")
|
|
gateways = await service.get_available_gateways()
|
|
assert isinstance(gateways, list), "网关列表应为列表"
|
|
print(f" ✓ 找到 {len(gateways)} 个网关:")
|
|
for gw in gateways:
|
|
print(f" - {gw['gateway_name']}: {gw.get('display_name', gw['gateway_type'])}")
|
|
|
|
# 3. 测试获取网关配置
|
|
print("\n[3/7] 测试获取网关配置...")
|
|
for gw in gateways:
|
|
setting = await service.get_gateway_setting(gw['gateway_name'])
|
|
assert isinstance(setting, dict), f"网关 {gw['gateway_name']} 配置应为字典"
|
|
print(f" ✓ {gw['gateway_name']} 配置: {list(setting.keys())}")
|
|
|
|
# 4. 测试获取合约列表
|
|
print("\n[4/7] 测试获取合约列表...")
|
|
contracts = await service.get_contracts()
|
|
assert isinstance(contracts, list), "合约列表应为列表"
|
|
print(f" ✓ 找到 {len(contracts)} 个合约")
|
|
|
|
# 5. 测试获取账户
|
|
print("\n[5/7] 测试获取账户...")
|
|
accounts = await service.get_accounts()
|
|
assert isinstance(accounts, list), "账户列表应为列表"
|
|
print(f" ✓ 找到 {len(accounts)} 个账户:")
|
|
for acc in accounts:
|
|
print(f" - {acc['account_id']}: balance={acc['balance']}, available={acc['available']}")
|
|
|
|
# 6. 测试获取持仓
|
|
print("\n[6/7] 测试获取持仓...")
|
|
positions = await service.get_positions()
|
|
assert isinstance(positions, list), "持仓列表应为列表"
|
|
print(f" ✓ 找到 {len(positions)} 个持仓")
|
|
|
|
# 7. 测试获取订单
|
|
print("\n[7/7] 测试获取订单...")
|
|
orders = await service.get_orders()
|
|
assert isinstance(orders, list), "订单列表应为列表"
|
|
print(f" ✓ 找到 {len(orders)} 个订单")
|
|
|
|
# 测试获取活动订单
|
|
active_orders = await service.get_active_orders()
|
|
assert isinstance(active_orders, list), "活动订单列表应为列表"
|
|
print(f" ✓ 找到 {len(active_orders)} 个活动订单")
|
|
|
|
# 测试获取成交
|
|
trades = await service.get_trades()
|
|
assert isinstance(trades, list), "成交列表应为列表"
|
|
print(f" ✓ 找到 {len(trades)} 个成交")
|
|
|
|
# 清理
|
|
print("\n清理资源...")
|
|
await service.shutdown()
|
|
print(" ✓ 服务已关闭")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✓ 所有测试通过!")
|
|
print("=" * 60)
|
|
|
|
|
|
async def test_converter():
|
|
"""测试数据转换器"""
|
|
print("\n" + "=" * 60)
|
|
print("数据转换器测试")
|
|
print("=" * 60)
|
|
|
|
# 直接导入转换函数，避免导入整个 API 模块
|
|
# Direction mapping
|
|
DIRECTION_MAP = {
|
|
"LONG": "buy",
|
|
"SHORT": "sell",
|
|
"NET": "net",
|
|
}
|
|
ORDER_TYPE_MAP = {
|
|
"限价": "limit",
|
|
"LIMIT": "limit",
|
|
"市价": "market",
|
|
"MARKET": "market",
|
|
"STOP": "stop",
|
|
}
|
|
STATUS_MAP = {
|
|
"提交中": "submitting",
|
|
"SUBMITTING": "submitting",
|
|
"未成交": "not_traded",
|
|
"NOTTRADED": "not_traded",
|
|
"部分成交": "part_traded",
|
|
"PARTTRADED": "part_traded",
|
|
"全部成交": "all_traded",
|
|
"ALLTRADED": "all_traded",
|
|
"已撤销": "cancelled",
|
|
"CANCELLED": "cancelled",
|
|
"拒单": "rejected",
|
|
"REJECTED": "rejected",
|
|
}
|
|
|
|
print("\n[1/4] 测试方向转换...")
|
|
assert DIRECTION_MAP.get("LONG") == "buy"
|
|
assert DIRECTION_MAP.get("SHORT") == "sell"
|
|
print(" ✓ 方向转换正确")
|
|
|
|
print("\n[2/4] 测试订单类型转换...")
|
|
assert ORDER_TYPE_MAP.get("LIMIT") == "limit"
|
|
assert ORDER_TYPE_MAP.get("MARKET") == "market"
|
|
print(" ✓ 订单类型转换正确")
|
|
|
|
print("\n[3/4] 测试状态转换...")
|
|
assert STATUS_MAP.get("SUBMITTING") == "submitting"
|
|
assert STATUS_MAP.get("ALLTRADED") == "all_traded"
|
|
print(" ✓ 状态转换正确")
|
|
|
|
print("\n[4/4] 测试安全数值转换...")
|
|
def safe_float(value, default=0.0):
|
|
try:
|
|
return float(value) if value is not None else default
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
def safe_int(value, default=0):
|
|
try:
|
|
return int(value) if value is not None else default
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
assert safe_float("100.5") == 100.5
|
|
assert safe_float(None, 0) == 0
|
|
assert safe_int("100") == 100
|
|
assert safe_int(None, 0) == 0
|
|
print(" ✓ 数值转换正确")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✓ 转换器测试通过!")
|
|
print("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(test_service())
|
|
asyncio.run(test_converter())
|
|
except Exception as e:
|
|
print(f"\n✗ 测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|