LQAdminPlatform/test/test_db_connection_detailed.py

#!/usr/bin/env python3
"""
测试数据库连接
"""
import os
import sys

# 导入配置
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
from app.core.config import config_handler

print("=" * 60)
print("测试数据库连接")
print("=" * 60)

# 测试环境变量
database_url = config_handler.get("admin_app", "DATABASE_URL", "")
print(f"\n1. DATABASE_URL: {database_url}")

# 解析 URL
from urllib.parse import urlparse
parsed = urlparse(database_url)
print(f"\n2. 解析结果:")
print(f"   协议: {parsed.scheme}")
print(f"   主机: {parsed.hostname}")
print(f"   端口: {parsed.port}")
print(f"   用户: {parsed.username}")
print(f"   密码: {'*' * len(parsed.password) if parsed.password else 'None'}")
print(f"   数据库: {parsed.path[1:] if parsed.path else 'None'}")

# 测试 pymysql 连接
print(f"\n3. 测试 pymysql 连接...")
try:
    import pymysql
    conn = pymysql.connect(
        host=parsed.hostname,
        port=parsed.port,
        user=parsed.username,
        password=parsed.password,
        database=parsed.path[1:] if parsed.path else None,
        charset='utf8mb4'
    )
    print("   ✅ pymysql 连接成功！")
    
    # 测试查询
    cursor = conn.cursor()
    cursor.execute("SELECT VERSION()")
    version = cursor.fetchone()
    print(f"   MySQL 版本: {version[0]}")
    
    # 查询用户表
    cursor.execute("SHOW TABLES LIKE 'users'")
    result = cursor.fetchone()
    if result:
        print(f"   ✅ users 表存在")
        cursor.execute("SELECT COUNT(*) FROM users")
        count = cursor.fetchone()
        print(f"   用户数量: {count[0]}")
    else:
        print(f"   ❌ users 表不存在")
    
    cursor.close()
    conn.close()
    
except Exception as e:
    print(f"   ❌ 连接失败: {e}")
    import traceback
    traceback.print_exc()

print("\n" + "=" * 60)