""" 星耀数智(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)