DataOps-platform/test_enhanced_parse_neo4j.py

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

该脚本用于测试酒店职位数据Neo4j同步程序的增强功能，包括：
- 职位名称和级别名称的同步
- 节点关系的创建
"""

import os
import sys
import subprocess
import time

def test_enhanced_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("=" * 60)
    
    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✅ 增强版脚本执行成功!")
            
            # 检查输出中是否包含新的功能信息
            if "职位节点" in result.stdout and "级别节点" in result.stdout:
                print("✅ 检测到职位和级别节点创建功能")
            else:
                print("⚠️  未检测到职位和级别节点创建功能")
            
            if "BELONGS_TO" in result.stdout and "HAS_LEVEL" in result.stdout:
                print("✅ 检测到节点关系创建功能")
            else:
                print("⚠️  未检测到节点关系创建功能")
                
        else:
            print(f"\n❌ 增强版脚本执行失败，返回码: {result.returncode}")
            return False
            
        return True
        
    except Exception as e:
        print(f"执行增强版脚本时发生错误: {e}")
        return False

def test_import_enhanced_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模块导入成功")
        
        # 测试增强后的处理器类
        from app.core.data_parse.parse_neo4j_process import HotelPositionNeo4jProcessor
        print("✅ 增强后的处理器类导入成功")
        
        return True
        
    except ImportError as e:
        print(f"❌ 模块导入失败: {e}")
        return False

def test_enhanced_functionality():
    """测试增强功能"""
    print("\n测试增强功能...")
    
    try:
        from app.core.data_parse.parse_neo4j_process import HotelPositionNeo4jProcessor
        
        # 创建处理器实例
        processor = HotelPositionNeo4jProcessor()
        print("✅ 成功创建处理器实例")
        
        # 检查方法是否存在
        if hasattr(processor, 'create_relationship'):
            print("✅ 检测到关系创建方法")
        else:
            print("❌ 未检测到关系创建方法")
            
        if hasattr(processor, 'process_hotel_data'):
            print("✅ 检测到酒店数据处理方法")
        else:
            print("❌ 未检测到酒店数据处理方法")
            
        return True
        
    except Exception as e:
        print(f"❌ 测试增强功能时发生错误: {e}")
        return False

def main():
    """主函数"""
    print("酒店职位数据Neo4j同步程序增强功能测试")
    print("=" * 60)
    
    # 测试模块导入
    if not test_import_enhanced_modules():
        print("\n模块导入测试失败，跳过其他测试")
        return
    
    print("\n" + "=" * 60)
    
    # 测试增强功能
    if not test_enhanced_functionality():
        print("\n增强功能测试失败，跳过脚本执行测试")
        return
    
    print("\n" + "=" * 60)
    
    # 测试脚本执行
    if test_enhanced_parse_neo4j_process():
        print("\n🎉 所有增强功能测试通过!")
    else:
        print("\n💥 增强功能测试失败!")

if __name__ == "__main__":
    main()