diff --git a/backend/app/api/v1/endpoints/query.py b/backend/app/api/v1/endpoints/query.py index 89ceacc..90568b1 100644 --- a/backend/app/api/v1/endpoints/query.py +++ b/backend/app/api/v1/endpoints/query.py @@ -180,6 +180,7 @@ async def execute_query( try: username = current_user.display_name or current_user.username await QueryHistoryService.record_query( + db=db, user_id=current_user.id, username=username, instance_id=instance.id, diff --git a/backend/app/api/v1/endpoints/query_history.py b/backend/app/api/v1/endpoints/query_history.py index d7cd4b5..bc713eb 100644 --- a/backend/app/api/v1/endpoints/query_history.py +++ b/backend/app/api/v1/endpoints/query_history.py @@ -18,6 +18,7 @@ async def my_query_history( keyword: Optional[str] = None, page: int = 1, page_size: int = 20, + db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """获取我的查询历史。 @@ -25,6 +26,7 @@ async def my_query_history( 自动记录每次查询，支持按数据库类型、实例、关键词筛选。 """ return await QueryHistoryService.get_history( + db=db, user_id=current_user.id, db_type=db_type, instance_id=instance_id, @@ -37,10 +39,11 @@ async def my_query_history( @router.post("/{history_id}/favorite") async def toggle_favorite( history_id: int, + db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """切换查询历史收藏状态。""" - is_fav = await QueryHistoryService.toggle_favorite(current_user.id, history_id) + is_fav = await QueryHistoryService.toggle_favorite(db, current_user.id, history_id) return {"is_favorite": is_fav, "message": "已收藏" if is_fav else "已取消收藏"} @@ -48,20 +51,22 @@ async def toggle_favorite( async def my_favorites( page: int = 1, page_size: int = 20, + db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """获取我收藏的查询。""" return await QueryHistoryService.get_favorites( - user_id=current_user.id, page=page, page_size=page_size + db=db, user_id=current_user.id, page=page, page_size=page_size ) @router.get("/statistics") async def my_query_statistics( + db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """获取我的查询统计 — 了解自己的数据使用习惯。""" - return await QueryHistoryService.get_query_statistics(user_id=current_user.id) + return await QueryHistoryService.get_query_statistics(db=db, user_id=current_user.id) # ── 快捷查询 ── @@ -70,11 +75,12 @@ async def my_query_statistics( async def list_quick_queries( db_type: Optional[str] = None, keyword: Optional[str] = None, + db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """获取快捷查询列表（个人+团队共享）。""" items = await QueryHistoryService.get_quick_queries( - user_id=current_user.id, db_type=db_type, keyword=keyword + db=db, user_id=current_user.id, db_type=db_type, keyword=keyword ) return {"items": items, "total": len(items)} @@ -87,10 +93,12 @@ async def create_quick_query( description: str = "", database_pattern: str = "*", is_shared: bool = False, + db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """创建快捷查询。""" qq = await QueryHistoryService.create_quick_query( + db=db, name=name, description=description, query_content=query_content, diff --git a/backend/app/api/v1/endpoints/sql_assistant.py b/backend/app/api/v1/endpoints/sql_assistant.py index 171b347..ebf2251 100644 --- a/backend/app/api/v1/endpoints/sql_assistant.py +++ b/backend/app/api/v1/endpoints/sql_assistant.py @@ -1,11 +1,20 @@ """SQL助手 API — 人性化辅助功能。""" +import time from typing import Optional + from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from app.core.database import get_db +from app.core.crypto import decrypt_password +from app.core.permissions import check_instance_access, check_instance_permission from app.core.security import get_current_user from app.models.user import User +from app.models.instance import Instance from app.services.sql_assistant import SQLAssistant +from app.services.explain_service import ExplainService router = APIRouter(prefix="/sql-assistant", tags=["SQL助手"]) @@ -111,3 +120,63 @@ async def parse_explain( "warnings": result.warnings, "human_readable": result.human_readable, } + + +# ===== EXPLAIN 真实执行 API ===== + + +class ExplainRequest(BaseModel): + """EXPLAIN 请求体。""" + instance_id: int + database: str + query: str + + +@router.post("/explain") +async def execute_explain( + req: ExplainRequest, + db: AsyncSession = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """连接数据源执行 EXPLAIN 并解析为统一树结构。 + + 根据数据源类型自动构造对应的 EXPLAIN 语句: + - MySQL: EXPLAIN FORMAT=JSON + - PostgreSQL: EXPLAIN (FORMAT JSON, ANALYZE false) + - SQLite: EXPLAIN QUERY PLAN + + 返回统一的树形执行计划、优化警告和汇总统计。 + """ + if current_user.role == "guest": + raise HTTPException(status_code=403, detail="访客无权执行 EXPLAIN，请联系管理员申请权限") + + if not req.query or not req.query.strip(): + raise HTTPException(status_code=400, detail="查询语句不能为空") + + # 获取实例 + result = await db.execute(select(Instance).where(Instance.id == req.instance_id)) + instance = result.scalar_one_or_none() + if not instance: + raise HTTPException(status_code=404, detail="数据源不存在，请确认实例ID正确") + + # 权限检查 + await check_instance_access(current_user, instance, db) + await check_instance_permission(current_user, instance, "query:select", db) + + # 解密密码 + instance.password = decrypt_password(instance.password) + + start_time = time.time() + try: + explain_result = await ExplainService.execute_explain( + instance, req.database, req.query + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + raise HTTPException(status_code=500, detail=f"EXPLAIN 执行失败: {e}") + + elapsed_ms = int((time.time() - start_time) * 1000) + explain_result["elapsed_ms"] = elapsed_ms + + return explain_result diff --git a/backend/app/services/explain_service.py b/backend/app/services/explain_service.py new file mode 100644 index 0000000..375f28b --- /dev/null +++ b/backend/app/services/explain_service.py @@ -0,0 +1,789 @@ +"""EXPLAIN 执行服务 — 连接数据源执行 EXPLAIN 并解析为统一树结构。 + +支持: +- MySQL: EXPLAIN FORMAT=JSON +- PostgreSQL: EXPLAIN (FORMAT JSON, ANALYZE false) +- SQLite: EXPLAIN QUERY PLAN + +返回统一树结构: +{ + "execution_plan": [...], # 树形执行计划节点 + "warnings": [...], # 优化警告 + "summary": {...}, # 汇总统计 + "db_type": "mysql", + "raw_result": {...}, # 原始结果 +} +""" +import asyncio +import json +import logging +from typing import Any, Dict, List, Optional, Tuple + +# 顶层 import 以便测试 mock（实际运行时为可选依赖） +try: + import aiomysql # noqa: F401 +except ImportError: + aiomysql = None # type: ignore[assignment] + +try: + import asyncpg # noqa: F401 +except ImportError: + asyncpg = None # type: ignore[assignment] + +from app.models.instance import Instance + +logger = logging.getLogger(__name__) + + +# ===== 统一节点结构 ===== + +def make_plan_node( + id: int, + operation: str, + table: Optional[str] = None, + rows: Optional[int] = None, + cost: Optional[float] = None, + key: Optional[str] = None, + extra: Optional[str] = None, + children: Optional[List[Dict]] = None, + **kwargs, +) -> Dict[str, Any]: + """构建统一执行计划节点。""" + node = { + "id": id, + "operation": operation, + "table": table, + "rows": rows, + "cost": cost, + "key": key, + "extra": extra, + "children": children or [], + } + # 附加额外字段 + for k, v in kwargs.items(): + if v is not None: + node[k] = v + return node + + +# ===== MySQL EXPLAIN FORMAT=JSON 解析 ===== + +def parse_mysql_explain_json(raw_json: Any) -> Dict[str, Any]: + """解析 MySQL EXPLAIN FORMAT=JSON 的结果。 + + MySQL EXPLAIN FORMAT=JSON 结构示例: + { + "query_block": { + "select_id": 1, + "cost_info": {"query_cost": "0.35"}, + "table": { + "table_name": "users", + "access_type": "ALL", + "rows_examined_per_scan": 1000, + "rows_produced_per_join": 1000, + "filtered": "100.00", + "cost_info": {...}, + "used_columns": [...] + }, + "nested_loop": [...] + } + } + + Returns: + 统一结构的 Dict + """ + if isinstance(raw_json, str): + try: + raw_json = json.loads(raw_json) + except (json.JSONDecodeError, TypeError): + return _empty_result("mysql", raw_json, error="EXPLAIN JSON 解析失败") + + nodes: List[Dict] = [] + warnings: List[str] = [] + node_id_counter = [0] + + def next_id() -> int: + node_id_counter[0] += 1 + return node_id_counter[0] + + def parse_query_block(qb: Dict) -> Dict[str, Any]: + """递归解析 query_block。""" + node_id = next_id() + children = [] + operation = "Query Block" + table_name = None + access_type = None + key_used = None + rows_est = None + cost_val = None + extra_parts = [] + + # 表信息 + table_info = qb.get("table") + if table_info: + table_name = table_info.get("table_name") + access_type = table_info.get("access_type", "") + key_used = table_info.get("key") + possible_keys = table_info.get("possible_keys") + rows_est = table_info.get("rows_examined_per_scan") + filtered = table_info.get("filtered", "100.00") + + # 成本 + cost_info = table_info.get("cost_info", {}) + cost_str = cost_info.get("read_cost") or cost_info.get("eval_cost") or cost_info.get("prefix_cost") + if cost_str: + try: + cost_val = float(cost_str) + except (ValueError, TypeError): + pass + + # 操作类型映射 + operation = _mysql_access_type_to_operation(access_type) + + # Extra 信息 + if key_used: + extra_parts.append(f"使用索引: {key_used}") + if possible_keys and not key_used: + extra_parts.append(f"可选索引: {possible_keys}") + if filtered and filtered != "100.00": + extra_parts.append(f"过滤率: {filtered}%") + + used_key_parts = table_info.get("used_key_parts") + if used_key_parts: + extra_parts.append(f"索引列: {', '.join(str(p) for p in used_key_parts)}") + + # 警告检测 + if access_type == "ALL": + warnings.append(f"表 '{table_name}' 全表扫描，建议添加索引") + if access_type == "index": + warnings.append(f"表 '{table_name}' 全索引扫描") + + # 嵌套循环 / join + nested_loop = qb.get("nested_loop") + if nested_loop: + for nl_item in nested_loop: + child = parse_query_block(nl_item) + children.append(child) + + # 排序 + order_by = qb.get("ordering_operation") + if order_by: + sort_node = _parse_mysql_ordering(order_by, next_id()) + children.append(sort_node) + extra_parts.append("Using filesort") + warnings.append("使用了文件排序 (filesort)") + + # 分组 + grouping = qb.get("grouping_operation") + if grouping: + group_node = _parse_mysql_grouping(grouping, next_id()) + children.append(group_node) + extra_parts.append("Using temporary") + warnings.append("使用了临时表 (temporary)") + + # 子查询 + subqueries = qb.get("subqueries") or qb.get("optimized_away_subqueries") + if subqueries: + for sq in subqueries: + if "query_block" in sq: + child = parse_query_block(sq["query_block"]) + children.append(child) + + # 成本信息 + qb_cost = qb.get("cost_info", {}) + if not cost_val and "query_cost" in qb_cost: + try: + cost_val = float(qb_cost["query_cost"]) + except (ValueError, TypeError): + pass + + return make_plan_node( + id=node_id, + operation=operation, + table=table_name, + rows=int(rows_est) if rows_est else None, + cost=cost_val, + key=key_used, + extra="; ".join(extra_parts) if extra_parts else None, + children=children, + access_type=access_type, + ) + + # 解析顶层 query_block + query_block = raw_json.get("query_block") if isinstance(raw_json, dict) else None + if query_block: + root = parse_query_block(query_block) + nodes = [root] + else: + warnings.append("无法识别的 EXPLAIN JSON 结构") + + # 汇总 + summary = _build_summary(nodes, warnings, "mysql") + + return { + "execution_plan": nodes, + "warnings": warnings, + "summary": summary, + "db_type": "mysql", + "raw_result": raw_json, + } + + +def _parse_mysql_ordering(ordering: Dict, node_id: int) -> Dict: + """解析 MySQL 排序操作。""" + table_info = ordering.get("table", {}) + return make_plan_node( + id=node_id, + operation="Filesort", + table=table_info.get("table_name"), + extra="排序操作", + ) + + +def _parse_mysql_grouping(grouping: Dict, node_id: int) -> Dict: + """解析 MySQL 分组操作。""" + table_info = grouping.get("table", {}) + return make_plan_node( + id=node_id, + operation="Temporary Table", + table=table_info.get("table_name"), + extra="分组操作使用临时表", + ) + + +def _mysql_access_type_to_operation(access_type: Optional[str]) -> str: + """将 MySQL access_type 映射为统一操作名。""" + mapping = { + "system": "System Scan", + "const": "Const Row", + "eq_ref": "Unique Lookup", + "ref": "Index Lookup", + "fulltext": "Fulltext Search", + "ref_or_null": "Index Lookup (nullable)", + "index_merge": "Index Merge", + "unique_subquery": "Unique Subquery", + "index_subquery": "Index Subquery", + "range": "Index Range Scan", + "index": "Full Index Scan", + "ALL": "Table Scan", + } + return mapping.get(access_type or "", f"Unknown ({access_type})") + + +# ===== PostgreSQL EXPLAIN (FORMAT JSON) 解析 ===== + +def parse_postgres_explain_json(raw_json: Any) -> Dict[str, Any]: + """解析 PostgreSQL EXPLAIN (FORMAT JSON) 的结果。 + + PostgreSQL EXPLAIN JSON 结构示例: + [ + { + "Plan": { + "Node Type": "Seq Scan", + "Relation Name": "users", + "Alias": "users", + "Startup Cost": 0.00, + "Total Cost": 15.50, + "Plan Rows": 1000, + "Plan Width": 68, + "Plans": [ + { + "Node Type": "Index Scan", + ... + "Plans": [] + } + ] + } + } + ] + + Returns: + 统一结构的 Dict + """ + if isinstance(raw_json, str): + try: + raw_json = json.loads(raw_json) + except (json.JSONDecodeError, TypeError): + return _empty_result("postgres", raw_json, error="EXPLAIN JSON 解析失败") + + # PG 返回的是数组 + if isinstance(raw_json, list) and len(raw_json) > 0: + plan_root = raw_json[0].get("Plan", raw_json[0]) + elif isinstance(raw_json, dict): + plan_root = raw_json.get("Plan", raw_json) + else: + return _empty_result("postgres", raw_json, error="无法识别的 EXPLAIN JSON 结构") + + warnings: List[str] = [] + node_id_counter = [0] + + def next_id() -> int: + node_id_counter[0] += 1 + return node_id_counter[0] + + def parse_pg_plan(plan: Dict) -> Dict[str, Any]: + """递归解析 PostgreSQL Plan 节点。""" + node_id = next_id() + children = [] + + node_type = plan.get("Node Type", "Unknown") + relation_name = plan.get("Relation Name") or plan.get("Table Name") + alias = plan.get("Alias") + startup_cost = plan.get("Startup Cost") + total_cost = plan.get("Total Cost") + plan_rows = plan.get("Plan Rows") + plan_width = plan.get("Plan Width") + index_name = plan.get("Index Name") + index_cond = plan.get("Index Cond") + filter_str = plan.get("Filter") + hash_cond = plan.get("Hash Cond") + join_type = plan.get("Join Type") + sort_key = plan.get("Sort Key") + sort_method = plan.get("Sort Method") + group_key = plan.get("Group Key") + actual_rows = plan.get("Actual Rows") + actual_time = plan.get("Actual Total Time") + + # 操作类型映射 + operation = _pg_node_type_to_operation(node_type) + if join_type: + operation = f"{join_type} {_pg_node_type_to_operation(node_type)}" + + # Extra 信息 + extra_parts = [] + if index_cond: + extra_parts.append(f"索引条件: {index_cond}") + if filter_str: + extra_parts.append(f"过滤条件: {filter_str}") + if hash_cond: + extra_parts.append(f"Hash 条件: {hash_cond}") + if sort_key: + extra_parts.append(f"排序键: {sort_key}") + if sort_method: + extra_parts.append(f"排序方法: {sort_method}") + if group_key: + extra_parts.append(f"分组键: {group_key}") + if actual_rows is not None: + extra_parts.append(f"实际行数: {actual_rows}") + if actual_time is not None: + extra_parts.append(f"实际耗时: {actual_time}ms") + + # 警告检测 + if node_type == "Seq Scan": + warnings.append(f"表 '{relation_name or '?'}' 使用了顺序扫描 (Seq Scan)，建议检查索引") + if node_type == "Sort" and sort_method == "external" or plan.get("Sort Space Used"): + warnings.append(f"排序使用了外部排序 (磁盘)，建议增加 work_mem") + + # 子计划 + sub_plans = plan.get("Plans", []) + for sp in sub_plans: + children.append(parse_pg_plan(sp)) + + # 子查询/SubPlan + subplan_name = plan.get("Subplan Name") + if subplan_name: + extra_parts.append(f"子计划: {subplan_name}") + + kwargs = {} + if alias and alias != relation_name: + kwargs["alias"] = alias + if plan_width: + kwargs["plan_width"] = plan_width + if startup_cost is not None: + kwargs["startup_cost"] = startup_cost + if actual_time is not None: + kwargs["actual_time_ms"] = actual_time + if actual_rows is not None: + kwargs["actual_rows"] = actual_rows + + return make_plan_node( + id=node_id, + operation=operation, + table=relation_name, + rows=int(plan_rows) if plan_rows else None, + cost=float(total_cost) if total_cost is not None else None, + key=index_name, + extra="; ".join(extra_parts) if extra_parts else None, + children=children, + node_type=node_type, + **kwargs, + ) + + root = parse_pg_plan(plan_root) + nodes = [root] + + # 汇总 + summary = _build_summary(nodes, warnings, "postgres") + + return { + "execution_plan": nodes, + "warnings": warnings, + "summary": summary, + "db_type": "postgres", + "raw_result": raw_json, + } + + +def _pg_node_type_to_operation(node_type: str) -> str: + """将 PostgreSQL Node Type 映射为统一操作名。""" + mapping = { + "Seq Scan": "Table Scan", + "Index Scan": "Index Scan", + "Index Only Scan": "Index Only Scan", + "Bitmap Heap Scan": "Bitmap Heap Scan", + "Bitmap Index Scan": "Bitmap Index Scan", + "Tid Scan": "TID Scan", + "Subquery Scan": "Subquery Scan", + "Function Scan": "Function Scan", + "Table Function Scan": "Table Function Scan", + "Values Scan": "Values Scan", + "CTE Scan": "CTE Scan", + "Named Tuplestore Scan": "Named Tuplestore Scan", + "WorkTable Scan": "WorkTable Scan", + "Nested Loop": "Nested Loop Join", + "Merge Join": "Merge Join", + "Hash Join": "Hash Join", + "Sort": "Sort", + "Hash": "Hash", + "Aggregate": "Aggregate", + "Group": "Group", + "Append": "Append", + "Merge Append": "Merge Append", + "Limit": "Limit", + "Unique": "Unique", + "Gather": "Gather", + "Gather Merge": "Gather Merge", + "Materialize": "Materialize", + "Result": "Result", + "ProjectSet": "ProjectSet", + "WindowAgg": "Window Aggregate", + "Recursive Union": "Recursive Union", + "LockRows": "Lock Rows", + "ModifyTable": "Modify Table", + "SetOp": "Set Operation", + } + return mapping.get(node_type, node_type) + + +# ===== SQLite EXPLAIN QUERY PLAN 解析 ===== + +def parse_sqlite_explain(rows: List[Dict[str, Any]]) -> Dict[str, Any]: + """解析 SQLite EXPLAIN QUERY PLAN 的文本/行结果。 + + SQLite 返回格式: + id|parent|notused|detail + 3|0|0|SCAN TABLE users + 或 + id|parent|notused|detail + 2|0|0|SEARCH TABLE users USING INDEX idx_name (name=?) + + Returns: + 统一结构的 Dict + """ + warnings: List[str] = [] + node_id_counter = [0] + node_map: Dict[int, Dict] = {} + + def next_id() -> int: + node_id_counter[0] += 1 + return node_id_counter[0] + + for row in rows: + # SQLite EXPLAIN QUERY PLAN 返回: id, parent, notused, detail + detail = row.get("detail") or row.get("DETAIL") or "" + parent_id = row.get("parent") or row.get("PARENT") or 0 + sqlite_id = row.get("id") or row.get("ID") or next_id() + + table_name = None + operation = "Unknown" + key = None + extra = None + + detail_upper = detail.upper() + + # 解析 detail 文本 + if "SCAN TABLE" in detail_upper: + operation = "Table Scan" + # 提取表名: "SCAN TABLE users" + parts = detail.split("TABLE", 1) + if len(parts) > 1: + table_part = parts[1].strip().split()[0] if parts[1].strip() else None + table_name = table_part + warnings.append(f"表 '{table_name or '?'}' 全表扫描，建议添加索引") + elif "SEARCH TABLE" in detail_upper: + operation = "Index Lookup" + parts = detail.split("TABLE", 1) + if len(parts) > 1: + rest = parts[1].strip() + tokens = rest.split() + table_name = tokens[0] if tokens else None + # 提取索引名: "USING INDEX idx_name (col=?)" + if "USING INDEX" in detail_upper: + idx_parts = detail.split("USING INDEX", 1) + if len(idx_parts) > 1: + idx_rest = idx_parts[1].strip().split("(")[0].strip() + key = idx_rest + elif "USING COVERING INDEX" in detail_upper: + idx_parts = detail.split("USING COVERING INDEX", 1) + if len(idx_parts) > 1: + idx_rest = idx_parts[1].strip().split("(")[0].strip() + key = idx_rest + extra = "Covering Index" + elif "EXECUTE" in detail_upper: + operation = "Execute" + extra = detail + elif "CORRELATED SCALAR SUBQUERY" in detail_upper: + operation = "Correlated Subquery" + extra = detail + elif "CO-ROUTINE" in detail_upper: + operation = "Co-routine" + extra = detail + elif "SCAN" in detail_upper and "TABLE" not in detail_upper: + operation = "Scan" + extra = detail + else: + operation = detail if detail else "Unknown" + + node = make_plan_node( + id=next_id(), + operation=operation, + table=table_name, + key=key, + extra=extra, + ) + node_map[sqlite_id] = node + + # 构建树（基于 parent） + root_nodes = [] + for sqlite_id, node in node_map.items(): + parent = None + for row in rows: + rid = row.get("id") or row.get("ID") + if rid == sqlite_id: + parent = row.get("parent") or row.get("PARENT") or 0 + break + if parent == 0 or parent not in node_map: + root_nodes.append(node) + else: + parent_node = node_map[parent] + parent_node["children"].append(node) + + if not root_nodes and node_map: + root_nodes = list(node_map.values()) + + summary = _build_summary(root_nodes, warnings, "sqlite") + + return { + "execution_plan": root_nodes, + "warnings": warnings, + "summary": summary, + "db_type": "sqlite", + "raw_result": rows, + } + + +# ===== 辅助函数 ===== + +def _empty_result(db_type: str, raw: Any, error: str = "") -> Dict[str, Any]: + """返回空结果（解析失败时）。""" + return { + "execution_plan": [], + "warnings": [error] if error else ["EXPLAIN 结果为空"], + "summary": { + "total_cost": 0, + "estimated_rows": 0, + "has_full_scan": False, + "has_temporary": False, + "has_filesort": False, + "index_usage_ratio": 0.0, + }, + "db_type": db_type, + "raw_result": raw, + } + + +def _build_summary( + nodes: List[Dict], + warnings: List[str], + db_type: str, +) -> Dict[str, Any]: + """从执行计划树构建汇总信息。""" + total_cost = 0.0 + total_rows = 0 + has_full_scan = False + has_temporary = False + has_filesort = False + index_used_count = 0 + total_table_count = 0 + + def walk(node: Dict): + nonlocal total_cost, total_rows, has_full_scan, has_temporary, has_filesort + nonlocal index_used_count, total_table_count + + if node.get("cost") is not None: + total_cost = max(total_cost, float(node["cost"])) + + if node.get("rows") is not None: + total_rows += int(node["rows"]) + + op = (node.get("operation") or "").lower() + extra = str(node.get("extra") or "").lower() + + if "table scan" in op or "seq scan" in op or op == "system scan": + has_full_scan = True + total_table_count += 1 + elif "index" in op or "lookup" in op: + index_used_count += 1 + total_table_count += 1 + elif "scan" in op: + total_table_count += 1 + + if "temporary" in op or "temporary" in extra: + has_temporary = True + if "filesort" in op or "sort" in op or "filesort" in extra: + has_filesort = True + + for child in node.get("children", []): + walk(child) + + for n in nodes: + walk(n) + + index_usage_ratio = 0.0 + if total_table_count > 0: + index_usage_ratio = round(index_used_count / total_table_count, 2) + + return { + "total_cost": round(total_cost, 4), + "estimated_rows": total_rows, + "has_full_scan": has_full_scan, + "has_temporary": has_temporary, + "has_filesort": has_filesort, + "index_usage_ratio": index_usage_ratio, + } + + +# ===== EXPLAIN 执行入口 ===== + +class ExplainService: + """EXPLAIN 执行服务 — 统一入口。""" + + @staticmethod + async def execute_explain( + instance: Instance, + database: Optional[str], + query: str, + ) -> Dict[str, Any]: + """根据 db_type 构造并执行 EXPLAIN，返回统一树结构。 + + Args: + instance: 数据源实例 + database: 目标数据库名 + query: 要解释的 SQL 查询 + + Returns: + 统一 EXPLAIN 结果 Dict + """ + db_type = instance.db_type + + # 验证查询以 SELECT 或 WITH 开头 + stripped = query.strip() + upper = stripped.upper() + if not upper.startswith(("SELECT", "WITH")): + raise ValueError("EXPLAIN 仅支持 SELECT 或 WITH 查询") + + if db_type == "mysql": + return await ExplainService._explain_mysql(instance, database, query) + elif db_type in ("postgres", "postgresql"): + return await ExplainService._explain_postgres(instance, database, query) + elif db_type == "sqlite": + return await ExplainService._explain_sqlite(instance, query) + else: + raise ValueError(f"暂不支持 {db_type} 类型的 EXPLAIN") + + @staticmethod + async def _explain_mysql( + instance: Instance, database: Optional[str], query: str + ) -> Dict[str, Any]: + """执行 MySQL EXPLAIN FORMAT=JSON。""" + explain_sql = f"EXPLAIN FORMAT=JSON {query}" + + conn = await aiomysql.connect( + host=instance.host, + port=instance.port, + user=instance.username, + password=instance.password, + db=database or instance.database, + ) + try: + async with conn.cursor(aiomysql.DictCursor) as cur: + await asyncio.wait_for(cur.execute(explain_sql), timeout=30) + rows = await cur.fetchall() + # MySQL EXPLAIN FORMAT=JSON 返回一行一列 JSON + if rows: + raw = rows[0].get("EXPLAIN") or rows[0].get("JSON") or rows[0] + # raw 可能是字符串或已经解析的 dict + if isinstance(raw, str): + raw = json.loads(raw) + else: + raw = {} + except asyncio.TimeoutError: + raise ValueError("EXPLAIN 执行超时（30秒）") + finally: + conn.close() + + return parse_mysql_explain_json(raw) + + @staticmethod + async def _explain_postgres( + instance: Instance, database: Optional[str], query: str + ) -> Dict[str, Any]: + """执行 PostgreSQL EXPLAIN (FORMAT JSON)。""" + explain_sql = f"EXPLAIN (FORMAT JSON, ANALYZE false) {query}" + + dsn = ( + f"postgresql://{instance.username}:{instance.password}" + f"@{instance.host}:{instance.port}/{database or instance.database or 'postgres'}" + ) + conn = await asyncpg.connect(dsn) + try: + result = await asyncio.wait_for(conn.fetchval(explain_sql), timeout=30) + # asyncpg fetchval 返回 JSON 字符串 + if isinstance(result, str): + raw = json.loads(result) + elif isinstance(result, list): + raw = result + else: + raw = result + except asyncio.TimeoutError: + raise ValueError("EXPLAIN 执行超时（30秒）") + finally: + await conn.close() + + return parse_postgres_explain_json(raw) + + @staticmethod + async def _explain_sqlite(instance: Instance, query: str) -> Dict[str, Any]: + """执行 SQLite EXPLAIN QUERY PLAN。""" + import sqlite3 + + explain_sql = f"EXPLAIN QUERY PLAN {query}" + db_path = instance.host + + def _exec(): + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + cur = conn.execute(explain_sql) + rows = [dict(r) for r in cur.fetchall()] + conn.close() + return rows + + try: + rows = await asyncio.wait_for( + asyncio.get_event_loop().run_in_executor(None, _exec), + timeout=30, + ) + except asyncio.TimeoutError: + raise ValueError("EXPLAIN 执行超时（30秒）") + + return parse_sqlite_explain(rows) diff --git a/backend/app/services/query_history_service.py b/backend/app/services/query_history_service.py index 3ea882a..db0b0b5 100644 --- a/backend/app/services/query_history_service.py +++ b/backend/app/services/query_history_service.py @@ -1,4 +1,4 @@ -"""查询历史服务 — 记录和管理用户查询历史。 +"""查询历史服务 — 记录和管理用户查询历史（SQLite 持久化）。 支持: - 自动记录每次查询（内容、耗时、结果数、数据库类型） @@ -13,65 +13,27 @@ - 快捷查询模板，减少重复输入 - 查询统计帮助用户了解自己的数据使用习惯 """ +import re from typing import List, Optional, Dict, Any -from dataclasses import dataclass, field from datetime import datetime, timedelta from collections import Counter +from sqlalchemy import select, func, and_, or_ +from sqlalchemy.ext.asyncio import AsyncSession -@dataclass -class QueryHistoryEntry: - """查询历史条目。""" - id: int = 0 - user_id: int = 0 - username: str = "" - instance_id: int = 0 - instance_name: str = "" - db_type: str = "" # mysql / postgres / redis / mongo - database_name: str = "" - query_content: str = "" # SQL / Redis命令 / MongoDB查询 - query_type: str = "" # SELECT / GET / find 等 - execution_time_ms: int = 0 - result_rows: int = 0 - is_success: bool = True - error_message: Optional[str] = None - is_favorite: bool = False - note: str = "" # 用户备注 - created_at: str = "" - - -@dataclass -class QuickQuery: - """快捷查询模板。""" - id: int = 0 - name: str = "" # 查询名称，如"查看活跃用户数" - description: str = "" # 描述 - query_content: str = "" # 查询内容 - db_type: str = "" # 适用的数据库类型 - database_pattern: str = "" # 适用的数据库名称模式 - is_shared: bool = False # 是否团队共享 - created_by: int = 0 - created_by_name: str = "" - usage_count: int = 0 # 使用次数 - created_at: str = "" +from app.models.query_history import QueryHistory, QuickQueryModel class QueryHistoryService: - """查询历史服务。 + """查询历史服务（SQLite 持久化）。""" - 内存存储实现，生产环境应替换为数据库存储。 - """ - - _history: List[QueryHistoryEntry] = [] - _favorites: set = set() # (user_id, history_id) - _quick_queries: List[QuickQuery] = [] - _history_counter: int = 0 - _quick_counter: int = 0 _max_history_per_user: int = 500 # 每个用户最多保留500条 - @classmethod + # ── 查询历史 ── + + @staticmethod async def record_query( - cls, + db: AsyncSession, user_id: int, username: str, instance_id: int, @@ -84,15 +46,12 @@ class QueryHistoryService: result_rows: int, is_success: bool = True, error_message: Optional[str] = None, - ) -> QueryHistoryEntry: + ) -> QueryHistory: """记录一次查询。 每次查询自动调用，无需用户干预。 """ - cls._history_counter += 1 - - entry = QueryHistoryEntry( - id=cls._history_counter, + entry = QueryHistory( user_id=user_id, username=username, instance_id=instance_id, @@ -105,24 +64,39 @@ class QueryHistoryService: result_rows=result_rows, is_success=is_success, error_message=error_message, - created_at=datetime.now().isoformat(), ) - - cls._history.append(entry) + db.add(entry) + await db.flush() # 限制每个用户的历史记录数量 - user_entries = [h for h in cls._history if h.user_id == user_id] - if len(user_entries) > cls._max_history_per_user: - # 删除最旧的 - to_remove = len(user_entries) - cls._max_history_per_user - oldest_ids = {h.id for h in sorted(user_entries, key=lambda x: x.created_at)[:to_remove]} - cls._history = [h for h in cls._history if h.id not in oldest_ids] + count_stmt = select(func.count()).select_from(QueryHistory).where( + QueryHistory.user_id == user_id + ) + total = (await db.execute(count_stmt)).scalar() + if total > QueryHistoryService._max_history_per_user: + # 删除最旧的条目 + to_remove = total - QueryHistoryService._max_history_per_user + old_stmt = ( + select(QueryHistory.id) + .where(QueryHistory.user_id == user_id) + .order_by(QueryHistory.created_at.asc()) + .limit(to_remove) + ) + old_result = await db.execute(old_stmt) + old_ids = [row[0] for row in old_result.all()] + if old_ids: + delete_stmt = QueryHistory.__table__.delete().where( + QueryHistory.id.in_(old_ids) + ) + await db.execute(delete_stmt) + + await db.flush() return entry - @classmethod + @staticmethod async def get_history( - cls, + db: AsyncSession, user_id: int, db_type: Optional[str] = None, instance_id: Optional[int] = None, @@ -131,22 +105,24 @@ class QueryHistoryService: page_size: int = 20, ) -> Dict[str, Any]: """获取用户查询历史。""" - entries = [h for h in cls._history if h.user_id == user_id] + stmt = select(QueryHistory).where(QueryHistory.user_id == user_id) if db_type: - entries = [h for h in entries if h.db_type == db_type] + stmt = stmt.where(QueryHistory.db_type == db_type) if instance_id: - entries = [h for h in entries if h.instance_id == instance_id] + stmt = stmt.where(QueryHistory.instance_id == instance_id) if keyword: - kw = keyword.lower() - entries = [h for h in entries if kw in h.query_content.lower()] + stmt = stmt.where(QueryHistory.query_content.ilike(f"%{keyword}%")) - total = len(entries) - entries.sort(key=lambda x: x.created_at, reverse=True) + # 总数 + total_stmt = select(func.count()).select_from(stmt.subquery()) + total = (await db.execute(total_stmt)).scalar() - start = (page - 1) * page_size - end = start + page_size - page_entries = entries[start:end] + # 排序 + 分页 + stmt = stmt.order_by(QueryHistory.created_at.desc()) + stmt = stmt.offset((page - 1) * page_size).limit(page_size) + result = await db.execute(stmt) + entries = result.scalars().all() return { "total": total, @@ -158,42 +134,60 @@ class QueryHistoryService: "instance_name": e.instance_name, "db_type": e.db_type, "database_name": e.database_name, - "query_content": e.query_content[:200], + "query_content": (e.query_content or "")[:200], "query_type": e.query_type, "execution_time_ms": e.execution_time_ms, "result_rows": e.result_rows, "is_success": e.is_success, - "is_favorite": (user_id, e.id) in cls._favorites, + "is_favorite": e.is_favorite, "note": e.note, - "created_at": e.created_at, + "created_at": e.created_at.isoformat() if e.created_at else "", } - for e in page_entries + for e in entries ], } - @classmethod - async def toggle_favorite(cls, user_id: int, history_id: int) -> bool: + @staticmethod + async def toggle_favorite(db: AsyncSession, user_id: int, history_id: int) -> bool: """切换收藏状态。""" - key = (user_id, history_id) - if key in cls._favorites: - cls._favorites.remove(key) - return False # 已取消收藏 - else: - cls._favorites.add(key) - return True # 已收藏 + stmt = select(QueryHistory).where( + and_( + QueryHistory.id == history_id, + QueryHistory.user_id == user_id, + ) + ) + result = await db.execute(stmt) + entry = result.scalar_one_or_none() - @classmethod + if entry is None: + # 没有找到该条记录（可能不属于当前用户），直接返回 False + return False + + entry.is_favorite = not entry.is_favorite + await db.flush() + return entry.is_favorite + + @staticmethod async def get_favorites( - cls, user_id: int, page: int = 1, page_size: int = 20 + db: AsyncSession, user_id: int, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """获取收藏的查询。""" - favorite_ids = {hid for uid, hid in cls._favorites if uid == user_id} - entries = [h for h in cls._history if h.id in favorite_ids] - entries.sort(key=lambda x: x.created_at, reverse=True) + stmt = select(QueryHistory).where( + and_( + QueryHistory.user_id == user_id, + QueryHistory.is_favorite == True, # noqa: E712 + ) + ) - total = len(entries) - start = (page - 1) * page_size - end = start + page_size + # 总数 + total_stmt = select(func.count()).select_from(stmt.subquery()) + total = (await db.execute(total_stmt)).scalar() + + # 排序 + 分页 + stmt = stmt.order_by(QueryHistory.created_at.desc()) + stmt = stmt.offset((page - 1) * page_size).limit(page_size) + result = await db.execute(stmt) + entries = result.scalars().all() return { "total": total, @@ -205,21 +199,21 @@ class QueryHistoryService: "instance_name": e.instance_name, "db_type": e.db_type, "database_name": e.database_name, - "query_content": e.query_content[:200], + "query_content": (e.query_content or "")[:200], "query_type": e.query_type, "execution_time_ms": e.execution_time_ms, "result_rows": e.result_rows, - "created_at": e.created_at, + "created_at": e.created_at.isoformat() if e.created_at else "", } - for e in entries[start:end] + for e in entries ], } # ── 快捷查询 ── - @classmethod + @staticmethod async def create_quick_query( - cls, + db: AsyncSession, name: str, description: str, query_content: str, @@ -228,11 +222,9 @@ class QueryHistoryService: is_shared: bool, created_by: int, created_by_name: str, - ) -> QuickQuery: + ) -> QuickQueryModel: """创建快捷查询。""" - cls._quick_counter += 1 - qq = QuickQuery( - id=cls._quick_counter, + qq = QuickQueryModel( name=name, description=description, query_content=query_content, @@ -241,71 +233,102 @@ class QueryHistoryService: is_shared=is_shared, created_by=created_by, created_by_name=created_by_name, - created_at=datetime.now().isoformat(), ) - cls._quick_queries.append(qq) + db.add(qq) + await db.flush() return qq - @classmethod + @staticmethod async def get_quick_queries( - cls, + db: AsyncSession, user_id: int, db_type: Optional[str] = None, keyword: Optional[str] = None, ) -> List[Dict[str, Any]]: """获取快捷查询列表（个人+团队共享）。""" - result = [] - for qq in cls._quick_queries: - # 条件：共享的 或 自己创建的 - if qq.is_shared or qq.created_by == user_id: - if db_type and qq.db_type != db_type: - continue - if keyword and keyword.lower() not in qq.name.lower(): - continue - result.append({ - "id": qq.id, - "name": qq.name, - "description": qq.description, - "query_content": qq.query_content[:500], - "db_type": qq.db_type, - "is_shared": qq.is_shared, - "created_by_name": qq.created_by_name, - "usage_count": qq.usage_count, - "created_at": qq.created_at, - }) - return result + stmt = select(QuickQueryModel).where( + or_( + QuickQueryModel.is_shared == True, # noqa: E712 + QuickQueryModel.created_by == user_id, + ) + ) - @classmethod - async def use_quick_query(cls, quick_query_id: int) -> Optional[str]: + if db_type: + stmt = stmt.where(QuickQueryModel.db_type == db_type) + if keyword: + stmt = stmt.where(QuickQueryModel.name.ilike(f"%{keyword}%")) + + stmt = stmt.order_by(QuickQueryModel.created_at.desc()) + result = await db.execute(stmt) + queries = result.scalars().all() + + return [ + { + "id": qq.id, + "name": qq.name, + "description": qq.description, + "query_content": (qq.query_content or "")[:500], + "db_type": qq.db_type, + "is_shared": qq.is_shared, + "created_by_name": qq.created_by_name, + "usage_count": qq.usage_count, + "created_at": qq.created_at.isoformat() if qq.created_at else "", + } + for qq in queries + ] + + @staticmethod + async def use_quick_query(db: AsyncSession, quick_query_id: int) -> Optional[str]: """使用快捷查询，返回查询内容并增加使用次数。""" - for qq in cls._quick_queries: - if qq.id == quick_query_id: - qq.usage_count += 1 - return qq.query_content - return None + stmt = select(QuickQueryModel).where(QuickQueryModel.id == quick_query_id) + result = await db.execute(stmt) + qq = result.scalar_one_or_none() + if qq is None: + return None + qq.usage_count += 1 + await db.flush() + return qq.query_content - @classmethod - async def get_query_statistics(cls, user_id: int) -> Dict[str, Any]: + @staticmethod + async def get_query_statistics(db: AsyncSession, user_id: int) -> Dict[str, Any]: """获取用户查询统计。 人性化：帮助用户了解自己的数据使用习惯。 """ - entries = [h for h in cls._history if h.user_id == user_id] + # 总数 + count_stmt = select(func.count()).select_from(QueryHistory).where( + QueryHistory.user_id == user_id + ) + total_queries = (await db.execute(count_stmt)).scalar() - if not entries: + if not total_queries: return {"message": "暂无查询记录"} # 按数据库类型分布 - db_type_dist = Counter(h.db_type for h in entries) + db_type_stmt = ( + select(QueryHistory.db_type, func.count().label("cnt")) + .where(QueryHistory.user_id == user_id) + .group_by(QueryHistory.db_type) + ) + db_type_result = await db.execute(db_type_stmt) + db_type_dist = {row[0]: row[1] for row in db_type_result.all()} # 高频查询表（从SQL中提取表名） + sql_stmt = ( + select(QueryHistory.query_content) + .where( + and_( + QueryHistory.user_id == user_id, + QueryHistory.db_type.in_(["mysql", "postgres"]), + ) + ) + ) + sql_result = await db.execute(sql_stmt) tables = [] - for h in entries: - if h.db_type in ("mysql", "postgres"): - # 简单提取表名 - import re - found = re.findall(r"FROM\s+(\w+)", h.query_content, re.IGNORECASE) - tables.extend(found) + for row in sql_result.all(): + content = row[0] or "" + found = re.findall(r"FROM\s+(\w+)", content, re.IGNORECASE) + tables.extend(found) table_dist = Counter(tables).most_common(10) # 查询趋势（最近7天） @@ -313,58 +336,77 @@ class QueryHistoryService: daily_counts = {} for i in range(7): day = (now - timedelta(days=i)).strftime("%Y-%m-%d") - count = sum( - 1 for h in entries - if h.created_at.startswith(day) + day_start = datetime.strptime(day, "%Y-%m-%d") + day_end = day_start + timedelta(days=1) + day_count_stmt = select(func.count()).select_from(QueryHistory).where( + and_( + QueryHistory.user_id == user_id, + QueryHistory.created_at >= day_start, + QueryHistory.created_at < day_end, + ) ) - daily_counts[day] = count + day_count = (await db.execute(day_count_stmt)).scalar() + daily_counts[day] = day_count # 平均执行时间 - avg_time = sum(h.execution_time_ms for h in entries) / len(entries) + avg_stmt = select(func.avg(QueryHistory.execution_time_ms)).where( + QueryHistory.user_id == user_id + ) + avg_time = (await db.execute(avg_stmt)).scalar() or 0 - # 总查询次数 - total_queries = len(entries) + # 成功率 + success_stmt = select(func.count()).select_from(QueryHistory).where( + and_( + QueryHistory.user_id == user_id, + QueryHistory.is_success == True, # noqa: E712 + ) + ) + success_count = (await db.execute(success_stmt)).scalar() return { "total_queries": total_queries, - "avg_execution_time_ms": round(avg_time, 2), - "db_type_distribution": dict(db_type_dist), + "avg_execution_time_ms": round(float(avg_time), 2), + "db_type_distribution": db_type_dist, "top_tables": [{"table": t, "count": c} for t, c in table_dist], "daily_trend": daily_counts, - "success_rate": round( - sum(1 for h in entries if h.is_success) / total_queries * 100, 2 - ), + "success_rate": round(success_count / total_queries * 100, 2), } # ── 预置快捷查询模板 ── - @classmethod - def seed_default_quick_queries(cls): - """预置默认快捷查询模板。""" + @staticmethod + async def seed_default_quick_queries(db: AsyncSession): + """预置默认快捷查询模板（仅在表为空时插入）。""" + # 检查是否已有数据 + count_stmt = select(func.count()).select_from(QuickQueryModel) + count = (await db.execute(count_stmt)).scalar() + if count and count > 0: + return + defaults = [ - QuickQuery( - id=-1, name="查看所有表", description="列出数据库中所有表", + QuickQueryModel( + name="查看所有表", description="列出数据库中所有表", query_content="SHOW TABLES", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), - QuickQuery( - id=-2, name="查看表结构", description="查看指定表的列信息", + QuickQueryModel( + name="查看表结构", description="查看指定表的列信息", query_content="DESCRIBE {table_name}", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), - QuickQuery( - id=-3, name="查看慢查询", description="MySQL慢查询日志", + QuickQueryModel( + name="查看慢查询", description="MySQL慢查询日志", query_content="SELECT * FROM mysql.slow_log ORDER BY start_time DESC LIMIT 20", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), - QuickQuery( - id=-4, name="查看活跃连接", description="当前数据库连接情况", + QuickQueryModel( + name="查看活跃连接", description="当前数据库连接情况", query_content="SHOW PROCESSLIST", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), - QuickQuery( - id=-5, name="查看表大小", description="各表占用空间统计", + QuickQueryModel( + name="查看表大小", description="各表占用空间统计", query_content="""SELECT table_name AS `表名`, ROUND(data_length / 1024 / 1024, 2) AS `数据大小(MB)`, @@ -376,38 +418,34 @@ ORDER BY data_length DESC LIMIT 50""", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), - QuickQuery( - id=-6, name="Redis查看所有键", description="SCAN遍历所有键", + QuickQueryModel( + name="Redis查看所有键", description="SCAN遍历所有键", query_content="SCAN 0 MATCH * COUNT 50", db_type="redis", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), - QuickQuery( - id=-7, name="Redis查看服务器信息", description="Redis INFO命令", + QuickQueryModel( + name="Redis查看服务器信息", description="Redis INFO命令", query_content="INFO server", db_type="redis", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), - QuickQuery( - id=-8, name="PG查看所有表", description="PostgreSQL列出所有表", + QuickQueryModel( + name="PG查看所有表", description="PostgreSQL列出所有表", query_content="SELECT schemaname, tablename, n_tup_ins, n_tup_upd, n_tup_del FROM pg_stat_user_tables ORDER BY n_live_tup DESC LIMIT 50", db_type="postgres", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), - QuickQuery( - id=-9, name="MongoDB查找文档", description="查找集合中的文档", + QuickQueryModel( + name="MongoDB查找文档", description="查找集合中的文档", query_content='{"collection": "{collection_name}", "operation": "find", "limit": 20}', db_type="mongo", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), - QuickQuery( - id=-10, name="MongoDB统计文档数", description="统计集合文档数量", + QuickQueryModel( + name="MongoDB统计文档数", description="统计集合文档数量", query_content='{"collection": "{collection_name}", "operation": "countDocuments"}', db_type="mongo", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), ] - cls._quick_queries.extend(defaults) - cls._quick_counter = max(cls._quick_counter, 10) - - -# 初始化预置快捷查询 -QueryHistoryService.seed_default_quick_queries() + db.add_all(defaults) + await db.flush() diff --git a/backend/tests/test_explain.py b/backend/tests/test_explain.py new file mode 100644 index 0000000..20fcaa2 --- /dev/null +++ b/backend/tests/test_explain.py @@ -0,0 +1,994 @@ +"""EXPLAIN 服务解析测试 — 测试 MySQL / PostgreSQL / SQLite 解析逻辑 + API endpoint。 + +使用 mock 的数据库连接，不依赖真实数据库。 +""" +import json +import asyncio +import sys +from unittest.mock import patch, AsyncMock, MagicMock + +import pytest +from httpx import AsyncClient + +from app.core.crypto import encrypt_password +from app.models.instance import Instance +from app.services.explain_service import ( + ExplainService, + parse_mysql_explain_json, + parse_postgres_explain_json, + parse_sqlite_explain, + make_plan_node, + _empty_result, +) +from tests.conftest import auth_header + + +# =========================================================================== +# Part 1: MySQL EXPLAIN JSON 解析测试 +# =========================================================================== + + +class TestMySQLExplainParsing: + """测试 MySQL EXPLAIN FORMAT=JSON 结果解析。""" + + def test_simple_table_scan(self): + """测试简单全表扫描。""" + raw = { + "query_block": { + "select_id": 1, + "cost_info": {"query_cost": "0.35"}, + "table": { + "table_name": "users", + "access_type": "ALL", + "rows_examined_per_scan": 1000, + "rows_produced_per_join": 1000, + "filtered": "100.00", + "cost_info": {"read_cost": "0.25"}, + }, + } + } + + result = parse_mysql_explain_json(raw) + + assert result["db_type"] == "mysql" + assert len(result["execution_plan"]) == 1 + + root = result["execution_plan"][0] + assert root["operation"] == "Table Scan" + assert root["table"] == "users" + assert root["rows"] == 1000 + assert root["cost"] == 0.25 + + # 警告检测 + assert any("全表扫描" in w for w in result["warnings"]) + + # 汇总 + summary = result["summary"] + assert summary["has_full_scan"] is True + assert summary["estimated_rows"] == 1000 + assert summary["total_cost"] == 0.25 + + def test_index_lookup(self): + """测试索引查找。""" + raw = { + "query_block": { + "select_id": 1, + "cost_info": {"query_cost": "0.10"}, + "table": { + "table_name": "orders", + "access_type": "ref", + "possible_keys": "idx_user_id", + "key": "idx_user_id", + "used_key_parts": ["user_id"], + "rows_examined_per_scan": 10, + "filtered": "100.00", + }, + } + } + + result = parse_mysql_explain_json(raw) + + root = result["execution_plan"][0] + assert root["operation"] == "Index Lookup" + assert root["table"] == "orders" + assert root["key"] == "idx_user_id" + assert root["rows"] == 10 + + summary = result["summary"] + assert summary["has_full_scan"] is False + assert summary["index_usage_ratio"] == 1.0 # 1 table, 1 index usage + + def test_nested_loop_join(self): + """测试嵌套循环 JOIN。""" + raw = { + "query_block": { + "select_id": 1, + "cost_info": {"query_cost": "1.50"}, + "nested_loop": [ + { + "table": { + "table_name": "users", + "access_type": "ALL", + "rows_examined_per_scan": 100, + } + }, + { + "table": { + "table_name": "orders", + "access_type": "ref", + "key": "idx_user_id", + "rows_examined_per_scan": 5, + } + }, + ], + } + } + + result = parse_mysql_explain_json(raw) + + root = result["execution_plan"][0] + assert root["operation"] == "Query Block" + assert len(root["children"]) == 2 + + child1 = root["children"][0] + assert child1["operation"] == "Table Scan" + assert child1["table"] == "users" + + child2 = root["children"][1] + assert child2["operation"] == "Index Lookup" + assert child2["table"] == "orders" + assert child2["key"] == "idx_user_id" + + def test_const_and_eq_ref(self): + """测试 const / eq_ref 类型映射。""" + # 直接测试 _mysql_access_type_to_operation + from app.services.explain_service import _mysql_access_type_to_operation + assert _mysql_access_type_to_operation("const") == "Const Row" + assert _mysql_access_type_to_operation("eq_ref") == "Unique Lookup" + assert _mysql_access_type_to_operation("range") == "Index Range Scan" + assert _mysql_access_type_to_operation("ALL") == "Table Scan" + assert _mysql_access_type_to_operation("index") == "Full Index Scan" + + def test_json_string_input(self): + """测试字符串 JSON 输入。""" + raw_str = json.dumps({ + "query_block": { + "select_id": 1, + "table": { + "table_name": "t", + "access_type": "ALL", + "rows_examined_per_scan": 50, + }, + } + }) + + result = parse_mysql_explain_json(raw_str) + assert result["db_type"] == "mysql" + assert result["execution_plan"][0]["table"] == "t" + + def test_invalid_json_string(self): + """测试无效 JSON 字符串。""" + result = parse_mysql_explain_json("not valid json{{{") + assert result["db_type"] == "mysql" + assert len(result["warnings"]) > 0 + assert any("解析失败" in w for w in result["warnings"]) + + def test_empty_result(self): + """测试空结果。""" + result = parse_mysql_explain_json({}) + assert result["db_type"] == "mysql" + assert result["execution_plan"] == [] + + def test_total_cost_uses_max(self): + """测试 total_cost 取最大值。""" + raw = { + "query_block": { + "select_id": 1, + "cost_info": {"query_cost": "10.5"}, + "nested_loop": [ + { + "table": { + "table_name": "a", + "access_type": "ALL", + "rows_examined_per_scan": 100, + "cost_info": {"read_cost": "5.0"}, + } + }, + { + "table": { + "table_name": "b", + "access_type": "ref", + "rows_examined_per_scan": 10, + "cost_info": {"read_cost": "3.0"}, + } + }, + ], + } + } + + result = parse_mysql_explain_json(raw) + # total_cost should be the max of all costs + assert result["summary"]["total_cost"] == 10.5 + + +# =========================================================================== +# Part 2: PostgreSQL EXPLAIN JSON 解析测试 +# =========================================================================== + + +class TestPostgreSQLExplainParsing: + """测试 PostgreSQL EXPLAIN (FORMAT JSON) 结果解析。""" + + def test_seq_scan(self): + """测试顺序扫描。""" + raw = [ + { + "Plan": { + "Node Type": "Seq Scan", + "Relation Name": "users", + "Alias": "users", + "Startup Cost": 0.00, + "Total Cost": 15.50, + "Plan Rows": 1000, + "Plan Width": 68, + } + } + ] + + result = parse_postgres_explain_json(raw) + + assert result["db_type"] == "postgres" + assert len(result["execution_plan"]) == 1 + + root = result["execution_plan"][0] + assert root["operation"] == "Table Scan" + assert root["table"] == "users" + assert root["rows"] == 1000 + assert root["cost"] == 15.50 + + assert any("顺序扫描" in w for w in result["warnings"]) + assert result["summary"]["has_full_scan"] is True + + def test_index_scan(self): + """测试索引扫描。""" + raw = [ + { + "Plan": { + "Node Type": "Index Scan", + "Relation Name": "orders", + "Alias": "orders", + "Index Name": "idx_user_id", + "Startup Cost": 0.15, + "Total Cost": 8.50, + "Plan Rows": 10, + "Plan Width": 48, + "Index Cond": "(user_id = 42)", + } + } + ] + + result = parse_postgres_explain_json(raw) + + root = result["execution_plan"][0] + assert root["operation"] == "Index Scan" + assert root["table"] == "orders" + assert root["key"] == "idx_user_id" + assert root["rows"] == 10 + assert "索引条件" in (root["extra"] or "") + + def test_join_with_children(self): + """测试 JOIN 嵌套结构。""" + raw = [ + { + "Plan": { + "Node Type": "Nested Loop", + "Startup Cost": 0.15, + "Total Cost": 25.50, + "Plan Rows": 100, + "Plan Width": 120, + "Plans": [ + { + "Node Type": "Seq Scan", + "Relation Name": "users", + "Total Cost": 12.00, + "Plan Rows": 100, + }, + { + "Node Type": "Index Scan", + "Relation Name": "orders", + "Index Name": "idx_user_id", + "Total Cost": 0.30, + "Plan Rows": 1, + "Index Cond": "(user_id = users.id)", + }, + ], + } + } + ] + + result = parse_postgres_explain_json(raw) + + root = result["execution_plan"][0] + assert root["operation"] == "Nested Loop Join" + assert len(root["children"]) == 2 + + assert root["children"][0]["operation"] == "Table Scan" + assert root["children"][0]["table"] == "users" + + assert root["children"][1]["operation"] == "Index Scan" + assert root["children"][1]["table"] == "orders" + assert root["children"][1]["key"] == "idx_user_id" + + def test_hash_join(self): + """测试 Hash Join。""" + raw = [ + { + "Plan": { + "Node Type": "Hash Join", + "Join Type": "Inner", + "Relation Name": None, + "Hash Cond": "users.id = orders.user_id", + "Total Cost": 50.00, + "Plan Rows": 500, + "Plans": [ + { + "Node Type": "Seq Scan", + "Relation Name": "users", + "Total Cost": 15.00, + "Plan Rows": 1000, + }, + { + "Node Type": "Hash", + "Total Cost": 20.00, + "Plan Rows": 500, + "Plans": [ + { + "Node Type": "Seq Scan", + "Relation Name": "orders", + "Total Cost": 18.00, + "Plan Rows": 500, + } + ], + }, + ], + } + } + ] + + result = parse_postgres_explain_json(raw) + root = result["execution_plan"][0] + assert "Inner" in root["operation"] + assert "Hash Join" in root["operation"] + assert len(root["children"]) == 2 + # Hash has a nested child + hash_child = root["children"][1] + assert hash_child["operation"] == "Hash" + assert len(hash_child["children"]) == 1 + assert hash_child["children"][0]["table"] == "orders" + + def test_sort_and_aggregate(self): + """测试排序和聚合。""" + raw = [ + { + "Plan": { + "Node Type": "Aggregate", + "Total Cost": 30.00, + "Plan Rows": 5, + "Group Key": ["department"], + "Plans": [ + { + "Node Type": "Sort", + "Total Cost": 28.00, + "Plan Rows": 1000, + "Sort Key": ["salary DESC"], + "Sort Method": "quicksort", + "Plans": [ + { + "Node Type": "Seq Scan", + "Relation Name": "employees", + "Total Cost": 15.00, + "Plan Rows": 1000, + } + ], + } + ], + } + } + ] + + result = parse_postgres_explain_json(raw) + root = result["execution_plan"][0] + assert root["operation"] == "Aggregate" + assert "分组键" in (root["extra"] or "") + + sort_child = root["children"][0] + assert sort_child["operation"] == "Sort" + assert "排序键" in (sort_child["extra"] or "") + + seq_child = sort_child["children"][0] + assert seq_child["operation"] == "Table Scan" + + def test_json_string_input(self): + """测试 JSON 字符串输入。""" + raw_str = json.dumps([ + {"Plan": {"Node Type": "Seq Scan", "Relation Name": "t", "Total Cost": 5.0, "Plan Rows": 10}} + ]) + result = parse_postgres_explain_json(raw_str) + assert result["db_type"] == "postgres" + assert result["execution_plan"][0]["table"] == "t" + + def test_invalid_json_string(self): + """测试无效 JSON 字符串。""" + result = parse_postgres_explain_json("invalid{{{") + assert any("解析失败" in w for w in result["warnings"]) + + def test_pg_node_type_mapping(self): + """测试 PG 节点类型映射。""" + from app.services.explain_service import _pg_node_type_to_operation + assert _pg_node_type_to_operation("Seq Scan") == "Table Scan" + assert _pg_node_type_to_operation("Index Scan") == "Index Scan" + assert _pg_node_type_to_operation("Index Only Scan") == "Index Only Scan" + assert _pg_node_type_to_operation("Hash Join") == "Hash Join" + assert _pg_node_type_to_operation("Sort") == "Sort" + assert _pg_node_type_to_operation("Aggregate") == "Aggregate" + assert _pg_node_type_to_operation("Limit") == "Limit" + assert _pg_node_type_to_operation("UnknownType") == "UnknownType" + + def test_actual_rows_and_time(self): + """测试 ANALYZE 包含的实际行数和耗时。""" + raw = [ + { + "Plan": { + "Node Type": "Seq Scan", + "Relation Name": "big_table", + "Total Cost": 100.0, + "Plan Rows": 50000, + "Actual Rows": 49998, + "Actual Total Time": 23.456, + } + } + ] + result = parse_postgres_explain_json(raw) + root = result["execution_plan"][0] + assert root.get("actual_rows") == 49998 + assert root.get("actual_time_ms") == 23.456 + assert "实际行数" in (root["extra"] or "") + assert "实际耗时" in (root["extra"] or "") + + +# =========================================================================== +# Part 3: SQLite EXPLAIN QUERY PLAN 解析测试 +# =========================================================================== + + +class TestSQLiteExplainParsing: + """测试 SQLite EXPLAIN QUERY PLAN 结果解析。""" + + def test_table_scan(self): + """测试全表扫描。""" + rows = [ + {"id": 3, "parent": 0, "notused": 0, "detail": "SCAN TABLE users"}, + ] + result = parse_sqlite_explain(rows) + + assert result["db_type"] == "sqlite" + assert len(result["execution_plan"]) == 1 + + root = result["execution_plan"][0] + assert root["operation"] == "Table Scan" + assert root["table"] == "users" + assert any("全表扫描" in w for w in result["warnings"]) + + def test_index_search(self): + """测试索引搜索。""" + rows = [ + {"id": 2, "parent": 0, "notused": 0, "detail": "SEARCH TABLE users USING INDEX idx_name (name=?)"}, + ] + result = parse_sqlite_explain(rows) + + root = result["execution_plan"][0] + assert root["operation"] == "Index Lookup" + assert root["table"] == "users" + assert root["key"] == "idx_name" + + def test_covering_index(self): + """测试覆盖索引。""" + rows = [ + {"id": 2, "parent": 0, "notused": 0, "detail": "SEARCH TABLE users USING COVERING INDEX idx_name (name=?)"}, + ] + result = parse_sqlite_explain(rows) + + root = result["execution_plan"][0] + assert root["key"] == "idx_name" + assert "Covering Index" in (root["extra"] or "") + + def test_subquery(self): + """测试子查询。""" + rows = [ + {"id": 3, "parent": 0, "notused": 0, "detail": "SCAN TABLE users"}, + {"id": 6, "parent": 0, "notused": 0, "detail": "CORRELATED SCALAR SUBQUERY 2"}, + ] + result = parse_sqlite_explain(rows) + assert len(result["execution_plan"]) == 2 + + def test_empty_rows(self): + """测试空结果。""" + result = parse_sqlite_explain([]) + assert result["db_type"] == "sqlite" + assert result["execution_plan"] == [] + + +# =========================================================================== +# Part 4: 统一辅助函数测试 +# =========================================================================== + + +class TestHelperFunctions: + """测试辅助函数。""" + + def test_make_plan_node_defaults(self): + """测试 make_plan_node 默认值。""" + node = make_plan_node(id=1, operation="Table Scan") + assert node["id"] == 1 + assert node["operation"] == "Table Scan" + assert node["table"] is None + assert node["rows"] is None + assert node["cost"] is None + assert node["key"] is None + assert node["extra"] is None + assert node["children"] == [] + + def test_make_plan_node_full(self): + """测试 make_plan_node 完整参数。""" + node = make_plan_node( + id=2, + operation="Index Scan", + table="orders", + rows=100, + cost=5.5, + key="idx_id", + extra="Using index", + children=[{"id": 3, "operation": "Child"}], + alias="o", + ) + assert node["table"] == "orders" + assert node["rows"] == 100 + assert node["cost"] == 5.5 + assert node["key"] == "idx_id" + assert len(node["children"]) == 1 + assert node["alias"] == "o" + + def test_empty_result(self): + """测试空结果构造。""" + result = _empty_result("mysql", {"foo": "bar"}, error="test error") + assert result["db_type"] == "mysql" + assert result["execution_plan"] == [] + assert "test error" in result["warnings"] + assert result["raw_result"] == {"foo": "bar"} + + def test_empty_result_no_error(self): + """测试无 error 的空结果。""" + result = _empty_result("postgres", None) + assert "EXPLAIN 结果为空" in result["warnings"][0] + + +# =========================================================================== +# Part 5: ExplainService.execute_explain 路由测试 (mock DB) +# =========================================================================== + + +class TestExplainEndpoint: + """测试 POST /api/v1/sql-assistant/explain API endpoint。""" + + @pytest.fixture + async def mysql_instance(self, db): + """创建 MySQL 测试实例。""" + inst = Instance( + name="TestMySQL", + db_type="mysql", + host="127.0.0.1", + port=3306, + username="root", + password=encrypt_password("test123"), + database="test_db", + env="development", + status="disconnected", + ) + db.add(inst) + await db.commit() + await db.refresh(inst) + return inst + + @pytest.fixture + async def pg_instance(self, db): + """创建 PostgreSQL 测试实例。""" + inst = Instance( + name="TestPG", + db_type="postgres", + host="127.0.0.1", + port=5432, + username="postgres", + password=encrypt_password("test123"), + database="test_db", + env="development", + status="disconnected", + ) + db.add(inst) + await db.commit() + await db.refresh(inst) + return inst + + @pytest.fixture + async def sqlite_instance(self, db): + """创建 SQLite 测试实例。""" + inst = Instance( + name="TestSQLite", + db_type="sqlite", + host="/tmp/test.db", + port=0, + username="", + password="", + database="", + env="development", + status="disconnected", + ) + db.add(inst) + await db.commit() + await db.refresh(inst) + return inst + + async def test_explain_mysql_success(self, client, admin_token, mysql_instance): + """测试 MySQL EXPLAIN 成功。""" + # Mock aiomysql + mock_cursor = AsyncMock() + mock_cursor.execute = AsyncMock() + mock_cursor.fetchall = AsyncMock(return_value=[ + {"EXPLAIN": json.dumps({ + "query_block": { + "select_id": 1, + "cost_info": {"query_cost": "0.35"}, + "table": { + "table_name": "users", + "access_type": "ALL", + "rows_examined_per_scan": 1000, + "filtered": "100.00", + }, + } + })} + ]) + mock_cursor.description = [("EXPLAIN",)] + mock_cursor.__aenter__ = AsyncMock(return_value=mock_cursor) + mock_cursor.__aexit__ = AsyncMock(return_value=False) + + mock_conn = AsyncMock() + mock_conn.cursor = MagicMock(return_value=mock_cursor) + mock_conn.close = MagicMock() + + with patch("app.services.explain_service.aiomysql") as mock_aiomysql: + mock_aiomysql.connect = AsyncMock(return_value=mock_conn) + mock_aiomysql.DictCursor = MagicMock() + + resp = await client.post( + "/api/v1/sql-assistant/explain", + json={ + "instance_id": mysql_instance.id, + "database": "test_db", + "query": "SELECT * FROM users", + }, + headers=auth_header(admin_token), + ) + + assert resp.status_code == 200 + data = resp.json() + assert data["db_type"] == "mysql" + assert len(data["execution_plan"]) == 1 + assert data["execution_plan"][0]["operation"] == "Table Scan" + assert data["execution_plan"][0]["table"] == "users" + assert "elapsed_ms" in data + assert isinstance(data["summary"], dict) + + async def test_explain_postgres_success(self, client, admin_token, pg_instance): + """测试 PostgreSQL EXPLAIN 成功。""" + mock_conn = AsyncMock() + mock_conn.fetchval = AsyncMock(return_value=json.dumps([ + { + "Plan": { + "Node Type": "Seq Scan", + "Relation Name": "users", + "Total Cost": 15.50, + "Plan Rows": 1000, + } + } + ])) + mock_conn.close = AsyncMock() + + with patch("app.services.explain_service.asyncpg") as mock_asyncpg: + mock_asyncpg.connect = AsyncMock(return_value=mock_conn) + + resp = await client.post( + "/api/v1/sql-assistant/explain", + json={ + "instance_id": pg_instance.id, + "database": "test_db", + "query": "SELECT * FROM users", + }, + headers=auth_header(admin_token), + ) + + assert resp.status_code == 200 + data = resp.json() + assert data["db_type"] == "postgres" + assert len(data["execution_plan"]) == 1 + assert data["execution_plan"][0]["operation"] == "Table Scan" + assert data["execution_plan"][0]["table"] == "users" + + async def test_explain_empty_query(self, client, admin_token, mysql_instance): + """测试空查询报错。""" + resp = await client.post( + "/api/v1/sql-assistant/explain", + json={ + "instance_id": mysql_instance.id, + "database": "test_db", + "query": "", + }, + headers=auth_header(admin_token), + ) + assert resp.status_code == 400 + + async def test_explain_non_select_query(self, client, admin_token, mysql_instance): + """测试非 SELECT 查询报错。""" + mock_conn = AsyncMock() + mock_conn.cursor = MagicMock() + mock_conn.close = MagicMock() + + with patch("app.services.explain_service.aiomysql") as mock_aiomysql: + mock_aiomysql.connect = AsyncMock(return_value=mock_conn) + mock_aiomysql.DictCursor = MagicMock() + + resp = await client.post( + "/api/v1/sql-assistant/explain", + json={ + "instance_id": mysql_instance.id, + "database": "test_db", + "query": "DELETE FROM users WHERE id = 1", + }, + headers=auth_header(admin_token), + ) + + assert resp.status_code == 400 + assert "SELECT" in resp.json()["detail"] or "WITH" in resp.json()["detail"] + + async def test_explain_instance_not_found(self, client, admin_token): + """测试不存在的实例。""" + resp = await client.post( + "/api/v1/sql-assistant/explain", + json={ + "instance_id": 99999, + "database": "test_db", + "query": "SELECT 1", + }, + headers=auth_header(admin_token), + ) + assert resp.status_code == 404 + + async def test_explain_guest_forbidden(self, client, guest_token, mysql_instance): + """测试访客无权限。""" + resp = await client.post( + "/api/v1/sql-assistant/explain", + json={ + "instance_id": mysql_instance.id, + "database": "test_db", + "query": "SELECT 1", + }, + headers=auth_header(guest_token), + ) + assert resp.status_code == 403 + + async def test_explain_with_with_clause(self, client, admin_token, pg_instance): + """测试 WITH (CTE) 查询。""" + mock_conn = AsyncMock() + mock_conn.fetchval = AsyncMock(return_value=json.dumps([ + { + "Plan": { + "Node Type": "CTE Scan", + "Total Cost": 5.0, + "Plan Rows": 10, + } + } + ])) + mock_conn.close = AsyncMock() + + with patch("app.services.explain_service.asyncpg") as mock_asyncpg: + mock_asyncpg.connect = AsyncMock(return_value=mock_conn) + + resp = await client.post( + "/api/v1/sql-assistant/explain", + json={ + "instance_id": pg_instance.id, + "database": "test_db", + "query": "WITH cte AS (SELECT 1) SELECT * FROM cte", + }, + headers=auth_header(admin_token), + ) + + assert resp.status_code == 200 + data = resp.json() + assert data["db_type"] == "postgres" + + async def test_explain_unsupported_db_type(self, client, admin_token, db): + """测试不支持的 db_type。""" + # Create a Redis instance + inst = Instance( + name="TestRedis", + db_type="redis", + host="127.0.0.1", + port=6379, + username="", + password=encrypt_password(""), + database="0", + env="development", + status="disconnected", + ) + db.add(inst) + await db.commit() + await db.refresh(inst) + + resp = await client.post( + "/api/v1/sql-assistant/explain", + json={ + "instance_id": inst.id, + "database": "0", + "query": "SELECT * FROM users", + }, + headers=auth_header(admin_token), + ) + # Either 400 or 500 depending on the error path + assert resp.status_code in (400, 500) + + +# =========================================================================== +# Part 6: ExplainService 单元测试 (直接 mock 连接) +# =========================================================================== + + +class TestExplainServiceUnit: + """直接测试 ExplainService 的路由逻辑。""" + + def _make_instance(self, db_type="mysql", **overrides): + defaults = dict( + name="Test", + db_type=db_type, + host="127.0.0.1", + port=3306, + username="root", + password="test123", + database="test_db", + ) + defaults.update(overrides) + inst = MagicMock(spec=Instance) + for k, v in defaults.items(): + setattr(inst, k, v) + return inst + + @pytest.mark.asyncio + async def test_mysql_explain_service(self): + """测试 ExplainService MySQL 路径。""" + instance = self._make_instance(db_type="mysql") + + mock_cursor = AsyncMock() + mock_cursor.execute = AsyncMock() + mock_cursor.fetchall = AsyncMock(return_value=[ + {"EXPLAIN": json.dumps({ + "query_block": { + "select_id": 1, + "table": { + "table_name": "t", + "access_type": "ALL", + "rows_examined_per_scan": 50, + }, + } + })} + ]) + mock_cursor.description = [("EXPLAIN",)] + mock_cursor.__aenter__ = AsyncMock(return_value=mock_cursor) + mock_cursor.__aexit__ = AsyncMock(return_value=False) + + mock_conn = AsyncMock() + mock_conn.cursor = MagicMock(return_value=mock_cursor) + mock_conn.close = MagicMock() + + with patch("app.services.explain_service.aiomysql") as mock_aiomysql: + mock_aiomysql.connect = AsyncMock(return_value=mock_conn) + mock_aiomysql.DictCursor = MagicMock() + + result = await ExplainService.execute_explain( + instance, "test_db", "SELECT * FROM t" + ) + + assert result["db_type"] == "mysql" + assert result["execution_plan"][0]["operation"] == "Table Scan" + + @pytest.mark.asyncio + async def test_postgres_explain_service(self): + """测试 ExplainService PostgreSQL 路径。""" + instance = self._make_instance(db_type="postgres", port=5432) + + mock_conn = AsyncMock() + mock_conn.fetchval = AsyncMock(return_value=json.dumps([ + {"Plan": {"Node Type": "Seq Scan", "Relation Name": "t", "Total Cost": 5.0, "Plan Rows": 10}} + ])) + mock_conn.close = AsyncMock() + + with patch("app.services.explain_service.asyncpg") as mock_asyncpg: + mock_asyncpg.connect = AsyncMock(return_value=mock_conn) + + result = await ExplainService.execute_explain( + instance, "test_db", "SELECT * FROM t" + ) + + assert result["db_type"] == "postgres" + assert result["execution_plan"][0]["operation"] == "Table Scan" + + @pytest.mark.asyncio + async def test_unsupported_db_type(self): + """测试不支持的 db_type。""" + instance = self._make_instance(db_type="redis") + + with pytest.raises(ValueError, match="暂不支持"): + await ExplainService.execute_explain(instance, "0", "SELECT * FROM key") + + @pytest.mark.asyncio + async def test_non_select_rejected(self): + """测试非 SELECT/WITH 查询被拒绝。""" + instance = self._make_instance(db_type="mysql") + + with pytest.raises(ValueError, match="SELECT"): + await ExplainService.execute_explain(instance, "test_db", "INSERT INTO t VALUES (1)") + + @pytest.mark.asyncio + async def test_with_query_accepted(self): + """测试 WITH (CTE) 查询被接受。""" + instance = self._make_instance(db_type="postgres", port=5432) + + mock_conn = AsyncMock() + mock_conn.fetchval = AsyncMock(return_value=json.dumps([ + {"Plan": {"Node Type": "CTE Scan", "Total Cost": 1.0, "Plan Rows": 1}} + ])) + mock_conn.close = AsyncMock() + + with patch("app.services.explain_service.asyncpg") as mock_asyncpg: + mock_asyncpg.connect = AsyncMock(return_value=mock_conn) + + result = await ExplainService.execute_explain( + instance, "test_db", "WITH cte AS (SELECT 1) SELECT * FROM cte" + ) + + assert result["db_type"] == "postgres" + + @pytest.mark.asyncio + async def test_sqlite_explain_service(self): + """测试 ExplainService SQLite 路径。""" + instance = self._make_instance(db_type="sqlite", host="/tmp/test_explain.db", port=0) + + mock_rows = [ + {"id": 2, "parent": 0, "notused": 0, "detail": "SEARCH TABLE users USING INDEX idx_id (id=?)"}, + ] + + with patch("app.services.explain_service.asyncio") as mock_asyncio_mod: + # We need to mock asyncio.get_event_loop().run_in_executor() + mock_loop = MagicMock() + mock_loop.run_in_executor = AsyncMock(return_value=mock_rows) + mock_asyncio_mod.get_event_loop.return_value = mock_loop + mock_asyncio_mod.wait_for = AsyncMock(return_value=mock_rows) + # Keep the real TimeoutError + mock_asyncio_mod.TimeoutError = asyncio.TimeoutError + + # Also need to patch sqlite3 inside the module + import sqlite3 as real_sqlite3 + + result = await ExplainService._explain_sqlite(instance, "SELECT * FROM users WHERE id = 1") + + assert result["db_type"] == "sqlite" + assert result["execution_plan"][0]["operation"] == "Index Lookup" + assert result["execution_plan"][0]["table"] == "users" + assert result["execution_plan"][0]["key"] == "idx_id" diff --git a/backend/tests/test_query_history.py b/backend/tests/test_query_history_service.py similarity index 64% rename from backend/tests/test_query_history.py rename to backend/tests/test_query_history_service.py index 2c71e0c..408f5f3 100644 --- a/backend/tests/test_query_history.py +++ b/backend/tests/test_query_history_service.py @@ -1,7 +1,7 @@ -"""查询历史和快捷查询 API 测试 — P1 审计闭环。 +"""查询历史和快捷查询 持久化测试 — 验证 SQLite 存储。 测试范围: -- 查询历史记录和查看 +- 查询历史记录和查看（CRUD + 分页 + 过滤） - 收藏切换 - 快捷查询 CRUD - 查询统计 @@ -10,10 +10,11 @@ """ import pytest import pytest_asyncio -from httpx import AsyncClient +from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import create_access_token, hash_password from app.models.user import User +from app.models.query_history import QueryHistory, QuickQueryModel from app.services.query_history_service import QueryHistoryService @@ -39,9 +40,10 @@ class TestQueryHistoryRecord: """查询历史记录功能测试。""" @pytest.mark.asyncio - async def test_record_query(self): + async def test_record_query(self, db: AsyncSession): """测试记录一条查询。""" entry = await QueryHistoryService.record_query( + db=db, user_id=1, username="admin", instance_id=1, @@ -53,6 +55,7 @@ class TestQueryHistoryRecord: execution_time_ms=50, result_rows=10, ) + await db.commit() assert entry.id > 0 assert entry.user_id == 1 assert entry.db_type == "mysql" @@ -63,9 +66,10 @@ class TestQueryHistoryRecord: assert entry.error_message is None @pytest.mark.asyncio - async def test_record_failed_query(self): + async def test_record_failed_query(self, db: AsyncSession): """测试记录失败的查询。""" entry = await QueryHistoryService.record_query( + db=db, user_id=1, username="admin", instance_id=1, @@ -79,11 +83,12 @@ class TestQueryHistoryRecord: is_success=False, error_message="Table doesn't exist", ) + await db.commit() assert entry.is_success is False assert entry.error_message == "Table doesn't exist" @pytest.mark.asyncio - async def test_record_multiple_db_types(self): + async def test_record_multiple_db_types(self, db: AsyncSession): """测试记录不同数据库类型的查询。""" for db_type, query, q_type in [ ("mysql", "SELECT 1", "SELECT"), @@ -92,6 +97,7 @@ class TestQueryHistoryRecord: ("mongo", '{"collection": "users", "operation": "find"}', "find"), ]: entry = await QueryHistoryService.record_query( + db=db, user_id=1, username="admin", instance_id=1, @@ -103,14 +109,16 @@ class TestQueryHistoryRecord: execution_time_ms=10, result_rows=1, ) + await db.commit() assert entry.db_type == db_type assert entry.query_type == q_type @pytest.mark.asyncio - async def test_record_query_long_content_truncated(self): + async def test_record_query_long_content_truncated(self, db: AsyncSession): """测试超长查询内容被截断。""" long_sql = "SELECT * FROM t WHERE id = " + "1" * 3000 entry = await QueryHistoryService.record_query( + db=db, user_id=1, username="admin", instance_id=1, @@ -122,18 +130,16 @@ class TestQueryHistoryRecord: execution_time_ms=10, result_rows=0, ) + await db.commit() assert len(entry.query_content) == 2000 @pytest.mark.asyncio - async def test_record_query_max_history_per_user(self): + async def test_record_query_max_history_per_user(self, db: AsyncSession): """测试每个用户最多保留500条记录。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - # 创建超过限制的记录 for i in range(510): await QueryHistoryService.record_query( + db=db, user_id=99, username="test_user", instance_id=1, @@ -145,13 +151,14 @@ class TestQueryHistoryRecord: execution_time_ms=i, result_rows=0, ) + await db.commit() - user_entries = [h for h in QueryHistoryService._history if h.user_id == 99] - assert len(user_entries) <= 500 - - # 清理 - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 + from sqlalchemy import select, func + count_stmt = select(func.count()).select_from(QueryHistory).where( + QueryHistory.user_id == 99 + ) + total = (await db.execute(count_stmt)).scalar() + assert total <= 500 # ── 查询历史查看 ── @@ -160,12 +167,8 @@ class TestQueryHistoryRecord: class TestQueryHistoryView: """查询历史查看功能测试。""" - async def _setup_history(self): + async def _setup_history(self, db: AsyncSession): """准备测试数据。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - entries = [ ("SELECT * FROM users", "mysql", 1, 50, 10), ("SELECT * FROM orders", "mysql", 1, 30, 20), @@ -174,6 +177,7 @@ class TestQueryHistoryView: ] for sql, db_type, inst_id, time_ms, rows in entries: await QueryHistoryService.record_query( + db=db, user_id=1, username="admin", instance_id=inst_id, @@ -185,67 +189,68 @@ class TestQueryHistoryView: execution_time_ms=time_ms, result_rows=rows, ) + await db.commit() @pytest.mark.asyncio - async def test_get_history_default(self): + async def test_get_history_default(self, db: AsyncSession): """测试默认查看历史。""" - await self._setup_history() - result = await QueryHistoryService.get_history(user_id=1) + await self._setup_history(db) + result = await QueryHistoryService.get_history(db=db, user_id=1) assert result["total"] == 4 assert result["page"] == 1 assert result["page_size"] == 20 assert len(result["items"]) == 4 @pytest.mark.asyncio - async def test_get_history_filter_by_db_type(self): + async def test_get_history_filter_by_db_type(self, db: AsyncSession): """测试按数据库类型筛选。""" - await self._setup_history() - result = await QueryHistoryService.get_history(user_id=1, db_type="mysql") + await self._setup_history(db) + result = await QueryHistoryService.get_history(db=db, user_id=1, db_type="mysql") assert result["total"] == 2 for item in result["items"]: assert item["db_type"] == "mysql" @pytest.mark.asyncio - async def test_get_history_filter_by_instance(self): + async def test_get_history_filter_by_instance(self, db: AsyncSession): """测试按实例ID筛选。""" - await self._setup_history() - result = await QueryHistoryService.get_history(user_id=1, instance_id=2) + await self._setup_history(db) + result = await QueryHistoryService.get_history(db=db, user_id=1, instance_id=2) assert result["total"] == 1 assert result["items"][0]["db_type"] == "redis" @pytest.mark.asyncio - async def test_get_history_filter_by_keyword(self): + async def test_get_history_filter_by_keyword(self, db: AsyncSession): """测试按关键词筛选。""" - await self._setup_history() - result = await QueryHistoryService.get_history(user_id=1, keyword="users") + await self._setup_history(db) + result = await QueryHistoryService.get_history(db=db, user_id=1, keyword="users") assert result["total"] == 1 assert "users" in result["items"][0]["query_content"].lower() @pytest.mark.asyncio - async def test_get_history_pagination(self): + async def test_get_history_pagination(self, db: AsyncSession): """测试分页。""" - await self._setup_history() - result = await QueryHistoryService.get_history(user_id=1, page=1, page_size=2) + await self._setup_history(db) + result = await QueryHistoryService.get_history(db=db, user_id=1, page=1, page_size=2) assert result["total"] == 4 assert len(result["items"]) == 2 - result2 = await QueryHistoryService.get_history(user_id=1, page=2, page_size=2) + result2 = await QueryHistoryService.get_history(db=db, user_id=1, page=2, page_size=2) assert result2["total"] == 4 assert len(result2["items"]) == 2 @pytest.mark.asyncio - async def test_get_history_other_user_empty(self): + async def test_get_history_other_user_empty(self, db: AsyncSession): """测试查看其他用户的历史为空。""" - await self._setup_history() - result = await QueryHistoryService.get_history(user_id=999) + await self._setup_history(db) + result = await QueryHistoryService.get_history(db=db, user_id=999) assert result["total"] == 0 assert result["items"] == [] @pytest.mark.asyncio - async def test_history_sorted_by_time_desc(self): + async def test_history_sorted_by_time_desc(self, db: AsyncSession): """测试历史按时间倒序排列。""" - await self._setup_history() - result = await QueryHistoryService.get_history(user_id=1) + await self._setup_history(db) + result = await QueryHistoryService.get_history(db=db, user_id=1) times = [item["created_at"] for item in result["items"]] assert times == sorted(times, reverse=True) @@ -256,61 +261,61 @@ class TestQueryHistoryView: class TestQueryHistoryFavorite: """查询收藏功能测试。""" - async def _setup_for_favorite(self): - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - + async def _setup_for_favorite(self, db: AsyncSession): entry = await QueryHistoryService.record_query( + db=db, user_id=1, username="admin", instance_id=1, instance_name="测试", db_type="mysql", database_name="test", query_content="SELECT 1", query_type="SELECT", execution_time_ms=10, result_rows=1, ) + await db.commit() return entry.id @pytest.mark.asyncio - async def test_toggle_favorite_add(self): + async def test_toggle_favorite_add(self, db: AsyncSession): """测试添加收藏。""" - entry_id = await self._setup_for_favorite() - is_fav = await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id) + entry_id = await self._setup_for_favorite(db) + is_fav = await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id) + await db.commit() assert is_fav is True @pytest.mark.asyncio - async def test_toggle_favorite_remove(self): + async def test_toggle_favorite_remove(self, db: AsyncSession): """测试取消收藏。""" - entry_id = await self._setup_for_favorite() - await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id) - is_fav = await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id) + entry_id = await self._setup_for_favorite(db) + await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id) + await db.commit() + is_fav = await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id) + await db.commit() assert is_fav is False @pytest.mark.asyncio - async def test_get_favorites(self): + async def test_get_favorites(self, db: AsyncSession): """测试获取收藏列表。""" - entry_id = await self._setup_for_favorite() - await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id) + entry_id = await self._setup_for_favorite(db) + await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id) + await db.commit() - result = await QueryHistoryService.get_favorites(user_id=1) + result = await QueryHistoryService.get_favorites(db=db, user_id=1) assert result["total"] == 1 assert result["items"][0]["id"] == entry_id @pytest.mark.asyncio - async def test_get_favorites_empty(self): + async def test_get_favorites_empty(self, db: AsyncSession): """测试无收藏时返回空列表。""" - QueryHistoryService._history = [] - QueryHistoryService._favorites = set() - - result = await QueryHistoryService.get_favorites(user_id=1) + result = await QueryHistoryService.get_favorites(db=db, user_id=1) assert result["total"] == 0 assert result["items"] == [] @pytest.mark.asyncio - async def test_favorites_per_user(self): + async def test_favorites_per_user(self, db: AsyncSession): """测试收藏是用户隔离的。""" - entry_id = await self._setup_for_favorite() - await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id) + entry_id = await self._setup_for_favorite(db) + await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id) + await db.commit() - result = await QueryHistoryService.get_favorites(user_id=999) + result = await QueryHistoryService.get_favorites(db=db, user_id=999) assert result["total"] == 0 @@ -320,15 +325,11 @@ class TestQueryHistoryFavorite: class TestQuickQueries: """快捷查询功能测试。""" - async def _setup(self): - QueryHistoryService._quick_queries = [] - QueryHistoryService._quick_counter = 0 - @pytest.mark.asyncio - async def test_create_quick_query(self): + async def test_create_quick_query(self, db: AsyncSession): """测试创建快捷查询。""" - await self._setup() qq = await QueryHistoryService.create_quick_query( + db=db, name="查看活跃用户", description="查看最近活跃的用户列表", query_content="SELECT * FROM users WHERE last_login > DATE_SUB(NOW(), INTERVAL 7 DAY)", @@ -338,117 +339,121 @@ class TestQuickQueries: created_by=1, created_by_name="admin", ) + await db.commit() assert qq.id > 0 assert qq.name == "查看活跃用户" assert qq.is_shared is False assert qq.created_by == 1 @pytest.mark.asyncio - async def test_get_quick_queries_personal(self): + async def test_get_quick_queries_personal(self, db: AsyncSession): """测试获取个人快捷查询。""" - await self._setup() await QueryHistoryService.create_quick_query( - name="个人查询", description="", query_content="SELECT 1", + db=db, name="个人查询", description="", query_content="SELECT 1", db_type="mysql", database_pattern="*", is_shared=False, created_by=1, created_by_name="admin", ) await QueryHistoryService.create_quick_query( - name="他人查询", description="", query_content="SELECT 2", + db=db, name="他人查询", description="", query_content="SELECT 2", db_type="mysql", database_pattern="*", is_shared=False, created_by=2, created_by_name="dba", ) + await db.commit() # 用户1只能看到自己的 + 共享的 - items = await QueryHistoryService.get_quick_queries(user_id=1) + items = await QueryHistoryService.get_quick_queries(db=db, user_id=1) assert len(items) == 1 assert items[0]["name"] == "个人查询" @pytest.mark.asyncio - async def test_get_quick_queries_shared(self): + async def test_get_quick_queries_shared(self, db: AsyncSession): """测试获取共享快捷查询。""" - await self._setup() await QueryHistoryService.create_quick_query( - name="共享查询", description="", query_content="SELECT 1", + db=db, name="共享查询", description="", query_content="SELECT 1", db_type="mysql", database_pattern="*", is_shared=True, created_by=2, created_by_name="dba", ) + await db.commit() - items = await QueryHistoryService.get_quick_queries(user_id=1) + items = await QueryHistoryService.get_quick_queries(db=db, user_id=1) assert len(items) == 1 assert items[0]["is_shared"] is True @pytest.mark.asyncio - async def test_get_quick_queries_filter_db_type(self): + async def test_get_quick_queries_filter_db_type(self, db: AsyncSession): """测试按数据库类型筛选快捷查询。""" - await self._setup() await QueryHistoryService.create_quick_query( - name="MySQL查询", description="", query_content="SELECT 1", + db=db, name="MySQL查询", description="", query_content="SELECT 1", db_type="mysql", database_pattern="*", is_shared=True, created_by=1, created_by_name="admin", ) await QueryHistoryService.create_quick_query( - name="Redis查询", description="", query_content="GET key", + db=db, name="Redis查询", description="", query_content="GET key", db_type="redis", database_pattern="*", is_shared=True, created_by=1, created_by_name="admin", ) + await db.commit() - items = await QueryHistoryService.get_quick_queries(user_id=1, db_type="redis") + items = await QueryHistoryService.get_quick_queries(db=db, user_id=1, db_type="redis") assert len(items) == 1 assert items[0]["db_type"] == "redis" @pytest.mark.asyncio - async def test_get_quick_queries_filter_keyword(self): + async def test_get_quick_queries_filter_keyword(self, db: AsyncSession): """测试按关键词筛选快捷查询。""" - await self._setup() await QueryHistoryService.create_quick_query( - name="查看活跃用户", description="", query_content="SELECT 1", + db=db, name="查看活跃用户", description="", query_content="SELECT 1", db_type="mysql", database_pattern="*", is_shared=True, created_by=1, created_by_name="admin", ) await QueryHistoryService.create_quick_query( - name="查看慢查询", description="", query_content="SELECT 2", + db=db, name="查看慢查询", description="", query_content="SELECT 2", db_type="mysql", database_pattern="*", is_shared=True, created_by=1, created_by_name="admin", ) + await db.commit() - items = await QueryHistoryService.get_quick_queries(user_id=1, keyword="活跃") + items = await QueryHistoryService.get_quick_queries(db=db, user_id=1, keyword="活跃") assert len(items) == 1 assert "活跃" in items[0]["name"] @pytest.mark.asyncio - async def test_use_quick_query(self): + async def test_use_quick_query(self, db: AsyncSession): """测试使用快捷查询。""" - await self._setup() qq = await QueryHistoryService.create_quick_query( - name="测试", description="", query_content="SELECT 1", + db=db, name="测试", description="", query_content="SELECT 1", db_type="mysql", database_pattern="*", is_shared=True, created_by=1, created_by_name="admin", ) + await db.commit() - content = await QueryHistoryService.use_quick_query(qq.id) + content = await QueryHistoryService.use_quick_query(db, qq.id) + await db.commit() assert content == "SELECT 1" @pytest.mark.asyncio - async def test_use_quick_query_increments_count(self): + async def test_use_quick_query_increments_count(self, db: AsyncSession): """测试使用快捷查询增加使用次数。""" - await self._setup() qq = await QueryHistoryService.create_quick_query( - name="测试", description="", query_content="SELECT 1", + db=db, name="测试", description="", query_content="SELECT 1", db_type="mysql", database_pattern="*", is_shared=True, created_by=1, created_by_name="admin", ) + await db.commit() assert qq.usage_count == 0 - await QueryHistoryService.use_quick_query(qq.id) + await QueryHistoryService.use_quick_query(db, qq.id) + await db.commit() assert qq.usage_count == 1 - await QueryHistoryService.use_quick_query(qq.id) + await QueryHistoryService.use_quick_query(db, qq.id) + await db.commit() assert qq.usage_count == 2 @pytest.mark.asyncio - async def test_use_quick_query_not_found(self): + async def test_use_quick_query_not_found(self, db: AsyncSession): """测试使用不存在的快捷查询。""" - content = await QueryHistoryService.use_quick_query(99999) + content = await QueryHistoryService.use_quick_query(db, 99999) assert content is None @@ -459,24 +464,22 @@ class TestDefaultQuickQueries: """预置快捷查询模板测试。""" @pytest.mark.asyncio - async def test_default_templates_exist(self): + async def test_default_templates_exist(self, db: AsyncSession): """测试预置模板存在。""" - QueryHistoryService._quick_queries = [] - QueryHistoryService._quick_counter = 0 - QueryHistoryService.seed_default_quick_queries() + await QueryHistoryService.seed_default_quick_queries(db) + await db.commit() - items = await QueryHistoryService.get_quick_queries(user_id=1) + items = await QueryHistoryService.get_quick_queries(db=db, user_id=1) # 预置了10个模板 assert len(items) == 10 @pytest.mark.asyncio - async def test_default_templates_cover_db_types(self): + async def test_default_templates_cover_db_types(self, db: AsyncSession): """测试预置模板覆盖所有数据库类型。""" - QueryHistoryService._quick_queries = [] - QueryHistoryService._quick_counter = 0 - QueryHistoryService.seed_default_quick_queries() + await QueryHistoryService.seed_default_quick_queries(db) + await db.commit() - items = await QueryHistoryService.get_quick_queries(user_id=1) + items = await QueryHistoryService.get_quick_queries(db=db, user_id=1) db_types = {item["db_type"] for item in items} assert "mysql" in db_types assert "redis" in db_types @@ -484,27 +487,36 @@ class TestDefaultQuickQueries: assert "mongo" in db_types @pytest.mark.asyncio - async def test_default_templates_all_shared(self): + async def test_default_templates_all_shared(self, db: AsyncSession): """测试预置模板都是共享的。""" - QueryHistoryService._quick_queries = [] - QueryHistoryService._quick_counter = 0 - QueryHistoryService.seed_default_quick_queries() + await QueryHistoryService.seed_default_quick_queries(db) + await db.commit() - items = await QueryHistoryService.get_quick_queries(user_id=1) + items = await QueryHistoryService.get_quick_queries(db=db, user_id=1) for item in items: assert item["is_shared"] is True @pytest.mark.asyncio - async def test_default_templates_system_created(self): + async def test_default_templates_system_created(self, db: AsyncSession): """测试预置模板由系统创建。""" - QueryHistoryService._quick_queries = [] - QueryHistoryService._quick_counter = 0 - QueryHistoryService.seed_default_quick_queries() + await QueryHistoryService.seed_default_quick_queries(db) + await db.commit() - items = await QueryHistoryService.get_quick_queries(user_id=1) + items = await QueryHistoryService.get_quick_queries(db=db, user_id=1) for item in items: assert item["created_by_name"] == "系统" + @pytest.mark.asyncio + async def test_seed_idempotent(self, db: AsyncSession): + """测试 seed 是幂等的（多次调用不会重复插入）。""" + await QueryHistoryService.seed_default_quick_queries(db) + await db.commit() + await QueryHistoryService.seed_default_quick_queries(db) + await db.commit() + + items = await QueryHistoryService.get_quick_queries(db=db, user_id=1) + assert len(items) == 10 + # ── 查询统计 ── @@ -512,14 +524,10 @@ class TestDefaultQuickQueries: class TestQueryStatistics: """查询统计功能测试。""" - async def _setup_stats(self): - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - + async def _setup_stats(self, db: AsyncSession): for i in range(5): await QueryHistoryService.record_query( - user_id=1, username="admin", instance_id=1, + db=db, user_id=1, username="admin", instance_id=1, instance_name="测试", db_type="mysql", database_name="test", query_content=f"SELECT * FROM table_{i} WHERE id = 1", query_type="SELECT", @@ -528,29 +536,30 @@ class TestQueryStatistics: ) for i in range(3): await QueryHistoryService.record_query( - user_id=1, username="admin", instance_id=2, + db=db, user_id=1, username="admin", instance_id=2, instance_name="测试Redis", db_type="redis", database_name="db0", query_content="GET mykey", query_type="GET", execution_time_ms=5, result_rows=1, ) + await db.commit() @pytest.mark.asyncio - async def test_statistics_basic(self): + async def test_statistics_basic(self, db: AsyncSession): """测试基本统计。""" - await self._setup_stats() - stats = await QueryHistoryService.get_query_statistics(user_id=1) + await self._setup_stats(db) + stats = await QueryHistoryService.get_query_statistics(db=db, user_id=1) assert stats["total_queries"] == 8 assert stats["avg_execution_time_ms"] > 0 assert stats["success_rate"] == 100.0 @pytest.mark.asyncio - async def test_statistics_db_type_distribution(self): + async def test_statistics_db_type_distribution(self, db: AsyncSession): """测试数据库类型分布统计。""" - await self._setup_stats() - stats = await QueryHistoryService.get_query_statistics(user_id=1) + await self._setup_stats(db) + stats = await QueryHistoryService.get_query_statistics(db=db, user_id=1) assert "mysql" in stats["db_type_distribution"] assert "redis" in stats["db_type_distribution"] @@ -558,27 +567,26 @@ class TestQueryStatistics: assert stats["db_type_distribution"]["redis"] == 3 @pytest.mark.asyncio - async def test_statistics_daily_trend(self): + async def test_statistics_daily_trend(self, db: AsyncSession): """测试每日趋势。""" - await self._setup_stats() - stats = await QueryHistoryService.get_query_statistics(user_id=1) + await self._setup_stats(db) + stats = await QueryHistoryService.get_query_statistics(db=db, user_id=1) assert "daily_trend" in stats assert len(stats["daily_trend"]) == 7 @pytest.mark.asyncio - async def test_statistics_empty(self): + async def test_statistics_empty(self, db: AsyncSession): """测试无记录时返回提示。""" - QueryHistoryService._history = [] - stats = await QueryHistoryService.get_query_statistics(user_id=999) + stats = await QueryHistoryService.get_query_statistics(db=db, user_id=999) assert "message" in stats assert stats["message"] == "暂无查询记录" @pytest.mark.asyncio - async def test_statistics_top_tables(self): + async def test_statistics_top_tables(self, db: AsyncSession): """测试高频表统计。""" - await self._setup_stats() - stats = await QueryHistoryService.get_query_statistics(user_id=1) + await self._setup_stats(db) + stats = await QueryHistoryService.get_query_statistics(db=db, user_id=1) assert "top_tables" in stats # 应该能从SQL中提取表名 @@ -592,18 +600,14 @@ class TestQueryHistoryAPI: """查询历史 API 端点集成测试。""" @pytest.mark.asyncio - async def test_my_history_unauthorized(self, client: AsyncClient): + async def test_my_history_unauthorized(self, client): """测试未认证访问历史。""" resp = await client.get("/api/v1/query-history/my") assert resp.status_code == 401 @pytest.mark.asyncio - async def test_my_history_empty(self, client: AsyncClient, admin_headers: dict): + async def test_my_history_empty(self, client, admin_headers, db): """测试空历史记录。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - resp = await client.get("/api/v1/query-history/my", headers=admin_headers) assert resp.status_code == 200 data = resp.json() @@ -611,18 +615,15 @@ class TestQueryHistoryAPI: assert data["items"] == [] @pytest.mark.asyncio - async def test_my_history_with_data(self, client: AsyncClient, admin_headers: dict): + async def test_my_history_with_data(self, client, admin_headers, db, admin_user): """测试有数据的历史记录。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - await QueryHistoryService.record_query( - user_id=1, username="admin", instance_id=1, + db=db, user_id=admin_user.id, username="admin", instance_id=1, instance_name="测试", db_type="mysql", database_name="test", query_content="SELECT 1", query_type="SELECT", execution_time_ms=10, result_rows=1, ) + await db.commit() resp = await client.get("/api/v1/query-history/my", headers=admin_headers) assert resp.status_code == 200 @@ -630,24 +631,21 @@ class TestQueryHistoryAPI: assert data["total"] >= 1 @pytest.mark.asyncio - async def test_my_history_filter_db_type(self, client: AsyncClient, admin_headers: dict): + async def test_my_history_filter_db_type(self, client, admin_headers, db, admin_user): """测试按类型筛选。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - await QueryHistoryService.record_query( - user_id=1, username="admin", instance_id=1, + db=db, user_id=admin_user.id, username="admin", instance_id=1, instance_name="测试", db_type="mysql", database_name="test", query_content="SELECT 1", query_type="SELECT", execution_time_ms=10, result_rows=1, ) await QueryHistoryService.record_query( - user_id=1, username="admin", instance_id=2, + db=db, user_id=admin_user.id, username="admin", instance_id=2, instance_name="测试Redis", db_type="redis", database_name="db0", query_content="GET key", query_type="GET", execution_time_ms=5, result_rows=1, ) + await db.commit() resp = await client.get( "/api/v1/query-history/my?db_type=mysql", @@ -659,18 +657,15 @@ class TestQueryHistoryAPI: assert data["items"][0]["db_type"] == "mysql" @pytest.mark.asyncio - async def test_toggle_favorite_api(self, client: AsyncClient, admin_headers: dict): + async def test_toggle_favorite_api(self, client, admin_headers, db, admin_user): """测试收藏API。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - entry = await QueryHistoryService.record_query( - user_id=1, username="admin", instance_id=1, + db=db, user_id=admin_user.id, username="admin", instance_id=1, instance_name="测试", db_type="mysql", database_name="test", query_content="SELECT 1", query_type="SELECT", execution_time_ms=10, result_rows=1, ) + await db.commit() resp = await client.post( f"/api/v1/query-history/{entry.id}/favorite", @@ -681,12 +676,8 @@ class TestQueryHistoryAPI: assert data["is_favorite"] is True @pytest.mark.asyncio - async def test_favorites_api(self, client: AsyncClient, admin_headers: dict): + async def test_favorites_api(self, client, admin_headers, db): """测试收藏列表API。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - resp = await client.get("/api/v1/query-history/favorites", headers=admin_headers) assert resp.status_code == 200 data = resp.json() @@ -694,23 +685,18 @@ class TestQueryHistoryAPI: assert "items" in data @pytest.mark.asyncio - async def test_statistics_api(self, client: AsyncClient, admin_headers: dict): + async def test_statistics_api(self, client, admin_headers, db): """测试统计API。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - resp = await client.get("/api/v1/query-history/statistics", headers=admin_headers) assert resp.status_code == 200 data = resp.json() assert "message" in data or "total_queries" in data @pytest.mark.asyncio - async def test_quick_queries_list_api(self, client: AsyncClient, admin_headers: dict): + async def test_quick_queries_list_api(self, client, admin_headers, db): """测试快捷查询列表API。""" - QueryHistoryService._quick_queries = [] - QueryHistoryService._quick_counter = 0 - QueryHistoryService.seed_default_quick_queries() + await QueryHistoryService.seed_default_quick_queries(db) + await db.commit() resp = await client.get("/api/v1/query-history/quick-queries", headers=admin_headers) assert resp.status_code == 200 @@ -719,11 +705,8 @@ class TestQueryHistoryAPI: assert len(data["items"]) == 10 @pytest.mark.asyncio - async def test_quick_queries_create_api(self, client: AsyncClient, admin_headers: dict): + async def test_quick_queries_create_api(self, client, admin_headers, db): """测试创建快捷查询API。""" - QueryHistoryService._quick_queries = [] - QueryHistoryService._quick_counter = 0 - resp = await client.post( "/api/v1/query-history/quick-queries", params={ @@ -741,11 +724,10 @@ class TestQueryHistoryAPI: assert "id" in data @pytest.mark.asyncio - async def test_quick_queries_filter_api(self, client: AsyncClient, admin_headers: dict): + async def test_quick_queries_filter_api(self, client, admin_headers, db): """测试快捷查询筛选API。""" - QueryHistoryService._quick_queries = [] - QueryHistoryService._quick_counter = 0 - QueryHistoryService.seed_default_quick_queries() + await QueryHistoryService.seed_default_quick_queries(db) + await db.commit() resp = await client.get( "/api/v1/query-history/quick-queries?db_type=redis", @@ -757,29 +739,22 @@ class TestQueryHistoryAPI: assert item["db_type"] == "redis" @pytest.mark.asyncio - async def test_guest_can_view_own_history(self, client: AsyncClient, guest_headers: dict): + async def test_guest_can_view_own_history(self, client, guest_headers, db): """测试访客可以查看自己的历史（但无权执行查询）。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - resp = await client.get("/api/v1/query-history/my", headers=guest_headers) assert resp.status_code == 200 @pytest.mark.asyncio - async def test_pagination_parameters(self, client: AsyncClient, admin_headers: dict): + async def test_pagination_parameters(self, client, admin_headers, db, admin_user): """测试分页参数。""" - QueryHistoryService._history = [] - QueryHistoryService._history_counter = 0 - QueryHistoryService._favorites = set() - for i in range(15): await QueryHistoryService.record_query( - user_id=1, username="admin", instance_id=1, + db=db, user_id=admin_user.id, username="admin", instance_id=1, instance_name="测试", db_type="mysql", database_name="test", query_content=f"SELECT {i}", query_type="SELECT", execution_time_ms=10, result_rows=1, ) + await db.commit() resp = await client.get( "/api/v1/query-history/my?page=1&page_size=5", diff --git a/frontend/src/pages/AuditLogsPage.tsx b/frontend/src/pages/AuditLogsPage.tsx index 58f4182..9d3ca78 100644 --- a/frontend/src/pages/AuditLogsPage.tsx +++ b/frontend/src/pages/AuditLogsPage.tsx @@ -1,4 +1,4 @@ -import { useState, useEffect, useCallback, useMemo } from 'react'; +import { useState, useEffect, useCallback, useMemo, Fragment } from 'react'; import { Search, Download, Filter, ChevronDown, ChevronRight, FileText, User, Globe, CheckCircle2, XCircle, AlertTriangle, @@ -440,7 +440,7 @@ export default function AuditLogsPage() { const StatusIcon = statusConfig.icon; const TargetIcon = TARGET_ICONS[log.target_type] || FileText; return ( - <> + )} - + ); }) )}