config-man/examples/test_cloudflare_cdn.py

#!/usr/bin/env python3
"""
Cloudflare CDN刷新功能测试

这个脚本用于测试 config-man 的Cloudflare CDN刷新功能。
"""

import sys
import os
import json

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

from src.config_man.utils.cloudflare import CloudflareCDN, refresh_cloudflare_cache


def test_cloudflare_configuration():
    """测试Cloudflare配置"""
    
    print("🔧 测试Cloudflare配置...")
    
    cdn = CloudflareCDN()
    
    # 检查配置
    if cdn.is_configured():
        print("✅ Cloudflare配置完整")
        print(f"  API Token: {'已设置' if cdn.api_token else '未设置'}")
        print(f"  Zone ID: {'已设置' if cdn.zone_id else '未设置'}")
        print(f"  Timeout: {cdn.timeout}秒")
        print(f"  Retry Count: {cdn.retry_count}次")
        
        # 显示配置来源
        from src.config_man.utils.config import config
        config_api_token = config.get_cloudflare_api_token()
        config_zone_id = config.get_cloudflare_zone_id()
        
        if config_api_token:
            print("  📁 API Token来源: 配置文件")
        elif os.getenv('CLOUDFLARE_API_TOKEN'):
            print("  🌍 API Token来源: 环境变量")
        else:
            print("  ❌ API Token来源: 未设置")
            
        if config_zone_id:
            print("  📁 Zone ID来源: 配置文件")
        elif os.getenv('CLOUDFLARE_ZONE_ID'):
            print("  🌍 Zone ID来源: 环境变量")
        else:
            print("  ❌ Zone ID来源: 未设置")
        
        # 获取域名信息
        zone_info = cdn.get_zone_info()
        if zone_info:
            print(f"✅ 域名信息获取成功: {zone_info.get('name', 'Unknown')}")
        else:
            print("❌ 域名信息获取失败")
    else:
        print("❌ Cloudflare配置不完整")
        print("请通过以下方式之一设置:")
        print("1. 配置文件:")
        print("   config-man setup-cloudflare --api-token your_token --zone-id your_zone_id")
        print("2. 环境变量:")
        print("   export CLOUDFLARE_API_TOKEN='your_token'")
        print("   export CLOUDFLARE_ZONE_ID='your_zone_id'")


def test_cache_refresh():
    """测试缓存刷新功能"""
    
    print("\n🔄 测试缓存刷新功能...")
    
    # 测试URL列表
    test_urls = [
        "https://ab.bamboogames.cc/ab/0_29/androidconfig.json",
        "https://ab.bamboogames.cc/ab/0_29/iosconfig.json"
    ]
    
    print(f"测试URL列表: {test_urls}")
    
    # 测试缓存刷新
    success = refresh_cloudflare_cache(test_urls)
    
    if success:
        print("✅ Cloudflare CDN缓存刷新成功")
    else:
        print("❌ Cloudflare CDN缓存刷新失败")


def test_cache_refresh_by_tags():
    """测试通过标签刷新缓存"""
    
    print("\n🏷️ 测试通过标签刷新缓存...")
    
    cdn = CloudflareCDN()
    
    if not cdn.is_configured():
        print("❌ Cloudflare配置不完整，跳过标签测试")
        return
    
    # 测试标签
    test_tags = ["config-man", "version-0.29"]
    
    print(f"测试标签: {test_tags}")
    
    success = cdn.purge_cache_by_tags(test_tags)
    
    if success:
        print("✅ 通过标签刷新缓存成功")
    else:
        print("❌ 通过标签刷新缓存失败")


def test_entire_cache_refresh():
    """测试刷新整个缓存"""
    
    print("\n🌐 测试刷新整个缓存...")
    
    cdn = CloudflareCDN()
    
    if not cdn.is_configured():
        print("❌ Cloudflare配置不完整，跳过全量刷新测试")
        return
    
    print("警告: 这将刷新整个域名的缓存，请确认是否继续...")
    response = input("输入 'yes' 继续: ")
    
    if response.lower() == 'yes':
        success = cdn.purge_entire_cache()
        
        if success:
            print("✅ 整个缓存刷新成功")
        else:
            print("❌ 整个缓存刷新失败")
    else:
        print("⏭️ 跳过全量刷新测试")


def test_suggest_update_with_cdn():
    """测试建议更新功能中的CDN刷新"""
    
    print("\n📝 测试建议更新功能中的CDN刷新...")
    
    from src.config_man.core.suggest_update_command import SuggestUpdateCommand
    from src.config_man.utils.mock_rclone import patch_rclone
    
    # 启用模拟环境
    patch_rclone()
    
    # 创建建议更新命令处理器
    suggest_cmd = SuggestUpdateCommand("ab", "https://ab.bamboogames.cc")
    
    # 测试Android平台
    print("测试Android平台建议更新...")
    success = suggest_cmd.execute("android", "0.29", "0.30")
    
    if success:
        print("✅ Android平台建议更新成功，CDN刷新正常")
    else:
        print("❌ Android平台建议更新失败")
    
    # 测试iOS平台
    print("测试iOS平台建议更新...")
    success = suggest_cmd.execute("ios", "0.29", "0.30")
    
    if success:
        print("✅ iOS平台建议更新成功，CDN刷新正常")
    else:
        print("❌ iOS平台建议更新失败")


def show_cloudflare_setup_guide():
    """显示Cloudflare设置指南"""
    
    print("\n📖 Cloudflare设置指南")
    print("=" * 50)
    
    print("1. 获取Cloudflare API Token:")
    print("   - 登录Cloudflare控制台")
    print("   - 进入 'My Profile' -> 'API Tokens'")
    print("   - 创建新的API Token")
    print("   - 权限设置:")
    print("     * Zone:Zone:Read")
    print("     * Zone:Cache Purge:Edit")
    
    print("\n2. 获取Zone ID:")
    print("   - 在Cloudflare控制台中选择域名")
    print("   - 在右侧边栏找到 'Zone ID'")
    
    print("\n3. 设置环境变量:")
    print("   export CLOUDFLARE_API_TOKEN='your_api_token'")
    print("   export CLOUDFLARE_ZONE_ID='your_zone_id'")
    
    print("\n4. 验证配置:")
    print("   python examples/test_cloudflare_cdn.py")


if __name__ == "__main__":
    print("🧪 Cloudflare CDN刷新功能测试")
    print("=" * 50)
    
    test_cloudflare_configuration()
    test_cache_refresh()
    test_cache_refresh_by_tags()
    test_entire_cache_refresh()
    test_suggest_update_with_cdn()
    show_cloudflare_setup_guide()
    
    print("\n🎉 Cloudflare CDN刷新功能测试完成！")