DataOps-platform/scripts/test_deploy.py

#!/usr/bin/env python3
"""
测试自动部署功能

运行方式：
    python scripts/test_deploy.py
"""

import json
import sys
from pathlib import Path

# 添加项目根目录到路径
WORKSPACE_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(WORKSPACE_ROOT))


def test_ssh_connection():
    """测试 SSH 连接"""
    print("=" * 60)
    print("测试 1: SSH 连接测试")
    print("=" * 60)
    
    try:
        from scripts.auto_execute_tasks import test_ssh_connection
        result = test_ssh_connection()
        if result:
            print("✅ SSH 连接测试通过")
            return True
        else:
            print("❌ SSH 连接测试失败")
            return False
    except Exception as e:
        print(f"❌ 测试失败: {e}")
        return False


def test_deploy_functions():
    """测试部署函数是否可导入"""
    print("\n" + "=" * 60)
    print("测试 2: 部署函数导入测试")
    print("=" * 60)
    
    try:
        from scripts.auto_execute_tasks import (
            get_ssh_connection,
            deploy_script_to_production,
            deploy_n8n_workflow_to_production,
            auto_deploy_completed_task,
        )
        print("✅ 所有部署函数导入成功")
        return True
    except ImportError as e:
        print(f"❌ 函数导入失败: {e}")
        return False


def test_paramiko_installed():
    """测试 paramiko 是否已安装"""
    print("\n" + "=" * 60)
    print("测试 3: paramiko 库检查")
    print("=" * 60)
    
    try:
        import paramiko
        print(f"✅ paramiko 已安装，版本: {paramiko.__version__}")
        return True
    except ImportError:
        print("❌ paramiko 未安装")
        print("请运行: pip install paramiko")
        return False


def test_config():
    """测试配置是否正确"""
    print("\n" + "=" * 60)
    print("测试 4: 配置检查")
    print("=" * 60)
    
    try:
        from scripts.auto_execute_tasks import PRODUCTION_SERVER
        
        required_keys = ["host", "port", "username", "password", "script_path", "workflow_path"]
        missing_keys = [key for key in required_keys if key not in PRODUCTION_SERVER]
        
        if missing_keys:
            print(f"❌ 配置缺少必需字段: {missing_keys}")
            return False
        
        print("✅ 配置检查通过")
        print(f"   服务器: {PRODUCTION_SERVER['username']}@{PRODUCTION_SERVER['host']}:{PRODUCTION_SERVER['port']}")
        print(f"   脚本路径: {PRODUCTION_SERVER['script_path']}")
        print(f"   工作流路径: {PRODUCTION_SERVER['workflow_path']}")
        return True
        
    except Exception as e:
        print(f"❌ 配置检查失败: {e}")
        return False


def test_pending_tasks_file():
    """测试 pending_tasks.json 文件"""
    print("\n" + "=" * 60)
    print("测试 5: pending_tasks.json 文件检查")
    print("=" * 60)
    
    tasks_file = WORKSPACE_ROOT / "tasks" / "pending_tasks.json"
    
    if not tasks_file.exists():
        print("⚠️ pending_tasks.json 文件不存在（这是正常的，如果没有任务）")
        return True
    
    try:
        with tasks_file.open("r", encoding="utf-8") as f:
            tasks = json.load(f)
        
        if not isinstance(tasks, list):
            print("❌ pending_tasks.json 格式错误（应为数组）")
            return False
        
        print(f"✅ pending_tasks.json 文件正常，包含 {len(tasks)} 个任务")
        
        completed_tasks = [t for t in tasks if t.get("status") == "completed"]
        if completed_tasks:
            print(f"   其中 {len(completed_tasks)} 个任务已完成")
        
        return True
        
    except json.JSONDecodeError as e:
        print(f"❌ pending_tasks.json 解析失败: {e}")
        return False
    except Exception as e:
        print(f"❌ 文件读取失败: {e}")
        return False


def main():
    """主函数"""
    print("\n" + "=" * 70)
    print("🧪 自动部署功能测试套件")
    print("=" * 70)
    
    results = []
    
    # 运行所有测试
    results.append(("paramiko 库", test_paramiko_installed()))
    results.append(("配置检查", test_config()))
    results.append(("函数导入", test_deploy_functions()))
    results.append(("pending_tasks.json", test_pending_tasks_file()))
    
    # 只有在前面测试都通过的情况下才测试 SSH 连接
    if all(r[1] for r in results):
        results.append(("SSH 连接", test_ssh_connection()))
    else:
        print("\n⚠️ 跳过 SSH 连接测试（前置测试未通过）")
    
    # 输出测试结果
    print("\n" + "=" * 70)
    print("📊 测试结果汇总")
    print("=" * 70)
    
    for test_name, passed in results:
        status = "✅ 通过" if passed else "❌ 失败"
        print(f"{test_name:.<40} {status}")
    
    print("=" * 70)
    
    passed_count = sum(1 for _, passed in results if passed)
    total_count = len(results)
    
    if passed_count == total_count:
        print(f"✅ 所有测试通过 ({passed_count}/{total_count})")
        print("\n🎉 自动部署功能已就绪！")
        print("\n建议下一步操作：")
        print("  python scripts/auto_execute_tasks.py --chat-loop --use-agent")
        return 0
    else:
        print(f"⚠️ 部分测试失败 ({passed_count}/{total_count})")
        print("\n请根据上述错误信息进行修复")
        return 1


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