个股论坛重做

test_admin_query.py

@@ -0,0 +1,72 @@
+# -*- coding: utf-8 -*-
+"""
+测试管理员查询
+运行: python test_admin_query.py
+"""
+
+from sqlalchemy import create_engine, text
+
+# 数据库连接
+engine = create_engine(
+    "mysql+pymysql://root:Zzl33818!@127.0.0.1:3306/stock?charset=utf8mb4"
+)
+
+def test_admin_query():
+    print("=" * 50)
+    print("测试管理员查询")
+    print("=" * 50)
+
+    with engine.connect() as conn:
+        # 1. 查看表是否存在
+        print("\n1. 检查 community_admins 表...")
+        try:
+            result = conn.execute(text("SHOW TABLES LIKE 'community_admins'")).fetchone()
+            if result:
+                print("   ✓ 表存在")
+            else:
+                print("   ✗ 表不存在！")
+                return
+        except Exception as e:
+            print(f"   ✗ 错误: {e}")
+            return
+
+        # 2. 查看表中所有数据
+        print("\n2. 查看 community_admins 表中所有数据...")
+        try:
+            result = conn.execute(text("SELECT * FROM community_admins")).fetchall()
+            print(f"   共 {len(result)} 条记录:")
+            for row in result:
+                print(f"   - id={row.id}, user_id={row.user_id}, role={row.role}")
+        except Exception as e:
+            print(f"   ✗ 错误: {e}")
+
+        # 3. 查询 user_id=65
+        print("\n3. 查询 user_id=65...")
+        try:
+            sql = text("SELECT role, permissions FROM community_admins WHERE user_id = :user_id")
+            result = conn.execute(sql, {'user_id': 65}).fetchone()
+            if result:
+                print(f"   ✓ 找到记录: role={result.role}, permissions={result.permissions}")
+            else:
+                print("   ✗ 未找到 user_id=65 的记录！")
+        except Exception as e:
+            print(f"   ✗ 错误: {e}")
+
+        # 4. 查询 user_id='65' (字符串)
+        print("\n4. 查询 user_id='65' (字符串形式)...")
+        try:
+            sql = text("SELECT role, permissions FROM community_admins WHERE user_id = :user_id")
+            result = conn.execute(sql, {'user_id': '65'}).fetchone()
+            if result:
+                print(f"   ✓ 找到记录: role={result.role}")
+            else:
+                print("   ✗ 未找到记录")
+        except Exception as e:
+            print(f"   ✗ 错误: {e}")
+
+    print("\n" + "=" * 50)
+    print("测试完成")
+    print("=" * 50)
+
+if __name__ == '__main__':
+    test_admin_query()