DataOps-platform/test_hotel_group_brands.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试酒店集团品牌数据处理功能

该脚本用于测试parse_neo4j_process.py程序中新增的酒店集团品牌数据同步功能
"""

import os
import sys
import subprocess
import time

def test_hotel_group_brands_functionality():
    """测试酒店集团品牌数据处理功能"""
    
    # 获取脚本路径
    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("=" * 70)
    
    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:
                print("✅ 检测到酒店集团品牌数据处理功能")
            else:
                print("⚠️  未检测到酒店集团品牌数据处理功能")
            
            if "集团节点" in result.stdout and "品牌节点" in result.stdout:
                print("✅ 检测到集团和品牌节点创建功能")
            else:
                print("⚠️  未检测到集团和品牌节点创建功能")
            
            if "品牌级别节点" 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_hotel_group_brands_methods():
    """测试酒店集团品牌相关方法"""
    print("\n测试酒店集团品牌相关方法...")
    
    try:
        from app.core.data_parse.parse_neo4j_process import HotelPositionNeo4jProcessor
        
        # 创建处理器实例
        processor = HotelPositionNeo4jProcessor()
        print("✅ 成功创建处理器实例")
        
        # 检查酒店集团品牌相关方法是否存在
        if hasattr(processor, 'get_hotel_group_brands'):
            print("✅ 检测到获取酒店集团品牌数据方法")
        else:
            print("❌ 未检测到获取酒店集团品牌数据方法")
            
        if hasattr(processor, 'process_hotel_group_brands'):
            print("✅ 检测到处理酒店集团品牌数据方法")
        else:
            print("❌ 未检测到处理酒店集团品牌数据方法")
            
        if hasattr(processor, 'create_relationship'):
            print("✅ 检测到关系创建方法")
        else:
            print("❌ 未检测到关系创建方法")
            
        return True
        
    except Exception as e:
        print(f"❌ 测试酒店集团品牌方法时发生错误: {e}")
        return False

def test_data_structure():
    """测试数据结构"""
    print("\n测试数据结构...")
    
    try:
        from app.core.data_parse.parse_neo4j_process import HotelPositionNeo4jProcessor
        
        processor = HotelPositionNeo4jProcessor()
        
        # 测试连接数据库（如果可能）
        if processor.connect_postgresql():
            print("✅ PostgreSQL数据库连接成功")
            
            # 尝试获取酒店集团品牌数据
            try:
                brands = processor.get_hotel_group_brands()
                print(f"✅ 成功获取酒店集团品牌数据，共 {len(brands)} 条记录")
                
                if brands:
                    print("数据示例:")
                    for i, brand in enumerate(brands[:3]):
                        print(f"  {i+1}. 集团: {brand['group_name_zh']}, 品牌: {brand['brand_name_zh']}, 级别: {brand['positioning_level_zh']}")
                else:
                    print("⚠️  未获取到酒店集团品牌数据，可能是表为空或查询条件不匹配")
                    
            except Exception as e:
                print(f"⚠️  获取酒店集团品牌数据时发生错误: {e}")
                print("这可能是正常的，如果数据库中没有相应的表或数据")
        else:
            print("⚠️  PostgreSQL数据库连接失败，跳过数据获取测试")
            
        return True
        
    except Exception as e:
        print(f"❌ 测试数据结构时发生错误: {e}")
        return False

def main():
    """主函数"""
    print("酒店集团品牌数据处理功能测试")
    print("=" * 70)
    
    # 测试模块导入
    if not test_import_enhanced_modules():
        print("\n模块导入测试失败，跳过其他测试")
        return
    
    print("\n" + "=" * 70)
    
    # 测试酒店集团品牌相关方法
    if not test_hotel_group_brands_methods():
        print("\n酒店集团品牌方法测试失败，跳过其他测试")
        return
    
    print("\n" + "=" * 70)
    
    # 测试数据结构
    test_data_structure()
    
    print("\n" + "=" * 70)
    
    # 测试脚本执行
    if test_hotel_group_brands_functionality():
        print("\n🎉 酒店集团品牌功能测试通过!")
    else:
        print("\n💥 酒店集团品牌功能测试失败!")

if __name__ == "__main__":
    main()