#!/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()