DataOps-platform/test_parse_neo4j_process.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试 parse_neo4j_process.py 程序

该脚本用于测试酒店职位数据Neo4j同步程序的功能
"""

import os
import sys
import subprocess
import time

def test_parse_neo4j_process():
    """测试parse_neo4j_process.py程序"""
    
    # 获取脚本路径
    script_path = os.path.join('app', 'core', 'data_parse', 'parse_neo4j_process.py')
    
    if not os.path.exists(script_path):
        print(f"错误: 找不到脚本文件 {script_path}")
        return False
    
    print(f"开始测试脚本: {script_path}")
    print("=" * 50)
    
    try:
        # 运行脚本
        start_time = time.time()
        result = subprocess.run(
            [sys.executable, script_path],
            capture_output=True,
            text=True,
            encoding='utf-8',
            cwd=os.getcwd()
        )
        end_time = time.time()
        
        # 输出结果
        print(f"脚本执行时间: {end_time - start_time:.2f} 秒")
        print(f"返回码: {result.returncode}")
        print("\n标准输出:")
        print(result.stdout)
        
        if result.stderr:
            print("\n标准错误:")
            print(result.stderr)
        
        # 判断执行结果
        if result.returncode == 0:
            print("\n✅ 脚本执行成功!")
            return True
        else:
            print(f"\n❌ 脚本执行失败，返回码: {result.returncode}")
            return False
            
    except Exception as e:
        print(f"执行脚本时发生错误: {e}")
        return False

def test_import_modules():
    """测试模块导入"""
    print("测试模块导入...")
    
    try:
        # 测试配置模块
        from app.config.config import config, current_env
        print("✅ 配置模块导入成功")
        
        # 测试Neo4j驱动模块
        from app.services.neo4j_driver import Neo4jDriver
        print("✅ Neo4j驱动模块导入成功")
        
        # 测试SQLAlchemy模块
        from sqlalchemy import create_engine, text
        print("✅ SQLAlchemy模块导入成功")
        
        return True
        
    except ImportError as e:
        print(f"❌ 模块导入失败: {e}")
        return False

def main():
    """主函数"""
    print("酒店职位数据Neo4j同步程序测试")
    print("=" * 50)
    
    # 测试模块导入
    if not test_import_modules():
        print("\n模块导入测试失败，跳过脚本执行测试")
        return
    
    print("\n" + "=" * 50)
    
    # 测试脚本执行
    if test_parse_neo4j_process():
        print("\n🎉 所有测试通过!")
    else:
        print("\n💥 测试失败!")

if __name__ == "__main__":
    main()