AcidMap/test_cd_integration.py

#!/usr/bin/env python3
"""
Cd预测集成测试脚本
@description: 测试Cd预测功能是否正确集成到AcidMap系统中
@author: AcidMap Team
@version: 1.0.0
"""

import os
import sys
import asyncio
import logging
from datetime import datetime

# 添加app目录到Python路径
sys.path.append(os.path.join(os.path.dirname(__file__), 'app'))

# 设置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

async def test_cd_prediction_integration():
    """
    测试Cd预测功能集成
    
    @description: 验证配置、服务和包装器是否正常工作
    """
    logger = logging.getLogger(__name__)
    logger.info("开始测试Cd预测功能集成")
    
    try:
        # 测试1: 验证配置模块
        logger.info("测试1: 验证配置模块")
        from app.config.cd_prediction_config import cd_config
        
        logger.info(f"项目根目录: {cd_config.project_root}")
        logger.info(f"Cd系统路径: {cd_config.get_cd_system_path()}")
        logger.info(f"输出目录: {cd_config.get_output_dir()}")
        
        # 检查Cd系统是否存在
        cd_system_path = cd_config.get_cd_system_path()
        if not os.path.exists(cd_system_path):
            logger.error(f"Cd预测系统路径不存在: {cd_system_path}")
            return False
        
        logger.info("✅ 配置模块测试通过")
        
        # 测试2: 验证包装器
        logger.info("测试2: 验证包装器模块")
        from app.utils.cd_prediction_wrapper import CdPredictionWrapper
        
        wrapper = CdPredictionWrapper(cd_system_path)
        logger.info("✅ 包装器初始化成功")
        
        # 测试3: 验证服务类
        logger.info("测试3: 验证服务类")
        from app.services.cd_prediction_service import CdPredictionService
        
        service = CdPredictionService()
        logger.info("✅ 服务类初始化成功")
        
        # 测试4: 检查输出目录
        logger.info("测试4: 检查输出目录")
        
        figures_dir = cd_config.get_output_dir("figures")
        raster_dir = cd_config.get_output_dir("raster")
        
        logger.info(f"图片输出目录: {figures_dir}")
        logger.info(f"栅格输出目录: {raster_dir}")
        
        if os.path.exists(figures_dir) and os.path.exists(raster_dir):
            logger.info("✅ 输出目录创建成功")
        else:
            logger.warning("⚠️ 输出目录可能有问题")
        
        # 测试5: 检查Cd系统的关键文件
        logger.info("测试5: 检查Cd系统的关键文件")
        
        required_files = [
            "main.py",
            "config.py", 
            "models",
            "analysis"
        ]
        
        missing_files = []
        for file_path in required_files:
            full_path = os.path.join(cd_system_path, file_path)
            if not os.path.exists(full_path):
                missing_files.append(file_path)
        
        if missing_files:
            logger.error(f"❌ Cd系统缺少关键文件: {missing_files}")
            return False
        else:
            logger.info("✅ Cd系统文件完整性检查通过")
        
        logger.info("🎉 所有测试通过！Cd预测功能集成成功！")
        return True
        
    except ImportError as e:
        logger.error(f"❌ 导入模块失败: {str(e)}")
        return False
    except Exception as e:
        logger.error(f"❌ 测试过程中发生错误: {str(e)}")
        return False

def test_api_endpoints():
    """
    测试API端点是否正确注册
    
    @description: 验证FastAPI应用是否正确包含Cd预测路由
    """
    logger = logging.getLogger(__name__)
    logger.info("测试API端点注册")
    
    try:
        from app.main import app
        
        # 获取所有路由
        routes = []
        for route in app.routes:
            if hasattr(route, 'path'):
                routes.append(route.path)
        
        # 检查Cd预测相关路由
        cd_routes = [route for route in routes if '/cd-prediction/' in route]
        
        expected_routes = [
            '/api/cd-prediction/crop-cd/generate-map',
            '/api/cd-prediction/effective-cd/generate-map',
            '/api/cd-prediction/crop-cd/download-map',
            '/api/cd-prediction/effective-cd/download-map',
            '/api/cd-prediction/crop-cd/download-histogram',
            '/api/cd-prediction/effective-cd/download-histogram'
        ]
        
        logger.info(f"发现的Cd预测路由: {cd_routes}")
        
        # 检查是否所有预期路由都存在
        found_routes = 0
        for expected_route in expected_routes:
            if any(expected_route in route for route in cd_routes):
                found_routes += 1
                logger.info(f"✅ 找到路由: {expected_route}")
            else:
                logger.warning(f"⚠️ 未找到路由: {expected_route}")
        
        if found_routes == len(expected_routes):
            logger.info("✅ 所有API端点注册成功")
            return True
        else:
            logger.warning(f"⚠️ 只找到 {found_routes}/{len(expected_routes)} 个预期路由")
            return False
            
    except Exception as e:
        logger.error(f"❌ API端点测试失败: {str(e)}")
        return False

def main():
    """
    主测试函数
    
    @description: 运行所有集成测试
    """
    print("=" * 60)
    print("🔬 Cd预测功能集成测试")
    print("=" * 60)
    
    # 运行异步测试
    integration_success = asyncio.run(test_cd_prediction_integration())
    
    # 运行API测试
    api_success = test_api_endpoints()
    
    print("\n" + "=" * 60)
    print("📊 测试结果汇总")
    print("=" * 60)
    
    if integration_success:
        print("✅ 集成功能测试: 通过")
    else:
        print("❌ 集成功能测试: 失败")
    
    if api_success:
        print("✅ API端点测试: 通过")
    else:
        print("❌ API端点测试: 失败")
    
    overall_success = integration_success and api_success
    
    if overall_success:
        print("\n🎉 恭喜！Cd预测功能已成功集成到AcidMap系统中！")
        print("\n📝 接下来您可以:")
        print("   1. 启动FastAPI服务器: uvicorn app.main:app --reload")
        print("   2. 访问API文档: http://localhost:8000/docs")
        print("   3. 测试Cd预测接口")
        print("   4. 参考 INTEGRATION_GUIDE.md 了解详细使用方法")
    else:
        print("\n❌ 集成存在问题，请检查错误信息并修复")
        print("\n🔧 常见问题排查:")
        print("   1. 确保 Cd_Prediction_Integrated_System 文件夹存在且完整")
        print("   2. 检查Python依赖是否安装完整")
        print("   3. 验证文件路径和权限")
    
    print("=" * 60)
    
    return overall_success

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)