logan
88e5f4e7ee
- 新增 crypto.go: AES-256-GCM 加解密，与 Python backend 兼容 - auth.go: Login 添加 LDAP fallback（本地优先→LDAP绑定→自动创建本地用户） - ldap.go: 实现真实 LDAP 连接/用户搜索/密码验证/OU列表/用户列表 - 所有 LDAP 操作解密 admin_password 后使用 - 新增/更新密码存储时加密 - router.go: 补充 delete-user 路由 - 添加 go-ldap/v3 依赖
327 lines
12 KiB
Python
327 lines
12 KiB
Python
"""SQL 助手服务测试 — 验证格式化、高亮、风险检测、EXPLAIN 解析和查询建议。"""
|
|
|
|
import pytest
|
|
from app.services.sql_assistant import SQLAssistant
|
|
|
|
|
|
# ===== SQL 格式化 =====
|
|
|
|
|
|
class TestSQLFormat:
|
|
"""SQL 格式化测试。"""
|
|
|
|
def test_format_simple_select(self):
|
|
"""应格式化简单 SELECT。"""
|
|
result = SQLAssistant.format_sql("select * from users where id = 1")
|
|
assert result.is_valid is True
|
|
assert "SELECT" in result.formatted
|
|
assert "FROM" in result.formatted
|
|
|
|
def test_format_empty_sql(self):
|
|
"""空 SQL 应返回无效。"""
|
|
result = SQLAssistant.format_sql("")
|
|
assert result.is_valid is False
|
|
assert result.error is not None
|
|
|
|
def test_format_none_sql(self):
|
|
"""None 应返回无效。"""
|
|
result = SQLAssistant.format_sql(None)
|
|
assert result.is_valid is False
|
|
|
|
def test_format_whitespace_sql(self):
|
|
"""纯空格 SQL 应返回无效。"""
|
|
result = SQLAssistant.format_sql(" ")
|
|
assert result.is_valid is False
|
|
|
|
def test_format_preserves_content(self):
|
|
"""格式化不应丢失 SQL 内容。"""
|
|
sql = "SELECT id, name FROM users WHERE status = 'active' ORDER BY id LIMIT 10"
|
|
result = SQLAssistant.format_sql(sql)
|
|
assert result.is_valid is True
|
|
assert "users" in result.formatted
|
|
assert "status" in result.formatted
|
|
|
|
def test_format_complex_join(self):
|
|
"""应格式化复杂 JOIN 查询。"""
|
|
sql = "select u.id, u.name, o.total from users u join orders o on u.id = o.user_id where o.total > 100"
|
|
result = SQLAssistant.format_sql(sql)
|
|
assert result.is_valid is True
|
|
assert "JOIN" in result.formatted.upper() or "join" in result.formatted
|
|
|
|
|
|
# ===== SQL 高亮 =====
|
|
|
|
|
|
class TestSQLHighlight:
|
|
"""SQL 语法高亮和风险分析测试。"""
|
|
|
|
def test_highlight_select(self):
|
|
"""应标记 SELECT 关键字。"""
|
|
result = SQLAssistant.highlight_sql("SELECT * FROM users")
|
|
assert "SELECT" in result.keywords
|
|
|
|
def test_highlight_extract_table(self):
|
|
"""应从 FROM 子句提取表名。"""
|
|
result = SQLAssistant.highlight_sql("SELECT * FROM users WHERE id = 1")
|
|
assert "users" in result.tables
|
|
|
|
def test_highlight_extract_join_table(self):
|
|
"""应从 JOIN 子句提取表名。"""
|
|
result = SQLAssistant.highlight_sql("SELECT * FROM users u JOIN orders o ON u.id = o.user_id")
|
|
assert "users" in result.tables
|
|
assert "orders" in result.tables
|
|
|
|
def test_highlight_delete_risk(self):
|
|
"""DELETE 应被标记为高风险。"""
|
|
result = SQLAssistant.highlight_sql("DELETE FROM users WHERE id = 1")
|
|
assert len(result.risks) > 0
|
|
risk_ops = [r["operation"] for r in result.risks]
|
|
assert "DELETE" in risk_ops
|
|
|
|
def test_highlight_update_risk(self):
|
|
"""UPDATE 应被标记为中等风险。"""
|
|
result = SQLAssistant.highlight_sql("UPDATE users SET name='x' WHERE id=1")
|
|
assert len(result.risks) > 0
|
|
risk_ops = [r["operation"] for r in result.risks]
|
|
assert "UPDATE" in risk_ops
|
|
|
|
def test_highlight_drop_critical_risk(self):
|
|
"""DROP 应被标记为关键风险。"""
|
|
result = SQLAssistant.highlight_sql("DROP TABLE users")
|
|
assert len(result.risks) > 0
|
|
critical = [r for r in result.risks if r["level"] == "critical"]
|
|
assert len(critical) > 0
|
|
|
|
def test_highlight_update_no_where_critical(self):
|
|
"""缺少 WHERE 的 UPDATE 应被标记为关键风险。"""
|
|
result = SQLAssistant.highlight_sql("UPDATE users SET name='x'")
|
|
assert len(result.risks) > 0
|
|
critical = [r for r in result.risks if r["level"] == "critical"]
|
|
assert len(critical) > 0
|
|
|
|
def test_highlight_delete_no_where_critical(self):
|
|
"""缺少 WHERE 的 DELETE 应被标记为关键风险。"""
|
|
result = SQLAssistant.highlight_sql("DELETE FROM users")
|
|
assert len(result.risks) > 0
|
|
critical = [r for r in result.risks if r["level"] == "critical"]
|
|
assert len(critical) > 0
|
|
|
|
def test_highlight_select_no_risk(self):
|
|
"""SELECT 语句不应有风险标记。"""
|
|
result = SQLAssistant.highlight_sql("SELECT id, name FROM users LIMIT 10")
|
|
assert len(result.risks) == 0
|
|
|
|
def test_highlight_produces_html(self):
|
|
"""高亮结果应包含 HTML 标签。"""
|
|
result = SQLAssistant.highlight_sql("SELECT * FROM users")
|
|
assert "sql-keyword" in result.highlighted
|
|
|
|
|
|
# ===== EXPLAIN 解析 =====
|
|
|
|
|
|
class TestExplainParse:
|
|
"""MySQL EXPLAIN 结果解析测试。"""
|
|
|
|
def test_parse_empty_explain(self):
|
|
"""空 EXPLAIN 应返回警告。"""
|
|
result = SQLAssistant.parse_explain([])
|
|
assert result.is_simple is True
|
|
assert len(result.warnings) > 0
|
|
|
|
def test_parse_simple_index_scan(self):
|
|
"""索引扫描的 EXPLAIN 应被正确解析。"""
|
|
rows = [{
|
|
"table": "users",
|
|
"type": "ref",
|
|
"key": "idx_status",
|
|
"rows": 100,
|
|
"Extra": "",
|
|
}]
|
|
result = SQLAssistant.parse_explain(rows)
|
|
assert result.has_index is True
|
|
assert result.has_full_scan is False
|
|
assert result.estimated_rows == 100
|
|
|
|
def test_parse_full_table_scan(self):
|
|
"""全表扫描的 EXPLAIN 应标记警告。"""
|
|
rows = [{
|
|
"table": "orders",
|
|
"type": "ALL",
|
|
"key": None,
|
|
"rows": 500000,
|
|
"Extra": "",
|
|
}]
|
|
result = SQLAssistant.parse_explain(rows)
|
|
assert result.has_full_scan is True
|
|
assert result.has_index is False
|
|
assert result.estimated_rows == 500000
|
|
assert len(result.warnings) > 0
|
|
|
|
def test_parse_filesort_warning(self):
|
|
"""Using filesort 应产生警告。"""
|
|
rows = [{
|
|
"table": "users",
|
|
"type": "ref",
|
|
"key": "idx_status",
|
|
"rows": 100,
|
|
"Extra": "Using filesort",
|
|
}]
|
|
result = SQLAssistant.parse_explain(rows)
|
|
filesort_warnings = [w for w in result.warnings if "filesort" in w.lower() or "排序" in w]
|
|
assert len(filesort_warnings) > 0
|
|
|
|
def test_parse_temporary_table_warning(self):
|
|
"""Using temporary 应产生警告。"""
|
|
rows = [{
|
|
"table": "users",
|
|
"type": "ref",
|
|
"key": "idx_status",
|
|
"rows": 100,
|
|
"Extra": "Using temporary",
|
|
}]
|
|
result = SQLAssistant.parse_explain(rows)
|
|
temp_warnings = [w for w in result.warnings if "临时表" in w or "temporary" in w.lower()]
|
|
assert len(temp_warnings) > 0
|
|
|
|
def test_parse_large_rows_warning(self):
|
|
"""预估行数过大应产生警告。"""
|
|
rows = [{
|
|
"table": "big_table",
|
|
"type": "ALL",
|
|
"key": None,
|
|
"rows": 500000,
|
|
"Extra": "",
|
|
}]
|
|
result = SQLAssistant.parse_explain(rows)
|
|
large_warnings = [w for w in result.warnings if "较大" in w or "较慢" in w]
|
|
assert len(large_warnings) > 0
|
|
|
|
def test_parse_multi_table_join(self):
|
|
"""多表 JOIN 应返回非简单查询。"""
|
|
rows = [
|
|
{"table": "users", "type": "ref", "key": "PRIMARY", "rows": 1, "Extra": ""},
|
|
{"table": "orders", "type": "ref", "key": "idx_user", "rows": 10, "Extra": ""},
|
|
]
|
|
result = SQLAssistant.parse_explain(rows)
|
|
assert result.is_simple is False
|
|
assert result.has_index is True
|
|
assert result.estimated_rows == 11
|
|
|
|
def test_parse_produces_human_readable(self):
|
|
"""应生成人类可读的执行计划描述。"""
|
|
rows = [{
|
|
"table": "users",
|
|
"type": "ref",
|
|
"key": "idx_name",
|
|
"rows": 50,
|
|
"Extra": "",
|
|
}]
|
|
result = SQLAssistant.parse_explain(rows)
|
|
assert result.human_readable != ""
|
|
assert "users" in result.human_readable
|
|
|
|
def test_parse_case_insensitive_keys(self):
|
|
"""应处理大写的 EXPLAIN 键名。"""
|
|
rows = [{
|
|
"TABLE": "users",
|
|
"TYPE": "ALL",
|
|
"KEY": None,
|
|
"ROWS": 1000,
|
|
"EXTRA": "",
|
|
}]
|
|
result = SQLAssistant.parse_explain(rows)
|
|
assert result.estimated_rows == 1000
|
|
|
|
|
|
# ===== 查询建议 =====
|
|
|
|
|
|
class TestQuerySuggestions:
|
|
"""SQL 查询建议测试。"""
|
|
|
|
def test_select_star_suggestion(self):
|
|
"""SELECT * 应触发优化建议。"""
|
|
suggestions = SQLAssistant.get_query_suggestions("SELECT * FROM users")
|
|
types = [s["type"] for s in suggestions]
|
|
assert "optimization" in types
|
|
|
|
def test_no_limit_suggestion(self):
|
|
"""缺少 LIMIT 应触发安全建议。"""
|
|
suggestions = SQLAssistant.get_query_suggestions("SELECT id, name FROM users WHERE status = 'active'")
|
|
msgs = [s["message"] for s in suggestions]
|
|
assert any("LIMIT" in m for m in msgs)
|
|
|
|
def test_order_by_rand_suggestion(self):
|
|
"""ORDER BY RAND() 应触发优化建议。"""
|
|
suggestions = SQLAssistant.get_query_suggestions("SELECT * FROM users ORDER BY RAND() LIMIT 5")
|
|
msgs = [s["message"] for s in suggestions]
|
|
assert any("RAND" in m for m in msgs)
|
|
|
|
def test_update_no_where_suggestion(self):
|
|
"""缺少 WHERE 的 UPDATE 应触发危险建议。"""
|
|
suggestions = SQLAssistant.get_query_suggestions("UPDATE users SET name='x'")
|
|
danger = [s for s in suggestions if s["type"] == "danger"]
|
|
assert len(danger) > 0
|
|
|
|
def test_delete_no_where_suggestion(self):
|
|
"""缺少 WHERE 的 DELETE 应触发危险建议。"""
|
|
suggestions = SQLAssistant.get_query_suggestions("DELETE FROM users")
|
|
danger = [s for s in suggestions if s["type"] == "danger"]
|
|
assert len(danger) > 0
|
|
|
|
def test_complex_query_comment_suggestion(self):
|
|
"""无注释的复杂查询应触发风格建议。"""
|
|
long_sql = "SELECT * FROM users WHERE " + " AND ".join(["column_name_" + str(i) + " = 'value_number_" + str(i) + "'" for i in range(30)])
|
|
suggestions = SQLAssistant.get_query_suggestions(long_sql)
|
|
style = [s for s in suggestions if s["type"] == "style"]
|
|
assert len(style) > 0
|
|
|
|
def test_many_joins_suggestion(self):
|
|
"""多个 JOIN 应触发优化建议。"""
|
|
sql = "SELECT * FROM a JOIN b ON a.id = b.aid JOIN c ON b.id = c.bid JOIN d ON c.id = d.cid JOIN e ON d.id = e.did"
|
|
suggestions = SQLAssistant.get_query_suggestions(sql)
|
|
join_suggestions = [s for s in suggestions if "JOIN" in s["message"]]
|
|
assert len(join_suggestions) > 0
|
|
|
|
def test_clean_select_no_suggestions(self):
|
|
"""干净的 SELECT 查询不应有危险建议。"""
|
|
suggestions = SQLAssistant.get_query_suggestions("SELECT id, name FROM users WHERE status = 'active' LIMIT 10")
|
|
danger = [s for s in suggestions if s["type"] == "danger"]
|
|
assert len(danger) == 0
|
|
|
|
|
|
# ===== 统一分析入口 =====
|
|
|
|
|
|
class TestAnalyze:
|
|
"""SQLAssistant.analyze 统一入口测试。"""
|
|
|
|
def test_analyze_returns_all_fields(self):
|
|
"""应返回所有分析字段。"""
|
|
result = SQLAssistant.analyze("SELECT * FROM users LIMIT 10")
|
|
assert "formatted" in result
|
|
assert "is_valid" in result
|
|
assert "highlighted" in result
|
|
assert "keywords" in result
|
|
assert "tables" in result
|
|
assert "risks" in result
|
|
assert "suggestions" in result
|
|
assert "risk_count" in result
|
|
assert "suggestion_count" in result
|
|
|
|
def test_analyze_select_star(self):
|
|
"""SELECT * 分析应有优化建议。"""
|
|
result = SQLAssistant.analyze("SELECT * FROM users")
|
|
assert result["suggestion_count"] > 0
|
|
|
|
def test_analyze_dangerous_update(self):
|
|
"""危险 UPDATE 分析应有风险标记和建议。"""
|
|
result = SQLAssistant.analyze("UPDATE users SET name='x'")
|
|
assert result["risk_count"] > 0
|
|
|
|
def test_analyze_with_db_type(self):
|
|
"""应支持指定数据库类型。"""
|
|
result = SQLAssistant.analyze("SELECT * FROM users", db_type="postgres")
|
|
assert result["is_valid"] is True
|