DataOps-platform/test_cors_config.py

#!/usr/bin/env python3
"""
测试CORS配置的脚本
用于验证Flask应用的跨域设置是否正确
"""

import requests
import json
from datetime import datetime

def test_cors_preflight():
    """测试CORS预检请求"""
    base_url = "http://company.citupro.com:5500"
    endpoint = "/api/data_parse/get-calendar-info"
    
    # 测试OPTIONS预检请求
    print("=== 测试CORS预检请求 ===")
    try:
        headers = {
            'Origin': 'http://localhost:5173',
            'Access-Control-Request-Method': 'GET',
            'Access-Control-Request-Headers': 'Content-Type'
        }
        
        response = requests.options(f"{base_url}{endpoint}", headers=headers)
        
        print(f"预检请求状态码: {response.status_code}")
        print(f"预检请求响应头:")
        for key, value in response.headers.items():
            if key.lower().startswith('access-control'):
                print(f"  {key}: {value}")
        
        if response.status_code == 200:
            print("✅ 预检请求成功")
        else:
            print("❌ 预检请求失败")
            
    except Exception as e:
        print(f"❌ 预检请求异常: {e}")

def test_cors_actual_request():
    """测试实际的跨域请求"""
    base_url = "http://company.citupro.com:5500"
    endpoint = "/api/data_parse/get-calendar-info"
    
    print("\n=== 测试实际跨域请求 ===")
    try:
        # 使用当前日期
        today = datetime.now().strftime("%Y-%m-%d")
        
        headers = {
            'Origin': 'http://localhost:5173',
            'Content-Type': 'application/json'
        }
        
        response = requests.get(
            f"{base_url}{endpoint}?date={today}", 
            headers=headers
        )
        
        print(f"实际请求状态码: {response.status_code}")
        print(f"响应头:")
        for key, value in response.headers.items():
            if key.lower().startswith('access-control'):
                print(f"  {key}: {value}")
        
        if response.status_code == 200:
            print("✅ 实际请求成功")
            try:
                data = response.json()
                print(f"响应数据: {json.dumps(data, ensure_ascii=False, indent=2)}")
            except:
                print(f"响应内容: {response.text}")
        else:
            print("❌ 实际请求失败")
            print(f"响应内容: {response.text}")
            
    except Exception as e:
        print(f"❌ 实际请求异常: {e}")

def test_cors_headers():
    """测试CORS响应头"""
    base_url = "http://company.citupro.com:5500"
    endpoint = "/api/data_parse/get-calendar-info"
    
    print("\n=== 测试CORS响应头 ===")
    try:
        today = datetime.now().strftime("%Y-%m-%d")
        
        response = requests.get(f"{base_url}{endpoint}?date={today}")
        
        print("所有响应头:")
        for key, value in response.headers.items():
            print(f"  {key}: {value}")
        
        # 检查关键的CORS头部
        cors_headers = [
            'Access-Control-Allow-Origin',
            'Access-Control-Allow-Methods',
            'Access-Control-Allow-Headers',
            'Access-Control-Allow-Credentials'
        ]
        
        print("\nCORS相关头部:")
        for header in cors_headers:
            value = response.headers.get(header, '未设置')
            print(f"  {header}: {value}")
            
    except Exception as e:
        print(f"❌ 测试响应头异常: {e}")

if __name__ == "__main__":
    print("开始测试CORS配置...")
    print(f"测试时间: {datetime.now()}")
    print(f"目标服务器: http://company.citupro.com:5500")
    print(f"模拟前端: http://localhost:5173")
    print("=" * 50)
    
    test_cors_preflight()
    test_cors_actual_request()
    test_cors_headers()
    
    print("\n" + "=" * 50)
    print("CORS配置测试完成")