book_rag_server/test/test_mysql_config.py

"""测试MySQL配置从.env文件读取"""

from utils.mysql.mysql_pool import get_mysql_pool
from utils.mysql.mysql_conn import get_mysql_conn


def test_mysql_pool_from_env():
    """测试从.env文件读取MySQL连接池配置"""
    print("=== 测试从.env文件读取MySQL连接池配置 ===")
    
    # 使用默认配置创建连接池（应从.env文件读取）
    pool = get_mysql_pool()
    print(f"连接池配置 - 主机: {pool.host}")
    print(f"连接池配置 - 端口: {pool.port}")
    print(f"连接池配置 - 用户名: {pool.user}")
    print(f"连接池配置 - 数据库: {pool.database}")
    print(f"连接池配置 - 字符集: {pool.charset}")
    print(f"连接池配置 - 连接池大小: {pool.pool_size}")
    
    print("\n✓ 从.env文件读取MySQL连接池配置成功！")


def test_mysql_conn_from_env():
    """测试从.env文件读取MySQL连接配置"""
    print("\n=== 测试从.env文件读取MySQL连接配置 ===")
    
    # 使用默认配置创建连接（应从.env文件读取）
    conn = get_mysql_conn()
    
    # 尝试执行简单查询
    try:
        # 获取游标上下文管理器
        with conn.get_cursor() as cursor:
            # 执行简单查询
            cursor.execute("SELECT 1 AS test")
            result = cursor.fetchone()
            print(f"执行简单查询结果: {result}")
        print("\n✓ 从.env文件读取MySQL连接配置成功！")
    except Exception as e:
        print(f"\n✗ 执行查询失败: {e}")
    finally:
        conn.close()


def test_mysql_conn_with_custom_params():
    """测试自定义参数覆盖.env配置"""
    print("\n=== 测试自定义参数覆盖.env配置 ===")
    
    # 使用自定义参数创建连接
    conn = get_mysql_conn(database="test")
    
    try:
        # 获取游标上下文管理器
        with conn.get_cursor() as cursor:
            # 执行简单查询
            cursor.execute("SELECT 1 AS test")
            result = cursor.fetchone()
            print(f"执行简单查询结果: {result}")
        print("\n✓ 自定义参数覆盖.env配置成功！")
    except Exception as e:
        print(f"\n✗ 执行查询失败: {e}")
    finally:
        conn.close()


if __name__ == "__main__":
    test_mysql_pool_from_env()
    test_mysql_conn_from_env()
    test_mysql_conn_with_custom_params()
    print("\n🎉 所有测试完成！")