sanguo_vnpy_v2/sanguo_web/test_phase1.py

#!/usr/bin/env python3
"""
Phase 1 基础功能测试脚本

测试 FastAPI 应用是否能正常启动和响应基本请求
"""
import sys
import os

# 添加项目路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

def test_imports():
    """测试所有模块能否正常导入"""
    print("=" * 60)
    print("Testing imports...")
    print("=" * 60)

    try:
        # 测试 FastAPI 应用导入
        from sanguo_web.api import app
        print("  FastAPI app: OK")

        # 测试数据模型导入
        from sanguo_web.api.models import (
            LoginRequest, TokenResponse, SendOrderRequest,
            OrderResponse, HealthResponse
        )
        print("  Models: OK")

        # 测试依赖注入导入
        from sanguo_web.api.deps import (
            create_access_token, verify_token,
            get_current_user
        )
        print("  Dependencies: OK")

        # 测试 VeighNa 服务导入
        from sanguo_web.services.main_service import VeighNaService
        print("  VeighNa Service: OK")

        # 测试数据库模型导入
        from sanguo_web.database import (
            User, APIToken, init_database, get_db
        )
        print("  Database: OK")

        # 测试 WebSocket 管理器导入
        from sanguo_web.websocket.manager import ConnectionManager
        print("  WebSocket Manager: OK")

        print("\nAll imports successful!")
        return True

    except Exception as e:
        print(f"\nImport failed: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_fastapi_routes():
    """测试 FastAPI 路由是否正确注册"""
    print("\n" + "=" * 60)
    print("Testing FastAPI routes...")
    print("=" * 60)

    try:
        from sanguo_web.api import app

        # 收集所有路由（包括 _IncludedRouter 中的路由）
        routes = []
        for route in app.routes:
            if hasattr(route, 'path') and hasattr(route, 'methods'):
                for method in route.methods or []:
                    routes.append(f"{method} {route.path}")
            elif type(route).__name__ == '_IncludedRouter' and hasattr(route, 'original_router'):
                # 处理新版 FastAPI 的 _IncludedRouter
                prefix = route.include_context.prefix if hasattr(route, 'include_context') and route.include_context else ""
                for r in route.original_router.routes:
                    if hasattr(r, 'path') and hasattr(r, 'methods'):
                        full_path = f"{prefix}{r.path}"
                        for method in r.methods or []:
                            routes.append(f"{method} {full_path}")

        # 关键路由检查
        key_routes = [
            ("GET", "/"),
            ("GET", "/health"),
            ("GET", "/docs"),
            ("POST", "/api/v1/auth/login"),
            ("POST", "/api/v1/auth/verify"),
            ("GET", "/api/v1/system/health"),
        ]

        print(f"\nTotal routes: {len(routes)}")
        print("\nKey routes check:")

        all_found = True
        for method, path in key_routes:
            found = any(f"{method} {path}" in r for r in routes)
            status = "OK" if found else "MISSING"
            print(f"  {method:6} {path:35} [{status}]")
            if not found:
                all_found = False

        if all_found:
            print("\nAll key routes registered!")
        else:
            print("\nWARNING: Some routes are missing!")
            print("\nRegistered routes:")
            for r in routes:
                print(f"  {r}")

        return all_found

    except Exception as e:
        print(f"\nRoute check failed: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_token_generation():
    """测试 JWT Token 生成和验证"""
    print("\n" + "=" * 60)
    print("Testing JWT Token...")
    print("=" * 60)

    try:
        from sanguo_web.api.deps import create_access_token, verify_token

        # 创建 Token
        token = create_access_token(data={"sub": "test_user"})
        print(f"  Token created: {token[:50]}...")

        # 验证 Token
        payload = verify_token(token)
        print(f"  Token verified, user: {payload.get('sub')}")

        if payload.get('sub') == 'test_user':
            print("\nJWT Token test passed!")
            return True
        else:
            print("\nJWT Token test failed: user mismatch")
            return False

    except Exception as e:
        print(f"\nJWT Token test failed: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_pydantic_models():
    """测试 Pydantic 模型验证"""
    print("\n" + "=" * 60)
    print("Testing Pydantic models...")
    print("=" * 60)

    try:
        from sanguo_web.api.models import (
            LoginRequest, SendOrderRequest, HealthResponse
        )

        # 测试 LoginRequest
        login = LoginRequest(username="admin", password="secret")
        print(f"  LoginRequest: {login.username}")

        # 测试 SendOrderRequest
        order = SendOrderRequest(
            symbol="IF2024",
            exchange="CFFEX",
            direction="buy",
            order_type="limit",
            volume=1.0,
            price=3500.0
        )
        print(f"  SendOrderRequest: {order.symbol} {order.direction.value}")

        # 测试 HealthResponse
        from datetime import datetime
        health = HealthResponse(
            status="healthy",
            service="Sanguo VeighNa",
            version="1.0.0",
            timestamp=datetime.utcnow()
        )
        print(f"  HealthResponse: {health.status}")

        print("\nPydantic models test passed!")
        return True

    except Exception as e:
        print(f"\nPydantic models test failed: {e}")
        import traceback
        traceback.print_exc()
        return False


def main():
    """运行所有测试"""
    print("\n" + "=" * 60)
    print("Sanguo VeighNa Web API - Phase 1 Tests")
    print("=" * 60)

    results = []

    # 运行测试
    results.append(("Imports", test_imports()))
    results.append(("Routes", test_fastapi_routes()))
    results.append(("JWT Token", test_token_generation()))
    results.append(("Pydantic Models", test_pydantic_models()))

    # 总结
    print("\n" + "=" * 60)
    print("Test Summary")
    print("=" * 60)

    passed = sum(1 for _, r in results if r)
    total = len(results)

    for name, result in results:
        status = "PASSED" if result else "FAILED"
        print(f"  {name:30} [{status}]")

    print(f"\nTotal: {passed}/{total} tests passed")

    if passed == total:
        print("\n✓ All Phase 1 tests passed!")
        return 0
    else:
        print(f"\n✗ {total - passed} test(s) failed")
        return 1


if __name__ == "__main__":
    sys.exit(main())