claude_dev
918bbed0fc
- 修复 deps.py 中 get_vn_service 的引用错误 (vn_service.vn_service -> vn_service) - 移除登录页面上的明文密码提示 - 改进前端错误处理，避免数据加载失败导致登录显示错误
438 lines
13 KiB
Python
438 lines
13 KiB
Python
"""
|
|
Sanguo VeighNa Web API 测试套件
|
|
测试所有 REST API 端点
|
|
"""
|
|
import pytest
|
|
import asyncio
|
|
from httpx import AsyncClient, ASGITransport
|
|
from fastapi import status
|
|
|
|
# 导入应用
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from sanguo_web.api import app
|
|
|
|
|
|
# ============================================
|
|
# Fixtures
|
|
# ============================================
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
"""创建测试客户端"""
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest.fixture
|
|
async def auth_token(client):
|
|
"""获取认证 Token"""
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "admin", "password": "admin123"}
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
return data.get("access_token")
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_headers(auth_token):
|
|
"""获取认证请求头"""
|
|
return {"Authorization": f"Bearer {auth_token}"}
|
|
|
|
|
|
# ============================================
|
|
# 系统端点测试
|
|
# ============================================
|
|
|
|
class TestSystemEndpoints:
|
|
"""系统端点测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_root(self, client):
|
|
"""测试根路径"""
|
|
response = await client.get("/")
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check(self, client):
|
|
"""测试健康检查"""
|
|
response = await client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "status" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_root(self, client):
|
|
"""测试 API 根路径"""
|
|
response = await client.get("/api")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "name" in data
|
|
assert "version" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_system_info(self, client, auth_headers):
|
|
"""测试系统信息"""
|
|
response = await client.get(
|
|
"/api/v1/system/info",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "version" in data or "status" in data
|
|
|
|
|
|
# ============================================
|
|
# 认证端点测试
|
|
# ============================================
|
|
|
|
class TestAuthEndpoints:
|
|
"""认证端点测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_success(self, client):
|
|
"""测试成功登录"""
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "admin", "password": "admin123"}
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
assert "token_type" in data
|
|
assert data["token_type"] == "bearer"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_wrong_password(self, client):
|
|
"""测试错误密码"""
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "admin", "password": "wrong_password"}
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_invalid_user(self, client):
|
|
"""测试无效用户"""
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "invalid_user", "password": "admin123"}
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_token_valid(self, client, auth_token):
|
|
"""测试有效 Token 验证"""
|
|
response = await client.post(
|
|
"/api/v1/auth/verify",
|
|
json={"token": auth_token}
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["valid"] is True
|
|
assert "user_info" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_token_invalid(self, client):
|
|
"""测试无效 Token 验证"""
|
|
response = await client.post(
|
|
"/api/v1/auth/verify",
|
|
json={"token": "invalid_token"}
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["valid"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_me(self, client, auth_headers):
|
|
"""测试获取当前用户信息"""
|
|
response = await client.get(
|
|
"/api/v1/auth/me",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "username" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logout(self, client, auth_headers):
|
|
"""测试登出"""
|
|
response = await client.post(
|
|
"/api/v1/auth/logout",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
# ============================================
|
|
# 网关端点测试
|
|
# ============================================
|
|
|
|
class TestGatewayEndpoints:
|
|
"""网关端点测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_available_gateways(self, client, auth_headers):
|
|
"""测试获取可用网关列表"""
|
|
response = await client.get(
|
|
"/api/v1/gateway/available",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_connected_gateways(self, client, auth_headers):
|
|
"""测试获取已连接网关列表"""
|
|
response = await client.get(
|
|
"/api/v1/gateway/connected",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_gateway_setting(self, client, auth_headers):
|
|
"""测试获取网关配置模板"""
|
|
response = await client.get(
|
|
"/api/v1/gateway/setting/CTP",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, dict)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_gateway(self, client, auth_headers):
|
|
"""测试连接网关（模拟）"""
|
|
response = await client.post(
|
|
"/api/v1/gateway/connect",
|
|
headers=auth_headers,
|
|
json={
|
|
"gateway_name": "CTP_TEST",
|
|
"gateway_type": "ctp",
|
|
"setting": {
|
|
"用户名": "test_user",
|
|
"密码": "test_pass",
|
|
"经纪商代码": "9999",
|
|
"交易服务器": "tcp://test服务器:41205",
|
|
"行情服务器": "tcp://test服务器:41213",
|
|
}
|
|
}
|
|
)
|
|
# 在 Mock 模式下可能返回 200 或 500
|
|
assert response.status_code in [200, 500]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unauthorized_access(self, client):
|
|
"""测试未授权访问"""
|
|
response = await client.get("/api/v1/gateway/available")
|
|
assert response.status_code == 401
|
|
|
|
|
|
# ============================================
|
|
# 行情端点测试
|
|
# ============================================
|
|
|
|
class TestMarketEndpoints:
|
|
"""行情端点测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_ticks(self, client, auth_headers):
|
|
"""测试获取行情数据"""
|
|
response = await client.get(
|
|
"/api/v1/market/ticks",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "ticks" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subscribe(self, client, auth_headers):
|
|
"""测试订阅行情"""
|
|
response = await client.post(
|
|
"/api/v1/market/subscribe",
|
|
headers=auth_headers,
|
|
json={
|
|
"symbol": "IF2501",
|
|
"exchange": "CFFEX"
|
|
}
|
|
)
|
|
assert response.status_code in [200, 202]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unsubscribe(self, client, auth_headers):
|
|
"""测试取消订阅"""
|
|
response = await client.post(
|
|
"/api/v1/market/unsubscribe",
|
|
headers=auth_headers,
|
|
json={
|
|
"symbol": "IF2501",
|
|
"exchange": "CFFEX"
|
|
}
|
|
)
|
|
assert response.status_code in [200, 202]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_contracts(self, client, auth_headers):
|
|
"""测试获取合约列表"""
|
|
response = await client.get(
|
|
"/api/v1/market/contracts",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "contracts" in data
|
|
|
|
|
|
# ============================================
|
|
# 交易端点测试
|
|
# ============================================
|
|
|
|
class TestTradingEndpoints:
|
|
"""交易端点测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_accounts(self, client, auth_headers):
|
|
"""测试获取账户信息"""
|
|
response = await client.get(
|
|
"/api/v1/trading/accounts",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_positions(self, client, auth_headers):
|
|
"""测试获取持仓信息"""
|
|
response = await client.get(
|
|
"/api/v1/trading/positions",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_orders(self, client, auth_headers):
|
|
"""测试获取委托列表"""
|
|
response = await client.get(
|
|
"/api/v1/trading/orders",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_active_orders(self, client, auth_headers):
|
|
"""测试获取活动委托"""
|
|
response = await client.get(
|
|
"/api/v1/trading/orders/active",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_order(self, client, auth_headers):
|
|
"""测试发送订单（模拟）"""
|
|
response = await client.post(
|
|
"/api/v1/trading/orders",
|
|
headers=auth_headers,
|
|
json={
|
|
"symbol": "IF2501",
|
|
"exchange": "CFFEX",
|
|
"direction": "buy",
|
|
"order_type": "limit",
|
|
"volume": 1,
|
|
"price": 3500.0
|
|
}
|
|
)
|
|
# 在没有连接网关的情况下可能返回错误
|
|
assert response.status_code in [200, 500]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_trades(self, client, auth_headers):
|
|
"""测试获取成交记录"""
|
|
response = await client.get(
|
|
"/api/v1/trading/trades",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "trades" in data
|
|
assert "total" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_account_summary(self, client, auth_headers):
|
|
"""测试获取账户综合信息"""
|
|
response = await client.get(
|
|
"/api/v1/trading/account",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code in [200, 404] # 可能没有账户数据
|
|
|
|
|
|
# ============================================
|
|
# 策略端点测试
|
|
# ============================================
|
|
|
|
class TestStrategyEndpoints:
|
|
"""策略端点测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_strategies(self, client, auth_headers):
|
|
"""测试获取策略列表"""
|
|
response = await client.get(
|
|
"/api/v1/strategy/list",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "strategies" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_strategy(self, client, auth_headers):
|
|
"""测试初始化策略"""
|
|
response = await client.post(
|
|
"/api/v1/strategy/test_strategy/init",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code in [200, 404]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_strategy(self, client, auth_headers):
|
|
"""测试启动策略"""
|
|
response = await client.post(
|
|
"/api/v1/strategy/test_strategy/start",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code in [200, 404]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_strategy(self, client, auth_headers):
|
|
"""测试停止策略"""
|
|
response = await client.post(
|
|
"/api/v1/strategy/test_strategy/stop",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code in [200, 404]
|
|
|
|
|
|
# ============================================
|
|
# 运行测试
|
|
# ============================================
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v", "--tb=short"])
|