DataOps-platform/test_neo4j_connection.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试Neo4j连接脚本
用于验证修改后的Neo4jDriver类是否可以在没有Flask的情况下工作
"""

import sys
import os

# 添加项目根目录到Python路径
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = current_dir
sys.path.insert(0, project_root)

try:
    from app.services.neo4j_driver import Neo4jDriver
    print("✅ 成功导入Neo4jDriver")
    
    # 测试Neo4j连接
    print("正在测试Neo4j连接...")
    neo4j_driver = Neo4jDriver(
        uri="bolt://192.168.3.143:7687",
        user="neo4j",
        password="cituneo4j",
        encrypted=False
    )
    
    if neo4j_driver.verify_connectivity():
        print("✅ Neo4j连接成功！")
        
        # 测试会话创建
        try:
            with neo4j_driver.get_session() as session:
                result = session.run("RETURN 1 as test")
                record = result.single()
                print(f"✅ 测试查询成功: {record['test']}")
        except Exception as e:
            print(f"❌ 测试查询失败: {e}")
    else:
        print("❌ Neo4j连接失败")
        
except ImportError as e:
    print(f"❌ 导入失败: {e}")
except Exception as e:
    print(f"❌ 其他错误: {e}")

print("测试完成")