claude_dev
918bbed0fc
- 修复 deps.py 中 get_vn_service 的引用错误 (vn_service.vn_service -> vn_service) - 移除登录页面上的明文密码提示 - 改进前端错误处理，避免数据加载失败导致登录显示错误
148 lines
4.2 KiB
Python
148 lines
4.2 KiB
Python
"""
|
|
Phase 2 API 测试
|
|
测试核心 API 功能
|
|
"""
|
|
import asyncio
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from sanguo_web.api import app, vn_service
|
|
from sanguo_web.services.main_service import VeighNaService
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
"""测试客户端"""
|
|
async with AsyncClient(app=app, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest.fixture
|
|
async def service():
|
|
"""初始化 VeighNa 服务"""
|
|
service = VeighNaService()
|
|
await service.initialize()
|
|
yield service
|
|
await service.shutdown()
|
|
|
|
|
|
class TestGatewayAPI:
|
|
"""网关 API 测试"""
|
|
|
|
async def test_list_gateways(self, client):
|
|
"""测试获取网关列表"""
|
|
response = await client.get("/api/v1/gateway/list")
|
|
assert response.status_code == 401 # 未认证
|
|
|
|
async def test_get_gateway_setting(self, client):
|
|
"""测试获取网关配置"""
|
|
# 跳过认证进行测试
|
|
# 实际应使用认证 Token
|
|
pass
|
|
|
|
|
|
class TestMarketAPI:
|
|
"""行情 API 测试"""
|
|
|
|
async def test_get_contracts(self, client):
|
|
"""测试获取合约列表"""
|
|
# 跳过认证测试
|
|
pass
|
|
|
|
|
|
class TestTradingAPI:
|
|
"""交易 API 测试"""
|
|
|
|
async def test_get_accounts(self, client):
|
|
"""测试获取账户"""
|
|
# 跳过认证测试
|
|
pass
|
|
|
|
|
|
class TestVeighNaService:
|
|
"""VeighNa 服务测试"""
|
|
|
|
async def test_initialize(self):
|
|
"""测试服务初始化"""
|
|
service = VeighNaService()
|
|
assert not service.is_initialized
|
|
await service.initialize()
|
|
assert service.is_initialized
|
|
await service.shutdown()
|
|
|
|
async def test_get_available_gateways(self, service):
|
|
"""测试获取可用网关"""
|
|
gateways = await service.get_available_gateways()
|
|
assert isinstance(gateways, list)
|
|
# Mock 模式下应有默认网关
|
|
assert len(gateways) > 0
|
|
|
|
async def test_get_gateway_setting(self, service):
|
|
"""测试获取网关配置"""
|
|
setting = await service.get_gateway_setting("CTP")
|
|
assert isinstance(setting, dict)
|
|
|
|
async def test_get_contracts(self, service):
|
|
"""测试获取合约列表"""
|
|
contracts = await service.get_contracts()
|
|
assert isinstance(contracts, list)
|
|
|
|
async def test_get_accounts(self, service):
|
|
"""测试获取账户"""
|
|
accounts = await service.get_accounts()
|
|
assert isinstance(accounts, list)
|
|
# Mock 模式下应有默认账户
|
|
if accounts:
|
|
assert "account_id" in accounts[0]
|
|
|
|
async def test_get_positions(self, service):
|
|
"""测试获取持仓"""
|
|
positions = await service.get_positions()
|
|
assert isinstance(positions, list)
|
|
|
|
async def test_get_orders(self, service):
|
|
"""测试获取订单"""
|
|
orders = await service.get_orders()
|
|
assert isinstance(orders, list)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 简单的手动测试
|
|
async def main():
|
|
print("Testing VeighNa Service...")
|
|
|
|
service = VeighNaService()
|
|
await service.initialize()
|
|
|
|
print("\n1. Available Gateways:")
|
|
gateways = await service.get_available_gateways()
|
|
for gw in gateways:
|
|
print(f" - {gw['gateway_name']}: {gw.get('display_name', gw['gateway_type'])}")
|
|
|
|
print("\n2. Gateway Setting (CTP):")
|
|
setting = await service.get_gateway_setting("CTP")
|
|
for key, value in setting.items():
|
|
print(f" - {key}: {value if value else '(empty)'}")
|
|
|
|
print("\n3. Contracts:")
|
|
contracts = await service.get_contracts()
|
|
print(f" Total: {len(contracts)} contracts")
|
|
|
|
print("\n4. Accounts:")
|
|
accounts = await service.get_accounts()
|
|
for acc in accounts:
|
|
print(f" - {acc['account_id']}: balance={acc['balance']}, available={acc['available']}")
|
|
|
|
print("\n5. Positions:")
|
|
positions = await service.get_positions()
|
|
print(f" Total: {len(positions)} positions")
|
|
|
|
print("\n6. Orders:")
|
|
orders = await service.get_orders()
|
|
print(f" Total: {len(orders)} orders")
|
|
|
|
await service.shutdown()
|
|
print("\n✓ All tests passed!")
|
|
|
|
asyncio.run(main())
|