test: 前后端对齐—容器复跑验证+清理僵尸测试
- test_main: FastAPI 0.139 _IncludedRouter 不再 flatten，改用 TestClient 探测路由 - datareader: 文件名(sh600000_daily)/patch target(vnpy.trader.database) 对齐 lazy import 实现 - alpha_lab/analyzer/data_adapter: vnpy.alpha/alphalens 容器专用本地 skip - 删 4 个测废弃 sanguo_web 的僵尸测试（-1267 行死代码） - pytest.ini: asyncio_mode=auto - frontend: package.json 加 test script（npm test 可跑） - NAS 容器 309 passed 全绿验证（Python 3.10，本机 303+6skip）
This commit is contained in:
@@ -6,6 +6,7 @@
|
||||
"scripts": {
|
||||
"dev": "vite",
|
||||
"build": "vue-tsc -b && vite build",
|
||||
"test": "vitest run",
|
||||
"preview": "vite preview"
|
||||
},
|
||||
"dependencies": {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
[pytest]
|
||||
pythonpath = .
|
||||
testpaths = tests
|
||||
asyncio_mode = auto
|
||||
|
||||
@@ -1,199 +0,0 @@
|
||||
# WebSocket 测试指南
|
||||
|
||||
## 概述
|
||||
|
||||
Phase 3 实现了 WebSocket 实时数据推送功能，包括：
|
||||
- WebSocket 连接管理
|
||||
- 事件监听器（Tick, Order, Trade, Position, Account, Log, Contract）
|
||||
- 订阅管理
|
||||
- 心跳机制
|
||||
|
||||
## 启动服务器
|
||||
|
||||
```bash
|
||||
cd /Users/chufeng/.openclaw/sanguo_projects/sanguo_vnpy_v2
|
||||
|
||||
# 启动 FastAPI 服务器
|
||||
uvicorn sanguo_web.api:app --host 0.0.0.0 --port 8000 --reload
|
||||
```
|
||||
|
||||
## 测试方法
|
||||
|
||||
### 方法 1: 使用 HTML 测试页面
|
||||
|
||||
1. 在浏览器中打开：
|
||||
```
|
||||
file:///Users/chufeng/.openclaw/sanguo_projects/sanguo_vnpy_v2/tests/websocket_test.html
|
||||
```
|
||||
|
||||
2. 点击"连接"按钮建立 WebSocket 连接
|
||||
|
||||
3. 选择要订阅的数据类型：
|
||||
- 行情 (Tick)
|
||||
- 订单 (Order)
|
||||
- 成交 (Trade)
|
||||
- 持仓 (Position)
|
||||
- 账户 (Account)
|
||||
- 日志 (Log)
|
||||
- 合约 (Contract)
|
||||
|
||||
4. 可选：输入品种代码（如：IF2501.CFFEX,IH2501.CFFEX）并订阅
|
||||
|
||||
5. 查看实时消息日志
|
||||
|
||||
### 方法 2: 使用 Python 测试脚本
|
||||
|
||||
```bash
|
||||
# 安装依赖
|
||||
pip install websockets
|
||||
|
||||
# 运行测试
|
||||
cd /Users/chufeng/.openclaw/sanguo_projects/sanguo_vnpy_v2
|
||||
python tests/test_websocket.py
|
||||
```
|
||||
|
||||
测试脚本会执行以下测试：
|
||||
1. 基本连接测试
|
||||
2. 订阅功能测试
|
||||
3. 消息接收测试
|
||||
4. 心跳机制测试
|
||||
5. 错误处理测试
|
||||
6. 认证连接测试
|
||||
|
||||
### 方法 3: 使用 wscat 命令行工具
|
||||
|
||||
```bash
|
||||
# 安装 wscat
|
||||
npm install -g wscat
|
||||
|
||||
# 连接 WebSocket
|
||||
wscat -c ws://localhost:8000/ws
|
||||
|
||||
# 发送订阅消息
|
||||
{"type":"subscribe","data":{"subscription":["tick","order","trade"]}}
|
||||
|
||||
# 发送心跳
|
||||
{"type":"ping","data":{}}
|
||||
|
||||
# 订阅品种
|
||||
{"type":"subscribe_symbol","data":{"symbol":["IF2501.CFFEX"]}}
|
||||
```
|
||||
|
||||
### 方法 4: 使用 JavaScript 控制台
|
||||
|
||||
在任何网页中打开浏览器控制台，运行：
|
||||
|
||||
```javascript
|
||||
// 创建 WebSocket 连接
|
||||
const ws = new WebSocket('ws://localhost:8000/ws');
|
||||
|
||||
// 监听连接事件
|
||||
ws.onopen = () => {
|
||||
console.log('Connected');
|
||||
|
||||
// 订阅行情数据
|
||||
ws.send(JSON.stringify({
|
||||
type: 'subscribe',
|
||||
data: { subscription: ['tick', 'order', 'trade'] }
|
||||
}));
|
||||
};
|
||||
|
||||
// 监听消息
|
||||
ws.onmessage = (event) => {
|
||||
const message = JSON.parse(event.data);
|
||||
console.log('Received:', message);
|
||||
};
|
||||
|
||||
// 发送心跳
|
||||
ws.send(JSON.stringify({ type: 'ping', data: {} }));
|
||||
```
|
||||
|
||||
## WebSocket 消息格式
|
||||
|
||||
### 订阅消息（客户端 -> 服务器）
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "subscribe",
|
||||
"data": {
|
||||
"subscription": ["tick", "order", "trade"]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 订阅品种（客户端 -> 服务器）
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "subscribe_symbol",
|
||||
"data": {
|
||||
"symbol": ["IF2501.CFFEX", "IH2501.CFFEX"]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 心跳消息（客户端 -> 服务器）
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "ping",
|
||||
"data": {}
|
||||
}
|
||||
```
|
||||
|
||||
### 推送消息（服务器 -> 客户端）
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "tick",
|
||||
"data": {
|
||||
"vt_symbol": "IF2501.CFFEX",
|
||||
"symbol": "IF2501",
|
||||
"exchange": "CFFEX",
|
||||
"last_price": 3500.0,
|
||||
"bid_price_1": 3499.0,
|
||||
"ask_price_1": 3501.0,
|
||||
"volume": 12345,
|
||||
"datetime": "2025-01-01T09:30:00"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 订阅类型
|
||||
|
||||
| 类型 | 说明 | 事件类型 |
|
||||
|------|------|----------|
|
||||
| tick | 行情数据 | eTick. |
|
||||
| order | 订单数据 | eOrder. |
|
||||
| trade | 成交数据 | eTrade. |
|
||||
| position | 持仓数据 | ePosition. |
|
||||
| account | 账户数据 | eAccount. |
|
||||
| log | 日志数据 | eLog |
|
||||
| contract | 合约数据 | eContract. |
|
||||
|
||||
## 验收标准
|
||||
|
||||
Phase 3 完成验收：
|
||||
- [x] WebSocket 管理器完成
|
||||
- [x] WebSocket 路由完成
|
||||
- [x] 事件监听器完成
|
||||
- [x] 心跳机制完成
|
||||
- [x] 订阅管理完成
|
||||
- [x] WebSocket 集成到 FastAPI
|
||||
- [x] 测试脚本和页面创建
|
||||
|
||||
## 故障排查
|
||||
|
||||
### 连接失败
|
||||
1. 检查服务器是否运行
|
||||
2. 检查 URL 是否正确
|
||||
3. 检查防火墙设置
|
||||
|
||||
### 没有收到消息
|
||||
1. 检查是否已订阅相应数据类型
|
||||
2. 检查 VeighNa 网关是否连接
|
||||
3. 检查是否有行情数据
|
||||
|
||||
### 心跳无响应
|
||||
1. 检查服务器负载
|
||||
2. 检查网络连接稳定性
|
||||
+15
-4
@@ -3,6 +3,8 @@
|
||||
Covers Task S0.1: build_app(config_path, static_dir) loads backtest.yaml and
|
||||
mounts SPA static files when the directory exists.
|
||||
"""
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from sanguo_api.main import build_app
|
||||
|
||||
|
||||
@@ -17,10 +19,20 @@ def _cfg(tmp_path) -> str:
|
||||
return str(cfg)
|
||||
|
||||
|
||||
def _has_route(app, path: str, method: str = "GET") -> bool:
|
||||
"""路由存在性探测（FastAPI 0.139+ 兼容）。
|
||||
|
||||
FastAPI 0.139 起 include_router 的子路由包成 _IncludedRouter，不再 flatten
|
||||
到 app.routes 顶层，因此 [r.path for r in app.routes] 看不到 /api/v1/* 路由。
|
||||
用 TestClient 实际探测：404 = 路由不存在，其他状态码（405/401/200）= 存在。
|
||||
前后端对齐的真实验证：前端能打的端点，后端确实挂载。
|
||||
"""
|
||||
return TestClient(app).request(method, path).status_code != 404
|
||||
|
||||
|
||||
def test_build_app_has_api_routes(tmp_path):
|
||||
app = build_app(_cfg(tmp_path))
|
||||
paths = [getattr(r, "path", "") for r in app.routes]
|
||||
assert "/api/v1/auth/login" in paths
|
||||
assert _has_route(app, "/api/v1/auth/login", "POST")
|
||||
|
||||
|
||||
def test_build_app_mounts_spa_when_static_exists(tmp_path):
|
||||
@@ -35,5 +47,4 @@ def test_build_app_mounts_spa_when_static_exists(tmp_path):
|
||||
def test_build_app_no_static_skips_mount(tmp_path):
|
||||
"""When static dir absent, build must still succeed (no SPA mount)."""
|
||||
app = build_app(_cfg(tmp_path), static_dir=str(tmp_path / "nope"))
|
||||
paths = [getattr(r, "path", "") for r in app.routes]
|
||||
assert "/api/v1/auth/login" in paths
|
||||
assert _has_route(app, "/api/v1/auth/login", "POST")
|
||||
|
||||
@@ -14,7 +14,7 @@ def test_read_parquet_daily_returns_bardata(tmp_path):
|
||||
"low": [9.8, 10.8], "close": [10.2, 11.2],
|
||||
"volume": [10000, 12000],
|
||||
})
|
||||
df.to_parquet(year_dir / "600000.parquet")
|
||||
df.to_parquet(year_dir / "sh600000_daily.parquet") # 实现读 {prefix}{symbol}_daily.parquet（line 24）
|
||||
|
||||
cfg = DataConfig(
|
||||
data_paths={"daily_dir": str(tmp_path)},
|
||||
@@ -64,7 +64,9 @@ def test_read_db_daily_configures_vnpy_settings():
|
||||
mock_db = MagicMock()
|
||||
mock_db.load_bar_data.return_value = [] # Return empty list to avoid data processing
|
||||
|
||||
with patch('sanguo_data.datareader.get_database', return_value=mock_db):
|
||||
# read_db_daily 内部 lazy import（from vnpy.trader.database import get_database），
|
||||
# 故 patch 源模块属性，非 sanguo_data.datareader（其上无 get_database）
|
||||
with patch('vnpy.trader.database.get_database', return_value=mock_db):
|
||||
# Call read_db_daily
|
||||
read_db_daily("600000", "2024-01-01", "2024-12-31", cfg)
|
||||
|
||||
|
||||
@@ -65,6 +65,11 @@ def test_compute_factors_calls_prepare_and_fetch(tmp_path):
|
||||
"""
|
||||
import pytest
|
||||
pytest.importorskip("polars")
|
||||
# vnpy.alpha 仅 vnpy_v4.4.0 源码提供（容器内 sys.path 引用）；本地 pip 装的 vnpy 无此子模块
|
||||
try:
|
||||
import vnpy.alpha # noqa: F401
|
||||
except ImportError:
|
||||
pytest.skip("vnpy.alpha 仅容器内可用（vnpy_v4.4.0 源码）")
|
||||
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
@@ -173,6 +178,11 @@ def test_compute_factors_passes_aware_periods_to_alpha_dataset(tmp_path):
|
||||
"""
|
||||
import pytest
|
||||
pytest.importorskip("polars")
|
||||
# vnpy.alpha 仅 vnpy_v4.4.0 源码提供（容器内 sys.path 引用）；本地 pip 装的 vnpy 无此子模块
|
||||
try:
|
||||
import vnpy.alpha # noqa: F401
|
||||
except ImportError:
|
||||
pytest.skip("vnpy.alpha 仅容器内可用（vnpy_v4.4.0 源码）")
|
||||
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
@@ -106,6 +106,7 @@ def test_run_factor_analysis_calls_tears(tmp_path):
|
||||
"""
|
||||
import pytest
|
||||
pytest.importorskip("polars")
|
||||
pytest.importorskip("alphalens")
|
||||
|
||||
from pathlib import Path
|
||||
from sanguo_factor.analyzer import run_factor_analysis
|
||||
|
||||
@@ -68,6 +68,12 @@ def test_convert_empty_bars():
|
||||
|
||||
def test_save_alpha_lab_data():
|
||||
"""Test save_alpha_lab_data creates AlphaLab and saves data."""
|
||||
import pytest
|
||||
pytest.importorskip("alphalens")
|
||||
try:
|
||||
import vnpy.alpha # noqa: F401
|
||||
except ImportError:
|
||||
pytest.skip("vnpy.alpha 仅容器内可用（vnpy_v4.4.0 源码）")
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from sanguo_factor.data_adapter import save_alpha_lab_data
|
||||
|
||||
@@ -1,437 +0,0 @@
|
||||
"""
|
||||
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"])
|
||||
@@ -1,416 +0,0 @@
|
||||
"""
|
||||
WebSocket 客户端测试脚本
|
||||
用于测试 WebSocket 实时数据推送功能
|
||||
"""
|
||||
import asyncio
|
||||
import websockets
|
||||
import json
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class WebSocketTestClient:
|
||||
"""WebSocket 测试客户端"""
|
||||
|
||||
def __init__(self, url: str = "ws://localhost:8000/ws"):
|
||||
"""
|
||||
初始化测试客户端
|
||||
|
||||
- **url**: WebSocket 服务器 URL
|
||||
"""
|
||||
self.url = url
|
||||
self.websocket: Optional[websockets.WebSocketClientProtocol] = None
|
||||
self.connected = False
|
||||
|
||||
async def connect(self, token: Optional[str] = None):
|
||||
"""
|
||||
连接到 WebSocket 服务器
|
||||
|
||||
- **token**: 可选的 JWT Token
|
||||
"""
|
||||
uri = f"{self.url}"
|
||||
if token:
|
||||
uri += f"?token={token}"
|
||||
|
||||
try:
|
||||
self.websocket = await websockets.connect(uri)
|
||||
self.connected = True
|
||||
print(f"[+] Connected to {self.url}")
|
||||
|
||||
# 接收欢迎消息
|
||||
welcome_msg = await self.websocket.recv()
|
||||
print(f"[+] Welcome message: {welcome_msg}")
|
||||
return json.loads(welcome_msg)
|
||||
except Exception as e:
|
||||
print(f"[-] Failed to connect: {e}")
|
||||
raise
|
||||
|
||||
async def disconnect(self):
|
||||
"""断开连接"""
|
||||
if self.websocket:
|
||||
await self.websocket.close()
|
||||
self.connected = False
|
||||
print("[+] Disconnected from server")
|
||||
|
||||
async def subscribe(self, subscription_types: list):
|
||||
"""
|
||||
订阅数据类型
|
||||
|
||||
- **subscription_types**: 订阅类型列表，如 ["tick", "order", "trade"]
|
||||
"""
|
||||
message = {
|
||||
"type": "subscribe",
|
||||
"data": {
|
||||
"subscription": subscription_types
|
||||
}
|
||||
}
|
||||
await self.send(message)
|
||||
print(f"[+] Subscribed to: {subscription_types}")
|
||||
|
||||
async def unsubscribe(self, subscription_types: list):
|
||||
"""
|
||||
取消订阅
|
||||
|
||||
- **subscription_types**: 订阅类型列表
|
||||
"""
|
||||
message = {
|
||||
"type": "unsubscribe",
|
||||
"data": {
|
||||
"subscription": subscription_types
|
||||
}
|
||||
}
|
||||
await self.send(message)
|
||||
print(f"[+] Unsubscribed from: {subscription_types}")
|
||||
|
||||
async def subscribe_symbol(self, symbols: list):
|
||||
"""
|
||||
订阅品种行情
|
||||
|
||||
- **symbols**: 品种代码列表，如 ["IF2501.CFFEX", "IH2501.CFFEX"]
|
||||
"""
|
||||
message = {
|
||||
"type": "subscribe_symbol",
|
||||
"data": {
|
||||
"symbol": symbols
|
||||
}
|
||||
}
|
||||
await self.send(message)
|
||||
print(f"[+] Subscribed to symbols: {symbols}")
|
||||
|
||||
async def unsubscribe_symbol(self, symbols: list):
|
||||
"""
|
||||
取消品种订阅
|
||||
|
||||
- **symbols**: 品种代码列表
|
||||
"""
|
||||
message = {
|
||||
"type": "unsubscribe_symbol",
|
||||
"data": {
|
||||
"symbol": symbols
|
||||
}
|
||||
}
|
||||
await self.send(message)
|
||||
print(f"[+] Unsubscribed from symbols: {symbols}")
|
||||
|
||||
async def send_ping(self):
|
||||
"""发送心跳"""
|
||||
message = {
|
||||
"type": "ping",
|
||||
"data": {}
|
||||
}
|
||||
await self.send(message)
|
||||
print("[+] Ping sent")
|
||||
|
||||
async def send(self, message: dict):
|
||||
"""
|
||||
发送消息
|
||||
|
||||
- **message**: 消息字典
|
||||
"""
|
||||
if not self.websocket or not self.connected:
|
||||
raise Exception("Not connected to WebSocket server")
|
||||
|
||||
await self.websocket.send(json.dumps(message))
|
||||
|
||||
async def receive(self, timeout: Optional[float] = None):
|
||||
"""
|
||||
接收消息
|
||||
|
||||
- **timeout**: 超时时间（秒）
|
||||
"""
|
||||
if not self.websocket or not self.connected:
|
||||
raise Exception("Not connected to WebSocket server")
|
||||
|
||||
try:
|
||||
message = await asyncio.wait_for(self.websocket.recv(), timeout=timeout)
|
||||
return json.loads(message)
|
||||
except asyncio.TimeoutError:
|
||||
return None
|
||||
|
||||
async def listen(self, duration: int = 10, print_messages: bool = True):
|
||||
"""
|
||||
监听消息
|
||||
|
||||
- **duration**: 监听时长（秒）
|
||||
- **print_messages**: 是否打印消息
|
||||
"""
|
||||
print(f"\n[*] Listening for messages ({duration}s)...")
|
||||
messages = []
|
||||
|
||||
try:
|
||||
while True:
|
||||
message = await asyncio.wait_for(self.websocket.recv(), timeout=duration)
|
||||
data = json.loads(message)
|
||||
messages.append(data)
|
||||
|
||||
if print_messages:
|
||||
msg_type = data.get("type", "unknown")
|
||||
print(f"[*] Received {msg_type}: {json.dumps(data, ensure_ascii=False)[:200]}...")
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
print(f"\n[+] Listening finished. Received {len(messages)} messages")
|
||||
return messages
|
||||
|
||||
|
||||
async def test_basic_connection():
|
||||
"""测试基本连接功能"""
|
||||
print("\n" + "="*50)
|
||||
print("TEST 1: Basic Connection")
|
||||
print("="*50)
|
||||
|
||||
client = WebSocketTestClient()
|
||||
|
||||
try:
|
||||
# 连接
|
||||
await client.connect()
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# 测试心跳
|
||||
await client.send_ping()
|
||||
response = await client.receive(timeout=2)
|
||||
if response and response.get("type") == "pong":
|
||||
print("[+] Ping/Pong test passed")
|
||||
|
||||
# 断开
|
||||
await client.disconnect()
|
||||
print("[+] Test passed")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"[-] Test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def test_subscription():
|
||||
"""测试订阅功能"""
|
||||
print("\n" + "="*50)
|
||||
print("TEST 2: Subscription")
|
||||
print("="*50)
|
||||
|
||||
client = WebSocketTestClient()
|
||||
|
||||
try:
|
||||
# 连接
|
||||
await client.connect()
|
||||
|
||||
# 订阅多个数据类型
|
||||
await client.subscribe(["tick", "order", "trade", "position", "account", "log"])
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
# 订阅品种
|
||||
await client.subscribe_symbol(["IF2501.CFFEX", "IH2501.CFFEX"])
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
# 取消订阅
|
||||
await client.unsubscribe(["log"])
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
# 断开
|
||||
await client.disconnect()
|
||||
print("[+] Test passed")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"[-] Test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def test_message_reception():
|
||||
"""测试消息接收功能"""
|
||||
print("\n" + "="*50)
|
||||
print("TEST 3: Message Reception")
|
||||
print("="*50)
|
||||
|
||||
client = WebSocketTestClient()
|
||||
|
||||
try:
|
||||
# 连接
|
||||
await client.connect()
|
||||
|
||||
# 订阅所有类型
|
||||
await client.subscribe(["tick", "order", "trade", "position", "account", "log", "contract"])
|
||||
|
||||
# 监听消息
|
||||
messages = await client.listen(duration=5)
|
||||
|
||||
print(f"\n[+] Received messages by type:")
|
||||
msg_types = {}
|
||||
for msg in messages:
|
||||
msg_type = msg.get("type", "unknown")
|
||||
msg_types[msg_type] = msg_types.get(msg_type, 0) + 1
|
||||
|
||||
for msg_type, count in msg_types.items():
|
||||
print(f" - {msg_type}: {count}")
|
||||
|
||||
# 断开
|
||||
await client.disconnect()
|
||||
print("[+] Test passed")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"[-] Test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def test_heartbeat():
|
||||
"""测试心跳机制"""
|
||||
print("\n" + "="*50)
|
||||
print("TEST 4: Heartbeat")
|
||||
print("="*50)
|
||||
|
||||
client = WebSocketTestClient()
|
||||
|
||||
try:
|
||||
# 连接
|
||||
await client.connect()
|
||||
|
||||
# 发送多次心跳
|
||||
for i in range(5):
|
||||
await client.send_ping()
|
||||
response = await client.receive(timeout=2)
|
||||
if response and response.get("type") == "pong":
|
||||
print(f"[+] Heartbeat {i+1}/5 successful")
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# 断开
|
||||
await client.disconnect()
|
||||
print("[+] Test passed")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"[-] Test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def test_error_handling():
|
||||
"""测试错误处理"""
|
||||
print("\n" + "="*50)
|
||||
print("TEST 5: Error Handling")
|
||||
print("="*50)
|
||||
|
||||
client = WebSocketTestClient()
|
||||
|
||||
try:
|
||||
# 连接
|
||||
await client.connect()
|
||||
|
||||
# 发送未知消息类型
|
||||
await client.send({
|
||||
"type": "unknown_type",
|
||||
"data": {}
|
||||
})
|
||||
|
||||
response = await client.receive(timeout=2)
|
||||
if response and response.get("type") == "error":
|
||||
print("[+] Error response received correctly")
|
||||
|
||||
# 断开
|
||||
await client.disconnect()
|
||||
print("[+] Test passed")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"[-] Test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def test_authenticated_connection():
|
||||
"""测试认证连接"""
|
||||
print("\n" + "="*50)
|
||||
print("TEST 6: Authenticated Connection")
|
||||
print("="*50)
|
||||
|
||||
# 注意：需要有效的 JWT Token
|
||||
# 这里测试无效 Token 的情况
|
||||
client = WebSocketTestClient()
|
||||
|
||||
try:
|
||||
# 使用无效 Token 连接
|
||||
await client.connect(token="invalid_token")
|
||||
print("[-] Should have failed with invalid token")
|
||||
await client.disconnect()
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"[+] Correctly rejected invalid token: {e}")
|
||||
return True
|
||||
|
||||
|
||||
async def run_all_tests():
|
||||
"""运行所有测试"""
|
||||
print("\n" + "="*50)
|
||||
print("WebSocket Test Suite")
|
||||
print("="*50)
|
||||
|
||||
tests = [
|
||||
("Basic Connection", test_basic_connection),
|
||||
("Subscription", test_subscription),
|
||||
("Message Reception", test_message_reception),
|
||||
("Heartbeat", test_heartbeat),
|
||||
("Error Handling", test_error_handling),
|
||||
("Authenticated Connection", test_authenticated_connection),
|
||||
]
|
||||
|
||||
results = []
|
||||
|
||||
for name, test_func in tests:
|
||||
try:
|
||||
result = await test_func()
|
||||
results.append((name, result))
|
||||
except Exception as e:
|
||||
print(f"[-] Test '{name}' crashed: {e}")
|
||||
results.append((name, False))
|
||||
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# 汇总结果
|
||||
print("\n" + "="*50)
|
||||
print("Test Results Summary")
|
||||
print("="*50)
|
||||
|
||||
passed = sum(1 for _, result in results if result)
|
||||
total = len(results)
|
||||
|
||||
for name, result in results:
|
||||
status = "✓ PASS" if result else "✗ FAIL"
|
||||
print(f"{status}: {name}")
|
||||
|
||||
print(f"\nTotal: {passed}/{total} tests passed")
|
||||
|
||||
return passed == total
|
||||
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
# 运行测试
|
||||
success = asyncio.run(run_all_tests())
|
||||
|
||||
if success:
|
||||
print("\n[+] All tests passed!")
|
||||
exit(0)
|
||||
else:
|
||||
print("\n[-] Some tests failed")
|
||||
exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,209 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
WebSocket 验证脚本
|
||||
验证 WebSocket 模块的基本结构和功能
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
|
||||
# 添加项目路径
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../vnpy_v4.4.0'))
|
||||
|
||||
|
||||
def test_imports():
|
||||
"""测试模块导入"""
|
||||
print("Testing imports...")
|
||||
|
||||
try:
|
||||
from sanguo_web.websocket.manager import ConnectionManager, manager
|
||||
print(" ✓ ConnectionManager imported")
|
||||
|
||||
from sanguo_web.websocket.routes import router
|
||||
print(" ✓ WebSocket router imported")
|
||||
|
||||
from sanguo_web.websocket.events import (
|
||||
EventMonitorManager,
|
||||
serialize_tick_data,
|
||||
serialize_order_data,
|
||||
serialize_trade_data
|
||||
)
|
||||
print(" ✓ Event monitors imported")
|
||||
|
||||
from sanguo_web.websocket import (
|
||||
ConnectionManager as CM,
|
||||
manager as mgr,
|
||||
router as ws_router,
|
||||
EventMonitorManager as EMM
|
||||
)
|
||||
print(" ✓ WebSocket module __init__ exports")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Import failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_manager():
|
||||
"""测试连接管理器"""
|
||||
print("\nTesting ConnectionManager...")
|
||||
|
||||
try:
|
||||
from sanguo_web.websocket.manager import ConnectionManager
|
||||
|
||||
manager = ConnectionManager()
|
||||
|
||||
# 测试连接计数
|
||||
count = manager.get_connection_count()
|
||||
assert count == 0, f"Expected 0 connections, got {count}"
|
||||
print(" ✓ Connection count initialized correctly")
|
||||
|
||||
# 测试用户连接计数
|
||||
user_count = manager.get_user_connection_count("test_user")
|
||||
assert user_count == 0, f"Expected 0 user connections, got {user_count}"
|
||||
print(" ✓ User connection count initialized correctly")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Manager test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_serialization():
|
||||
"""测试数据序列化函数"""
|
||||
print("\nTesting serialization functions...")
|
||||
|
||||
try:
|
||||
from sanguo_web.websocket.events import serialize_datetime
|
||||
from datetime import datetime
|
||||
|
||||
# 测试 datetime 序列化
|
||||
dt = datetime(2025, 1, 1, 12, 30, 45)
|
||||
serialized = serialize_datetime(dt)
|
||||
assert serialized == "2025-01-01T12:30:45", f"Expected ISO format, got {serialized}"
|
||||
print(" ✓ datetime serialization works")
|
||||
|
||||
# 测试 None 处理
|
||||
none_result = serialize_datetime(None)
|
||||
assert none_result is None, f"Expected None for None input, got {none_result}"
|
||||
print(" ✓ None datetime handled correctly")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Serialization test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_router():
|
||||
"""测试路由"""
|
||||
print("\nTesting WebSocket router...")
|
||||
|
||||
try:
|
||||
from sanguo_web.websocket.routes import router
|
||||
|
||||
# 检查路由
|
||||
routes = [route.path for route in router.routes]
|
||||
assert "/ws" in routes, "WebSocket route not found"
|
||||
print(" ✓ WebSocket /ws route exists")
|
||||
|
||||
assert "/ws/status" in routes, "WebSocket status route not found"
|
||||
print(" ✓ WebSocket /ws/status route exists")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Router test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_event_monitor_classes():
|
||||
"""测试事件监听器类"""
|
||||
print("\nTesting event monitor classes...")
|
||||
|
||||
try:
|
||||
from sanguo_web.websocket.events import (
|
||||
TickEventMonitor,
|
||||
OrderEventMonitor,
|
||||
TradeEventMonitor,
|
||||
PositionEventMonitor,
|
||||
AccountEventMonitor,
|
||||
LogEventMonitor,
|
||||
ContractEventMonitor,
|
||||
EventMonitorManager
|
||||
)
|
||||
|
||||
# 检查类是否具有必要的方法
|
||||
monitors = [
|
||||
TickEventMonitor,
|
||||
OrderEventMonitor,
|
||||
TradeEventMonitor,
|
||||
PositionEventMonitor,
|
||||
AccountEventMonitor,
|
||||
LogEventMonitor,
|
||||
ContractEventMonitor
|
||||
]
|
||||
|
||||
for monitor_class in monitors:
|
||||
# 检查是否有 stop 方法
|
||||
assert hasattr(monitor_class, 'stop'), f"{monitor_class.__name__} missing stop method"
|
||||
print(f" ✓ {monitor_class.__name__} has stop method")
|
||||
|
||||
# 检查 EventMonitorManager
|
||||
assert hasattr(EventMonitorManager, 'start_all'), "EventMonitorManager missing start_all method"
|
||||
assert hasattr(EventMonitorManager, 'stop_all'), "EventMonitorManager missing stop_all method"
|
||||
print(" ✓ EventMonitorManager has start_all and stop_all methods")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Event monitor test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
"""主测试函数"""
|
||||
print("="*50)
|
||||
print("WebSocket Phase 3 Verification")
|
||||
print("="*50)
|
||||
|
||||
tests = [
|
||||
("Import Test", test_imports),
|
||||
("Manager Test", test_manager),
|
||||
("Serialization Test", test_serialization),
|
||||
("Router Test", test_router),
|
||||
("Event Monitor Test", test_event_monitor_classes),
|
||||
]
|
||||
|
||||
results = []
|
||||
|
||||
for name, test_func in tests:
|
||||
print(f"\n{name}:")
|
||||
try:
|
||||
result = test_func()
|
||||
results.append((name, result))
|
||||
except Exception as e:
|
||||
print(f" ✗ Test crashed: {e}")
|
||||
results.append((name, False))
|
||||
|
||||
# 汇总结果
|
||||
print("\n" + "="*50)
|
||||
print("Test Results")
|
||||
print("="*50)
|
||||
|
||||
passed = sum(1 for _, result in results if result)
|
||||
total = len(results)
|
||||
|
||||
for name, result in results:
|
||||
status = "✓ PASS" if result else "✗ FAIL"
|
||||
print(f"{status}: {name}")
|
||||
|
||||
print(f"\nTotal: {passed}/{total} tests passed")
|
||||
|
||||
if passed == total:
|
||||
print("\n✓ All tests passed! Phase 3 implementation verified.")
|
||||
return 0
|
||||
else:
|
||||
print(f"\n✗ {total - passed} test(s) failed")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user