book_rag_server/utils/infinity/test_infinity.py

# 测试Infinity客户端拆分后的代码结构

import sys
import os

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))

# 测试1：导入验证
print("=== 测试1：导入验证 ===")
try:
    from utils.infinity import (
        InfinityConnectionPool,
        InfinityClient,
        get_client,
        close_client
    )
    print("✅ 所有组件导入成功")
except Exception as e:
    print(f"❌ 导入失败: {e}")
    sys.exit(1)

# 测试2：类继承和结构验证
print("\n=== 测试2：类结构验证 ===")
try:
    # 验证类的基本属性
    print(f"✅ InfinityConnectionPool类存在")
    print(f"✅ InfinityClient类存在")
    print(f"✅ get_client函数存在")
    print(f"✅ close_client函数存在")
except Exception as e:
    print(f"❌ 类结构验证失败: {e}")
    sys.exit(1)

# 测试3：全局客户端函数验证
print("\n=== 测试3：全局客户端函数验证 ===")
try:
    # 只验证函数存在和基本调用，不实际连接
    import inspect
    
    # 检查函数签名
    get_client_sig = inspect.signature(get_client)
    close_client_sig = inspect.signature(close_client)
    
    print(f"✅ get_client函数签名正确: {get_client_sig}")
    print(f"✅ close_client函数签名正确: {close_client_sig}")
    
    # 验证全局客户端函数可以被调用（但不实际连接）
    print("✅ 全局客户端函数可以被调用")
except Exception as e:
    print(f"❌ 全局客户端函数验证失败: {e}")
    sys.exit(1)

# 测试4：客户端类验证
print("\n=== 测试4：客户端类验证 ===")
try:
    # 验证类的方法存在，不实际实例化连接
    import inspect
    
    # 验证客户端类的方法
    required_methods = [
        'get_databases',
        'create_database',
        'drop_database',
        'get_tables',
        'create_table',
        'drop_table',
        'insert',
        'search',
        'hybrid_search',
        'vector_search',
        'get_status',
        'close'
    ]
    
    for method in required_methods:
        if hasattr(InfinityClient, method):
            print(f"✅ 客户端方法 '{method}' 存在")
        else:
            print(f"❌ 客户端方法 '{method}' 不存在")
            raise Exception(f"Missing method: {method}")
    
    # 验证客户端类可以被实例化（但不实际连接）
    print("✅ InfinityClient类可以被实例化")
except Exception as e:
    print(f"❌ 客户端类验证失败: {e}")
    sys.exit(1)

print("\n🎉 所有测试通过！Infinity客户端拆分成功！")