vf_react/test_wechat_api.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
微信 API 连接测试脚本
用于诊断微信登录 502 问题
"""

import sys
import time

def test_requests():
    """测试 requests 库是否正常工作"""
    print("=" * 50)
    print("🔍 测试 1: requests 库基本功能")
    print("=" * 50)
    
    try:
        import requests
        
        # 测试普通 HTTPS 请求
        print("  测试 HTTPS 请求...")
        start = time.time()
        response = requests.get("https://httpbin.org/get", timeout=10)
        elapsed = time.time() - start
        print(f"  ✅ httpbin.org 请求成功: {response.status_code} ({elapsed:.2f}s)")
        
        # 测试微信 API 连通性
        print("  测试微信 API 连通性...")
        start = time.time()
        response = requests.get("https://api.weixin.qq.com/", timeout=10)
        elapsed = time.time() - start
        print(f"  ✅ 微信 API 可达: {response.status_code} ({elapsed:.2f}s)")
        
        return True
    except Exception as e:
        print(f"  ❌ 请求失败: {e}")
        return False


def test_redis():
    """测试 Redis 连接"""
    print("\n" + "=" * 50)
    print("🔍 测试 2: Redis 连接")
    print("=" * 50)
    
    try:
        import redis
        r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
        
        # 测试连接
        r.ping()
        print("  ✅ Redis 连接成功")
        
        # 测试读写
        r.setex("test_key", 10, "test_value")
        value = r.get("test_key")
        r.delete("test_key")
        print(f"  ✅ Redis 读写测试成功: {value}")
        
        return True
    except Exception as e:
        print(f"  ❌ Redis 连接失败: {e}")
        return False


def test_gevent():
    """测试 gevent 环境"""
    print("\n" + "=" * 50)
    print("🔍 测试 3: Gevent 环境")
    print("=" * 50)
    
    try:
        from gevent import monkey
        is_patched = monkey.is_module_patched('socket')
        print(f"  Gevent monkey patched: {is_patched}")
        
        if is_patched:
            print("  ✅ Gevent 已正确 patch")
        else:
            print("  ⚠️ Gevent 未 patch，在 Gunicorn 下应该会自动 patch")
        
        return True
    except ImportError:
        print("  ⚠️ Gevent 未安装")
        return False


def test_wechat_access_token():
    """测试微信 access_token 获取（模拟）"""
    print("\n" + "=" * 50)
    print("🔍 测试 4: 微信 API 调用（模拟）")
    print("=" * 50)
    
    try:
        import requests
        
        # 使用一个无效的 code 测试 API 是否可达
        url = "https://api.weixin.qq.com/sns/oauth2/access_token"
        params = {
            'appid': 'test_appid',
            'secret': 'test_secret',
            'code': 'test_code',
            'grant_type': 'authorization_code'
        }
        
        print("  发送测试请求到微信 API...")
        start = time.time()
        response = requests.get(url, params=params, timeout=15)
        elapsed = time.time() - start
        
        data = response.json()
        print(f"  响应时间: {elapsed:.2f}s")
        print(f"  响应内容: {data}")
        
        # 预期会返回错误（因为参数无效），但这说明 API 可达
        if 'errcode' in data:
            print(f"  ✅ 微信 API 可达（预期的错误响应: {data.get('errmsg', '')}）")
            return True
        else:
            print("  ⚠️ 意外的响应格式")
            return True
            
    except requests.exceptions.Timeout:
        print("  ❌ 请求超时 - 网络可能有问题")
        return False
    except requests.exceptions.SSLError as e:
        print(f"  ❌ SSL 错误: {e}")
        print("  可能需要更新 CA 证书或检查网络代理设置")
        return False
    except Exception as e:
        print(f"  ❌ 请求失败: {e}")
        return False


def test_flask_app():
    """测试 Flask 应用是否能正常导入"""
    print("\n" + "=" * 50)
    print("🔍 测试 5: Flask 应用导入")
    print("=" * 50)
    
    try:
        print("  正在导入 app...")
        start = time.time()
        from app import app
        elapsed = time.time() - start
        print(f"  ✅ Flask 应用导入成功 ({elapsed:.2f}s)")
        return True
    except Exception as e:
        print(f"  ❌ Flask 应用导入失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def main():
    print("\n" + "=" * 60)
    print("   微信登录问题诊断工具")
    print("=" * 60)
    
    results = []
    
    results.append(("requests 库", test_requests()))
    results.append(("Redis 连接", test_redis()))
    results.append(("Gevent 环境", test_gevent()))
    results.append(("微信 API", test_wechat_access_token()))
    
    # Flask 导入测试可能会比较慢，放最后
    if '--skip-flask' not in sys.argv:
        results.append(("Flask 应用", test_flask_app()))
    
    print("\n" + "=" * 60)
    print("   测试结果汇总")
    print("=" * 60)
    
    all_passed = True
    for name, passed in results:
        status = "✅ 通过" if passed else "❌ 失败"
        print(f"  {name}: {status}")
        if not passed:
            all_passed = False
    
    print("\n" + "=" * 60)
    if all_passed:
        print("✅ 所有测试通过！")
        print("\n如果微信登录仍然 502，请检查:")
        print("  1. Gunicorn 日志中的具体错误信息")
        print("  2. 微信开放平台的回调 URL 配置")
        print("  3. 服务器防火墙是否允许访问微信 API")
    else:
        print("❌ 部分测试失败，请根据上述信息排查问题")
    print("=" * 60)


if __name__ == "__main__":
    main()