|
|
"""
|
|
|
星耀数智(AmazingData)集成测试
|
|
|
|
|
|
测试覆盖:
|
|
|
- API端点测试
|
|
|
- 数据一致性检查
|
|
|
- 多数据源对比
|
|
|
- 完整工作流测试
|
|
|
"""
|
|
|
|
|
|
import unittest
|
|
|
import asyncio
|
|
|
import sys
|
|
|
import os
|
|
|
import json
|
|
|
from datetime import datetime, timedelta
|
|
|
from unittest.mock import Mock, patch, AsyncMock
|
|
|
|
|
|
# 添加项目根目录到路径
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
|
|
|
|
|
class TestAPIEndpoints(unittest.TestCase):
|
|
|
"""测试API端点"""
|
|
|
|
|
|
def setUp(self):
|
|
|
"""测试前准备"""
|
|
|
self.base_url = 'http://localhost:8080/v1'
|
|
|
self.headers = {'X-Admin-Token': 'demo-api-key-2024'}
|
|
|
|
|
|
@patch('urllib.request.urlopen')
|
|
|
def test_adapters_endpoint(self, mock_urlopen):
|
|
|
"""测试适配器列表端点"""
|
|
|
# Mock响应
|
|
|
mock_response = Mock()
|
|
|
mock_response.read.return_value = json.dumps({
|
|
|
'code': 200,
|
|
|
'message': 'success',
|
|
|
'data': {
|
|
|
'adapters': [
|
|
|
{'name': 'amazingdata', 'status': 'active', 'type': 'stock'}
|
|
|
]
|
|
|
}
|
|
|
}).encode()
|
|
|
mock_urlopen.return_value = mock_response
|
|
|
|
|
|
import urllib.request
|
|
|
req = urllib.request.Request(
|
|
|
f'{self.base_url}/admin/adapters',
|
|
|
headers=self.headers
|
|
|
)
|
|
|
response = urllib.request.urlopen(req, timeout=10)
|
|
|
data = json.loads(response.read().decode())
|
|
|
|
|
|
self.assertEqual(data['code'], 200)
|
|
|
self.assertIn('adapters', data['data'])
|
|
|
|
|
|
# 检查是否有amazingdata适配器
|
|
|
adapter_names = [a['name'] for a in data['data']['adapters']]
|
|
|
self.assertIn('amazingdata', adapter_names)
|
|
|
|
|
|
@patch('urllib.request.urlopen')
|
|
|
def test_source_status_endpoint(self, mock_urlopen):
|
|
|
"""测试数据源状态端点"""
|
|
|
mock_response = Mock()
|
|
|
mock_response.read.return_value = json.dumps({
|
|
|
'code': 200,
|
|
|
'message': 'success',
|
|
|
'data': {
|
|
|
'sources': [
|
|
|
{'id': 'amazingdata', 'name': '星耀数智', 'status': 'healthy'}
|
|
|
]
|
|
|
}
|
|
|
}).encode()
|
|
|
mock_urlopen.return_value = mock_response
|
|
|
|
|
|
import urllib.request
|
|
|
req = urllib.request.Request(
|
|
|
f'{self.base_url}/admin/source/status',
|
|
|
headers={'X-API-Key': 'demo-api-key-2024'}
|
|
|
)
|
|
|
response = urllib.request.urlopen(req, timeout=10)
|
|
|
data = json.loads(response.read().decode())
|
|
|
|
|
|
self.assertEqual(data['code'], 200)
|
|
|
|
|
|
|
|
|
class TestDataConsistency(unittest.TestCase):
|
|
|
"""测试数据一致性"""
|
|
|
|
|
|
def test_date_format_consistency(self):
|
|
|
"""测试日期格式一致性"""
|
|
|
from app.adapters.amazingdata_adapter import AmazingDataAdapter
|
|
|
|
|
|
adapter = AmazingDataAdapter()
|
|
|
|
|
|
# 测试不同格式的日期转换为统一格式
|
|
|
test_cases = [
|
|
|
('2024-01-01', 20240101),
|
|
|
('2024/01/01', 20240101),
|
|
|
('20240101', 20240101),
|
|
|
(20240101, 20240101),
|
|
|
]
|
|
|
|
|
|
for input_date, expected in test_cases:
|
|
|
result = adapter._format_date(input_date)
|
|
|
self.assertEqual(result, expected, f"Failed for {input_date}")
|
|
|
|
|
|
def test_symbol_format_consistency(self):
|
|
|
"""测试代码格式一致性"""
|
|
|
# 股票代码应该包含交易所后缀
|
|
|
valid_stock_codes = [
|
|
|
'000001.SZ',
|
|
|
'600000.SH',
|
|
|
'688001.SH',
|
|
|
'430001.BJ'
|
|
|
]
|
|
|
|
|
|
for code in valid_stock_codes:
|
|
|
# 检查格式: 代码.交易所
|
|
|
parts = code.split('.')
|
|
|
self.assertEqual(len(parts), 2, f"Invalid code format: {code}")
|
|
|
self.assertIn(parts[1], ['SH', 'SZ', 'BJ'], f"Invalid exchange: {parts[1]}")
|
|
|
|
|
|
|
|
|
class TestXYSZSpecificFeatures(unittest.TestCase):
|
|
|
"""测试星耀数智特有功能"""
|
|
|
|
|
|
def setUp(self):
|
|
|
"""测试前准备"""
|
|
|
self.adapter = None
|
|
|
try:
|
|
|
from app.adapters.amazingdata_adapter import AmazingDataAdapter
|
|
|
self.adapter_class = AmazingDataAdapter
|
|
|
except ImportError:
|
|
|
self.skipTest("AmazingData adapter not available")
|
|
|
|
|
|
def test_futures_exchange_mapping(self):
|
|
|
"""测试期货交易所映射"""
|
|
|
adapter = self.adapter_class()
|
|
|
|
|
|
# 测试上海期货交易所品种
|
|
|
self.assertEqual(adapter._get_futures_exchange('CU'), 'SHFE')
|
|
|
self.assertEqual(adapter._get_futures_exchange('AU'), 'SHFE')
|
|
|
|
|
|
# 测试大连商品交易所品种
|
|
|
self.assertEqual(adapter._get_futures_exchange('A'), 'DCE')
|
|
|
self.assertEqual(adapter._get_futures_exchange('M'), 'DCE')
|
|
|
|
|
|
# 测试郑州商品交易所品种
|
|
|
self.assertEqual(adapter._get_futures_exchange('CF'), 'CZCE')
|
|
|
self.assertEqual(adapter._get_futures_exchange('SR'), 'CZCE')
|
|
|
|
|
|
# 测试中金所品种
|
|
|
self.assertEqual(adapter._get_futures_exchange('IF'), 'CFFEX')
|
|
|
self.assertEqual(adapter._get_futures_exchange('IC'), 'CFFEX')
|
|
|
|
|
|
# 测试能源中心品种
|
|
|
self.assertEqual(adapter._get_futures_exchange('SC'), 'INE')
|
|
|
|
|
|
|
|
|
class TestWorkflow(unittest.IsolatedAsyncioTestCase):
|
|
|
"""测试完整工作流"""
|
|
|
|
|
|
async def asyncSetUp(self):
|
|
|
"""异步测试前准备"""
|
|
|
from app.adapters.amazingdata_adapter import AmazingDataAdapter
|
|
|
self.adapter = AmazingDataAdapter()
|
|
|
|
|
|
async def test_complete_workflow_mocked(self):
|
|
|
"""测试完整工作流（Mock版本）"""
|
|
|
# Mock所有依赖
|
|
|
self.adapter._ad = Mock()
|
|
|
self.adapter._base_data = Mock()
|
|
|
self.adapter._market_data = Mock()
|
|
|
self.adapter._info_data = Mock()
|
|
|
self.adapter._calendar = Mock()
|
|
|
self.adapter._is_logged_in = True
|
|
|
self.adapter._connected = True
|
|
|
|
|
|
# 1. 获取股票列表
|
|
|
self.adapter._base_data.get_code_list.return_value = [
|
|
|
'000001.SZ', '600000.SH'
|
|
|
]
|
|
|
|
|
|
import pandas as pd
|
|
|
self.adapter._base_data.get_code_info.return_value = pd.DataFrame({
|
|
|
'symbol': ['平安银行', '浦发银行']
|
|
|
}, index=['000001.SZ', '600000.SH'])
|
|
|
|
|
|
symbols = await self.adapter.fetch_symbols('stock')
|
|
|
self.assertEqual(len(symbols), 2)
|
|
|
|
|
|
# 2. 获取K线数据
|
|
|
kline_df = pd.DataFrame({
|
|
|
'open': [10.0], 'high': [11.0], 'low': [9.0],
|
|
|
'close': [10.5], 'volume': [10000], 'amount': [105000]
|
|
|
}, index=pd.to_datetime(['2024-01-01']))
|
|
|
|
|
|
self.adapter._market_data.query_kline.return_value = {
|
|
|
'000001.SZ': kline_df
|
|
|
}
|
|
|
|
|
|
klines = await self.adapter.fetch_klines(
|
|
|
symbol='000001.SZ',
|
|
|
start='20240101',
|
|
|
end='20240101',
|
|
|
freq='1d'
|
|
|
)
|
|
|
self.assertEqual(len(klines), 1)
|
|
|
|
|
|
# 3. 健康检查
|
|
|
self.adapter._base_data.get_code_list.return_value = ['000001.SZ']
|
|
|
health = await self.adapter.health_check()
|
|
|
self.assertTrue(health)
|
|
|
|
|
|
|
|
|
class TestErrorHandling(unittest.TestCase):
|
|
|
"""测试错误处理"""
|
|
|
|
|
|
def test_connection_error(self):
|
|
|
"""测试连接错误处理"""
|
|
|
from app.adapters.amazingdata_adapter import AmazingDataAdapter
|
|
|
|
|
|
adapter = AmazingDataAdapter()
|
|
|
|
|
|
# 测试未登录时调用方法
|
|
|
with self.assertRaises(RuntimeError) as context:
|
|
|
adapter._check_login()
|
|
|
|
|
|
self.assertIn('未连接到数据源', str(context.exception))
|
|
|
|
|
|
def test_invalid_date_format(self):
|
|
|
"""测试无效日期格式处理"""
|
|
|
from app.adapters.amazingdata_adapter import AmazingDataAdapter
|
|
|
|
|
|
adapter = AmazingDataAdapter()
|
|
|
|
|
|
# 测试无效输入
|
|
|
with self.assertRaises(ValueError):
|
|
|
adapter._format_date(None)
|
|
|
|
|
|
with self.assertRaises(ValueError):
|
|
|
adapter._format_date([])
|
|
|
|
|
|
with self.assertRaises(ValueError):
|
|
|
adapter._format_date({'date': '2024-01-01'})
|
|
|
|
|
|
|
|
|
class TestPerformanceRequirements(unittest.TestCase):
|
|
|
"""测试性能要求"""
|
|
|
|
|
|
def test_large_symbol_list_handling(self):
|
|
|
"""测试大批量代码处理"""
|
|
|
from app.adapters.amazingdata_adapter import AmazingDataAdapter
|
|
|
|
|
|
adapter = AmazingDataAdapter()
|
|
|
|
|
|
# 模拟5000只股票
|
|
|
large_code_list = [f'{i:06d}.SZ' for i in range(1, 5001)]
|
|
|
|
|
|
# 分批处理测试
|
|
|
batch_size = 50
|
|
|
batches = [large_code_list[i:i+batch_size] for i in range(0, len(large_code_list), batch_size)]
|
|
|
|
|
|
self.assertEqual(len(batches), 100)
|
|
|
self.assertEqual(len(batches[0]), 50)
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
unittest.main(verbosity=2)
|
