etf/archive/legacy_tests/tests/test_flask_api.py

"""
Flask API 快速测试
==================
测试 Flask 服务的基本功能
"""

import sys
from pathlib import Path

# 添加项目根目录到路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

from dotenv import load_dotenv
load_dotenv()

import requests
import json


def test_api():
    """测试 API 功能"""
    BASE_URL = "http://localhost:5000"
    
    print("\n" + "="*60)
    print("Flask API 测试")
    print("="*60)
    
    # 测试1: 健康检查
    print("\n1. 测试健康检查...")
    try:
        response = requests.get(f"{BASE_URL}/health", timeout=5)
        print(f"   状态码: {response.status_code}")
        print(f"   响应: {json.dumps(response.json(), indent=2, ensure_ascii=False)[:200]}")
    except Exception as e:
        print(f"   ✗ 失败: {e}")
    
    # 测试2: 首页
    print("\n2. 测试首页...")
    try:
        response = requests.get(f"{BASE_URL}/", timeout=5)
        print(f"   状态码: {response.status_code}")
        data = response.json()
        print(f"   API名称: {data.get('name')}")
        print(f"   版本: {data.get('version')}")
        print(f"   SSH状态: {data.get('ssh_status')}")
    except Exception as e:
        print(f"   ✗ 失败: {e}")
    
    # 测试3: 资产类型检测
    print("\n3. 测试资产类型检测...")
    codes = ["000300.SH", "NDX", "BTC"]
    for code in codes:
        try:
            response = requests.get(
                f"{BASE_URL}/api/v1/asset-type",
                params={"code": code},
                timeout=5
            )
            data = response.json()
            print(f"   {code:15s} -> {data.get('asset_type'):15s} ({data.get('description')})")
        except Exception as e:
            print(f"   {code:15s} ✗ 失败: {e}")
    
    # 测试4: 获取K线数据
    print("\n4. 测试获取K线数据...")
    try:
        response = requests.get(
            f"{BASE_URL}/api/v1/ohlcv",
            params={
                "code": "000300.SH",
                "start": "2024-01-01",
                "end": "2024-01-31"
            },
            timeout=30
        )
        data = response.json()
        if "error" in data:
            print(f"   ✗ 错误: {data['error']}")
        else:
            print(f"   ✓ 获取成功: {data.get('count')} 条")
            print(f"   资产类型: {data.get('asset_type')}")
            if data.get('data'):
                latest = data['data'][-1]
                print(f"   最新数据: {latest.get('date')} 收盘 {latest.get('close')}")
    except Exception as e:
        print(f"   ✗ 失败: {e}")
    
    # 测试5: 批量获取
    print("\n5. 测试批量获取...")
    try:
        response = requests.post(
            f"{BASE_URL}/api/v1/ohlcv/batch",
            json={
                "codes": ["000300.SH", "510300.SH"],
                "start": "2024-01-01",
                "end": "2024-01-31"
            },
            timeout=60
        )
        data = response.json()
        print(f"   成功: {data.get('success_count')}/{data.get('total')}")
        print(f"   失败: {data.get('failed_count')}/{data.get('total')}")
        
        for code, result in data.get('results', {}).items():
            if 'error' in result:
                print(f"   ✗ {code}: {result['error']}")
            else:
                print(f"   ✓ {code}: {result['count']} 条")
    except Exception as e:
        print(f"   ✗ 失败: {e}")
    
    # 测试6: 支持的代码
    print("\n6. 测试获取支持的代码...")
    try:
        response = requests.get(f"{BASE_URL}/api/v1/supported-codes", timeout=5)
        data = response.json()
        print(f"   支持的资产类型: {len(data)} 种")
        for asset_type, info in list(data.items())[:3]:
            print(f"   - {asset_type}: {info.get('description')}")
            print(f"     示例: {', '.join(info.get('examples', [])[:3])}")
    except Exception as e:
        print(f"   ✗ 失败: {e}")
    
    print("\n" + "="*60)
    print("测试完成")
    print("="*60)


if __name__ == "__main__":
    print("\n" + "="*60)
    print("Flask API 测试客户端")
    print("="*60)
    print("\n请确保 Flask 服务已启动:")
    print("   ./start_flask_server.sh")
    print("或")
    print("   python core/datasource/flask_server.py")
    print("")
    
    test_api()