logan
4c9427e546
- P0: 实例访问控制全线接入权限组 (829测试) - P1: permissions字段激活 + 操作级权限检查 (843测试) - 前端: 删除数据权限页面, 权限组合并为唯一入口 - 新增: CodeMirror编辑器 + 查询模板 + 慢查询分析 - 新增: MongoDB管理页面 + Dashboard增强 - 修复: Pydantic V2迁移 + 前端TS错误修复
715 lines
29 KiB
Python
715 lines
29 KiB
Python
"""Tests for SQL Audit Engine V2 — 可配置规则引擎 (40+ 规则覆盖测试).
|
||
|
||
测试覆盖:
|
||
- DDL 规则 (12条): 主键、注释、索引、类型等
|
||
- DML 规则 (8条): WHERE、LIMIT、INSERT、REPLACE 等
|
||
- DQL 规则 (5条): SELECT *、LIMIT、函数、排序等
|
||
- 命名规范 (5条): 表名、字段名、索引名等
|
||
- 性能规则 (8条): 全表扫描、IN、LIKE、JOIN 等
|
||
- 安全规则 (5条): 敏感字段、密码、UNION、注释等
|
||
"""
|
||
|
||
import pytest
|
||
from app.services.audit_engine_v2 import (
|
||
SQLAuditEngineV2,
|
||
AuditReportV2,
|
||
AuditFindingV2,
|
||
Severity,
|
||
RuleCategory,
|
||
AuditRuleV2,
|
||
)
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 引擎基础测试
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TestSQLAuditEngineV2:
|
||
"""SQLAuditEngineV2 引擎核心功能测试。"""
|
||
|
||
@pytest.fixture
|
||
def engine(self):
|
||
"""返回不带数据库的同步引擎实例。"""
|
||
return SQLAuditEngineV2(db_session=None, env="development")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_audit_returns_report(self, engine):
|
||
"""audit() 应返回 AuditReportV2 实例。"""
|
||
report = await engine.audit("SELECT id, name FROM users LIMIT 10;")
|
||
assert isinstance(report, AuditReportV2)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clean_sql_passes(self, engine):
|
||
"""规范 SQL 应通过审核且评分为100。"""
|
||
report = await engine.audit("SELECT id, name FROM users WHERE id = 1 LIMIT 10;")
|
||
assert report.passed
|
||
assert report.score == 100
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_report_has_summary(self, engine):
|
||
"""审核报告应包含摘要。"""
|
||
report = await engine.audit("UPDATE users SET name = 'test' WHERE id = 1;")
|
||
assert report.summary
|
||
assert isinstance(report.summary, str)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_report_to_dict_backward_compat(self, engine):
|
||
"""to_dict() 应返回向后兼容的字典格式。"""
|
||
report = await engine.audit("SELECT * FROM users;")
|
||
d = report.to_dict()
|
||
assert "passed" in d
|
||
assert "score" in d
|
||
assert "findings" in d
|
||
assert "summary" in d
|
||
assert isinstance(d["findings"], list)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_error_reduces_score(self, engine):
|
||
"""ERROR 级别发现项应扣10分。"""
|
||
report = await engine.audit("UPDATE users SET status = 1;")
|
||
assert report.score < 100
|
||
assert not report.passed
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_db_type_filtering(self, engine):
|
||
"""规则应按 db_type 过滤。"""
|
||
report_mysql = await engine.audit("SELECT id FROM t LIMIT 10;", db_type="mysql")
|
||
report_postgres = await engine.audit("SELECT id FROM t LIMIT 10;", db_type="postgres")
|
||
# 两者都应通过（基本 SELECT）
|
||
assert report_mysql.passed
|
||
assert report_postgres.passed
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_sync_audit(self, engine):
|
||
"""同步审核应正常工作。"""
|
||
report = engine.audit_sync("SELECT id FROM users WHERE id = 1 LIMIT 10;")
|
||
assert report.passed
|
||
assert report.score == 100
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_rule_list(self, engine):
|
||
"""获取规则列表应返回非空列表。"""
|
||
rules = await engine.get_rule_list()
|
||
assert len(rules) >= 40 # 至少40条规则
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# DDL 规则测试
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TestDDLRules:
|
||
"""DDL 规则测试 (DDL-001 ~ DDL-012)。"""
|
||
|
||
@pytest.fixture
|
||
def engine(self):
|
||
return SQLAuditEngineV2(db_session=None, env="development")
|
||
|
||
# DDL-001: 建表必须包含主键
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_001_no_primary_key(self, engine):
|
||
report = await engine.audit("CREATE TABLE users (name VARCHAR(50));")
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-001"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_001_has_primary_key(self, engine):
|
||
report = await engine.audit("CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(50));")
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-001"]
|
||
assert len(findings) == 0
|
||
|
||
# DDL-002: 建表必须添加表注释
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_002_no_comment(self, engine):
|
||
report = await engine.audit("CREATE TABLE users (id INT PRIMARY KEY);")
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-002"]
|
||
assert len(findings) >= 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_002_has_comment(self, engine):
|
||
report = await engine.audit(
|
||
"CREATE TABLE users (id INT PRIMARY KEY) COMMENT='用户表';"
|
||
)
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-002"]
|
||
assert len(findings) == 0
|
||
|
||
# DDL-004: 禁止 DROP TABLE
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_004_drop_table(self, engine):
|
||
report = await engine.audit("DROP TABLE users;")
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-004"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
# DDL-005: 禁止 TRUNCATE TABLE
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_005_truncate(self, engine):
|
||
report = await engine.audit("TRUNCATE TABLE users;")
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-005"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
# DDL-006: 单表索引数量不得超过12个
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_006_too_many_indexes(self, engine):
|
||
# 创建15个索引
|
||
idx_defs = ", ".join([f"KEY idx_{i} (col{i})" for i in range(15)])
|
||
sql = f"CREATE TABLE t (id INT PRIMARY KEY, {', '.join([f'col{i} INT' for i in range(15)])}, {idx_defs});"
|
||
report = await engine.audit(sql)
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-006"]
|
||
assert len(findings) >= 1
|
||
|
||
# DDL-007: 单索引字段数不得超过8个
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_007_too_many_index_columns(self, engine):
|
||
cols = ", ".join([f"col{i}" for i in range(10)])
|
||
sql = f"CREATE TABLE t (id INT PRIMARY KEY, KEY idx_big ({cols}));"
|
||
report = await engine.audit(sql)
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-007"]
|
||
assert len(findings) >= 1
|
||
|
||
# DDL-008: 禁止使用 FLOAT/DOUBLE
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_008_float_type(self, engine):
|
||
report = await engine.audit(
|
||
"CREATE TABLE t (id INT PRIMARY KEY, price FLOAT);"
|
||
)
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-008"]
|
||
assert len(findings) >= 1
|
||
|
||
# DDL-009: CHAR 类型长度不得超过10
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_009_char_too_long(self, engine):
|
||
report = await engine.audit(
|
||
"CREATE TABLE t (id INT PRIMARY KEY, code CHAR(20));"
|
||
)
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-009"]
|
||
assert len(findings) >= 1
|
||
|
||
# DDL-010: 表名长度不得超过45字符
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_010_long_table_name(self, engine):
|
||
long_name = "a" * 50
|
||
report = await engine.audit(
|
||
f"CREATE TABLE {long_name} (id INT PRIMARY KEY);"
|
||
)
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-010"]
|
||
assert len(findings) >= 1
|
||
|
||
# DDL-011: 禁止外键约束
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_011_foreign_key(self, engine):
|
||
sql = """CREATE TABLE orders (
|
||
id INT PRIMARY KEY,
|
||
user_id INT,
|
||
FOREIGN KEY (user_id) REFERENCES users(id)
|
||
);"""
|
||
report = await engine.audit(sql)
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-011"]
|
||
assert len(findings) >= 1
|
||
|
||
# DDL-012: 必须包含 create_time/update_time
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_012_missing_audit_fields(self, engine):
|
||
report = await engine.audit(
|
||
"CREATE TABLE t (id INT PRIMARY KEY, name VARCHAR(50));"
|
||
)
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-012"]
|
||
assert len(findings) >= 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ddl_012_has_audit_fields(self, engine):
|
||
report = await engine.audit(
|
||
"CREATE TABLE t (id INT PRIMARY KEY, name VARCHAR(50), create_time DATETIME, update_time DATETIME);"
|
||
)
|
||
findings = [f for f in report.findings if f.rule_code == "DDL-012"]
|
||
assert len(findings) == 0
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# DML 规则测试
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TestDMLRules:
|
||
"""DML 规则测试 (DML-001 ~ DML-008)。"""
|
||
|
||
@pytest.fixture
|
||
def engine(self):
|
||
return SQLAuditEngineV2(db_session=None, env="development")
|
||
|
||
# DML-001: UPDATE/DELETE 必须包含 WHERE
|
||
@pytest.mark.asyncio
|
||
async def test_dml_001_update_no_where(self, engine):
|
||
report = await engine.audit("UPDATE users SET name = 'test';")
|
||
findings = [f for f in report.findings if f.rule_code == "DML-001"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dml_001_delete_no_where(self, engine):
|
||
report = await engine.audit("DELETE FROM users;")
|
||
findings = [f for f in report.findings if f.rule_code == "DML-001"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dml_001_update_with_where(self, engine):
|
||
report = await engine.audit("UPDATE users SET name = 'test' WHERE id = 1;")
|
||
findings = [f for f in report.findings if f.rule_code == "DML-001"]
|
||
assert len(findings) == 0
|
||
|
||
# DML-003: UPDATE/DELETE 禁止带 LIMIT
|
||
@pytest.mark.asyncio
|
||
async def test_dml_003_update_with_limit(self, engine):
|
||
report = await engine.audit("UPDATE users SET name = 'test' WHERE id = 1 LIMIT 10;")
|
||
findings = [f for f in report.findings if f.rule_code == "DML-003"]
|
||
assert len(findings) >= 1
|
||
|
||
# DML-007: 禁止 REPLACE INTO
|
||
@pytest.mark.asyncio
|
||
async def test_dml_007_replace_into(self, engine):
|
||
report = await engine.audit("REPLACE INTO users (id, name) VALUES (1, 'test');")
|
||
findings = [f for f in report.findings if f.rule_code == "DML-007"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
# DML-008: INSERT 必须指定字段列表
|
||
@pytest.mark.asyncio
|
||
async def test_dml_008_insert_no_columns(self, engine):
|
||
report = await engine.audit("INSERT INTO users VALUES (1, 'test');")
|
||
findings = [f for f in report.findings if f.rule_code == "DML-008"]
|
||
assert len(findings) >= 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dml_008_insert_with_columns(self, engine):
|
||
report = await engine.audit("INSERT INTO users (id, name) VALUES (1, 'test');")
|
||
findings = [f for f in report.findings if f.rule_code == "DML-008"]
|
||
assert len(findings) == 0
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# DQL 规则测试
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TestDQLRules:
|
||
"""DQL 规则测试 (DQL-001 ~ DQL-005)。"""
|
||
|
||
@pytest.fixture
|
||
def engine(self):
|
||
return SQLAuditEngineV2(db_session=None, env="development")
|
||
|
||
# DQL-001: 禁止 SELECT *
|
||
@pytest.mark.asyncio
|
||
async def test_dql_001_select_star(self, engine):
|
||
report = await engine.audit("SELECT * FROM users;")
|
||
findings = [f for f in report.findings if f.rule_code == "DQL-001"]
|
||
assert len(findings) >= 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dql_001_select_columns(self, engine):
|
||
report = await engine.audit("SELECT id, name FROM users;")
|
||
findings = [f for f in report.findings if f.rule_code == "DQL-001"]
|
||
assert len(findings) == 0
|
||
|
||
# DQL-002: SELECT 必须有 LIMIT
|
||
@pytest.mark.asyncio
|
||
async def test_dql_002_no_limit(self, engine):
|
||
report = await engine.audit("SELECT id, name FROM users;")
|
||
findings = [f for f in report.findings if f.rule_code == "DQL-002"]
|
||
assert len(findings) >= 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dql_002_has_limit(self, engine):
|
||
report = await engine.audit("SELECT id, name FROM users LIMIT 10;")
|
||
findings = [f for f in report.findings if f.rule_code == "DQL-002"]
|
||
assert len(findings) == 0
|
||
|
||
# DQL-003: 禁止在 WHERE 中使用函数
|
||
@pytest.mark.asyncio
|
||
async def test_dql_003_where_function(self, engine):
|
||
report = await engine.audit("SELECT id FROM users WHERE YEAR(created_at) = 2024;")
|
||
findings = [f for f in report.findings if f.rule_code == "DQL-003"]
|
||
# 只要检测到 WHERE + 函数模式
|
||
assert len(findings) >= 1
|
||
|
||
# DQL-005: ORDER BY 必须使用索引字段
|
||
@pytest.mark.asyncio
|
||
async def test_dql_005_order_by(self, engine):
|
||
report = await engine.audit("SELECT id FROM users ORDER BY name;")
|
||
findings = [f for f in report.findings if f.rule_code == "DQL-005"]
|
||
assert len(findings) >= 1
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 命名规范测试
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TestNamingRules:
|
||
"""命名规范测试 (NAME-001 ~ NAME-005)。"""
|
||
|
||
@pytest.fixture
|
||
def engine(self):
|
||
return SQLAuditEngineV2(db_session=None, env="development")
|
||
|
||
# NAME-001: 表名必须小写下划线命名
|
||
@pytest.mark.asyncio
|
||
async def test_name_001_uppercase_table(self, engine):
|
||
report = await engine.audit("CREATE TABLE UserInfo (id INT PRIMARY KEY);")
|
||
findings = [f for f in report.findings if f.rule_code == "NAME-001"]
|
||
assert len(findings) >= 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_name_001_snake_case_table(self, engine):
|
||
report = await engine.audit("CREATE TABLE user_info (id INT PRIMARY KEY);")
|
||
findings = [f for f in report.findings if f.rule_code == "NAME-001"]
|
||
assert len(findings) == 0
|
||
|
||
# NAME-002: 字段名必须小写下划线命名
|
||
@pytest.mark.asyncio
|
||
async def test_name_002_uppercase_column(self, engine):
|
||
report = await engine.audit(
|
||
"CREATE TABLE t (id INT PRIMARY KEY, UserName VARCHAR(50));"
|
||
)
|
||
findings = [f for f in report.findings if f.rule_code == "NAME-002"]
|
||
assert len(findings) >= 1
|
||
|
||
# NAME-004: 主键名必须为 id 或 pk_ 开头
|
||
@pytest.mark.asyncio
|
||
async def test_name_004_bad_pk_name(self, engine):
|
||
report = await engine.audit(
|
||
"CREATE TABLE t (user_id INT PRIMARY KEY, name VARCHAR(50));"
|
||
)
|
||
findings = [f for f in report.findings if f.rule_code == "NAME-004"]
|
||
assert len(findings) >= 1
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 性能规则测试
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TestPerfRules:
|
||
"""性能规则测试 (PERF-001 ~ PERF-008)。"""
|
||
|
||
@pytest.fixture
|
||
def engine(self):
|
||
return SQLAuditEngineV2(db_session=None, env="development")
|
||
|
||
# PERF-002: IN 子句值数量不得超过100个
|
||
@pytest.mark.asyncio
|
||
async def test_perf_002_large_in(self, engine):
|
||
values = ", ".join([str(i) for i in range(150)])
|
||
sql = f"SELECT id FROM users WHERE id IN ({values});"
|
||
report = await engine.audit(sql)
|
||
findings = [f for f in report.findings if f.rule_code == "PERF-002"]
|
||
assert len(findings) >= 1
|
||
|
||
# PERF-003: 禁止 LIKE '%xxx%'
|
||
@pytest.mark.asyncio
|
||
async def test_perf_003_full_like(self, engine):
|
||
report = await engine.audit("SELECT id FROM users WHERE name LIKE '%test%';")
|
||
findings = [f for f in report.findings if f.rule_code == "PERF-003"]
|
||
assert len(findings) >= 1
|
||
|
||
# PERF-005: 禁止 ORDER BY RAND()
|
||
@pytest.mark.asyncio
|
||
async def test_perf_005_order_by_rand(self, engine):
|
||
report = await engine.audit("SELECT id FROM users ORDER BY RAND() LIMIT 10;")
|
||
findings = [f for f in report.findings if f.rule_code == "PERF-005"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
# PERF-007: 大表禁止 OFFSET 分页
|
||
@pytest.mark.asyncio
|
||
async def test_perf_007_offset(self, engine):
|
||
report = await engine.audit("SELECT id FROM users LIMIT 10 OFFSET 1000;")
|
||
findings = [f for f in report.findings if f.rule_code == "PERF-007"]
|
||
assert len(findings) >= 1
|
||
|
||
# PERF-008: 禁止 SELECT DISTINCT
|
||
@pytest.mark.asyncio
|
||
async def test_perf_008_distinct(self, engine):
|
||
report = await engine.audit("SELECT DISTINCT name FROM users;")
|
||
findings = [f for f in report.findings if f.rule_code == "PERF-008"]
|
||
assert len(findings) >= 1
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 安全规则测试
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TestSecurityRules:
|
||
"""安全规则测试 (SEC-001 ~ SEC-005)。"""
|
||
|
||
@pytest.fixture
|
||
def engine(self):
|
||
return SQLAuditEngineV2(db_session=None, env="development")
|
||
|
||
# SEC-001: 禁止查询敏感字段
|
||
@pytest.mark.asyncio
|
||
async def test_sec_001_sensitive_fields(self, engine):
|
||
report = await engine.audit("SELECT id, password, secret FROM users;")
|
||
findings = [f for f in report.findings if f.rule_code == "SEC-001"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
# SEC-003: 禁止 UNION
|
||
@pytest.mark.asyncio
|
||
async def test_sec_003_union(self, engine):
|
||
report = await engine.audit("SELECT id FROM users UNION SELECT id FROM orders;")
|
||
findings = [f for f in report.findings if f.rule_code == "SEC-003"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
# SEC-004: 禁止注释符
|
||
@pytest.mark.asyncio
|
||
async def test_sec_004_comment_dash(self, engine):
|
||
report = await engine.audit("SELECT id FROM users WHERE id = 1; -- comment")
|
||
findings = [f for f in report.findings if f.rule_code == "SEC-004"]
|
||
assert len(findings) >= 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_sec_004_comment_block(self, engine):
|
||
report = await engine.audit("SELECT id /* comment */ FROM users;")
|
||
findings = [f for f in report.findings if f.rule_code == "SEC-004"]
|
||
assert len(findings) >= 1
|
||
|
||
# SEC-005: 禁止多语句
|
||
@pytest.mark.asyncio
|
||
async def test_sec_005_multi_statement(self, engine):
|
||
report = await engine.audit("UPDATE users SET name = 'a'; DELETE FROM logs;")
|
||
findings = [f for f in report.findings if f.rule_code == "SEC-005"]
|
||
assert len(findings) >= 1
|
||
assert findings[0].level == Severity.ERROR
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 数据类测试
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TestAuditDataClasses:
|
||
"""数据类单元测试。"""
|
||
|
||
def test_audit_rule_v2_dataclass(self):
|
||
"""AuditRuleV2 数据类应正常工作。"""
|
||
rule = AuditRuleV2(
|
||
code="DDL-001",
|
||
name="测试规则",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.ERROR,
|
||
)
|
||
assert rule.code == "DDL-001"
|
||
assert rule.compiled_pattern is None
|
||
|
||
def test_audit_rule_v2_pattern_compile(self):
|
||
"""AuditRuleV2 应能编译正则表达式。"""
|
||
rule = AuditRuleV2(
|
||
code="TEST-001",
|
||
name="测试规则",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
pattern=r"^test_",
|
||
)
|
||
assert rule.compiled_pattern is not None
|
||
assert rule.compiled_pattern.match("test_table")
|
||
|
||
def test_audit_finding_v2(self):
|
||
"""AuditFindingV2 数据类应正常工作。"""
|
||
finding = AuditFindingV2(
|
||
rule_code="DDL-001",
|
||
rule_name="测试规则",
|
||
level=Severity.ERROR,
|
||
message="测试消息",
|
||
position=10,
|
||
)
|
||
assert finding.rule_code == "DDL-001"
|
||
assert finding.position == 10
|
||
|
||
def test_audit_report_v2_score_calculation(self):
|
||
"""AuditReportV2 评分计算应正确。"""
|
||
report = AuditReportV2(sql="SELECT * FROM t;")
|
||
report.findings = [
|
||
AuditFindingV2("DQL-001", "禁止SELECT *", Severity.INFO, "msg1"),
|
||
AuditFindingV2("DDL-004", "禁止DROP TABLE", Severity.ERROR, "msg2"),
|
||
]
|
||
assert report.score == 88 # 100 - 10(ERROR) - 2(INFO) = 88
|
||
assert not report.passed
|
||
|
||
def test_audit_report_v2_empty_findings(self):
|
||
"""无发现项时评分应为100且通过。"""
|
||
report = AuditReportV2(sql="SELECT id FROM t WHERE id = 1;")
|
||
assert report.score == 100
|
||
assert report.passed
|
||
assert report.summary == "审核通过，未发现违规项"
|
||
|
||
def test_audit_report_backward_compat(self):
|
||
"""AuditReportV2 应向后兼容 AuditReport 格式。"""
|
||
report = AuditReportV2(sql="UPDATE t SET a = 1;")
|
||
report.findings.append(
|
||
AuditFindingV2("DML-001", "缺少WHERE", Severity.ERROR, "msg")
|
||
)
|
||
d = report.to_dict()
|
||
assert "passed" in d
|
||
assert "score" in d
|
||
assert "findings" in d
|
||
# 检查旧版格式兼容性
|
||
assert d["findings"][0]["rule"] == "DML-001"
|
||
assert d["findings"][0]["level"] == "error"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# API 集成测试
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TestAuditEngineV2Integration:
|
||
"""V2 引擎与 API 集成测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ticket_creation_uses_v2_engine(self, client, db, admin_token, test_instance):
|
||
"""创建工单时应使用 V2 引擎进行审核。"""
|
||
from tests.conftest import auth_header
|
||
resp = await client.post(
|
||
"/api/v1/tickets",
|
||
json={
|
||
"title": "V2引擎测试工单",
|
||
"db_type": "mysql",
|
||
"instance_id": test_instance.id,
|
||
"target_database": "test_db",
|
||
"sql_content": "SELECT * FROM users WHERE id = 1;",
|
||
"reason": "测试V2审核引擎",
|
||
"priority": "medium",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 201
|
||
data = resp.json()
|
||
assert "audit_report" in data
|
||
report = data["audit_report"]
|
||
assert "score" in report
|
||
assert "findings" in report
|
||
assert "summary" in report
|
||
# SELECT * 应触发 DQL-001 (INFO)
|
||
rules = [f["rule"] for f in report["findings"]]
|
||
assert "DQL-001" in rules or "DQL-002" in rules
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_audit_endpoint_returns_v2_format(self, client, db, admin_token, test_instance):
|
||
"""审核接口应返回 V2 格式报告。"""
|
||
from tests.conftest import auth_header
|
||
resp = await client.post(
|
||
"/api/v1/tickets/audit",
|
||
json={
|
||
"sql_content": "UPDATE users SET status = 1 WHERE id = 1;",
|
||
"instance_id": test_instance.id,
|
||
"database": "test_db",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "score" in data
|
||
assert "passed" in data
|
||
assert "summary" in data
|
||
assert "findings" in data
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_audit_endpoint_detects_dangerous_sql(self, client, db, admin_token, test_instance):
|
||
"""审核接口应能检测危险 SQL。"""
|
||
from tests.conftest import auth_header
|
||
resp = await client.post(
|
||
"/api/v1/tickets/audit",
|
||
json={
|
||
"sql_content": "DROP TABLE users;",
|
||
"instance_id": test_instance.id,
|
||
"database": "test_db",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["passed"] is False
|
||
rules = [f["rule"] for f in data["findings"]]
|
||
assert "DDL-004" in rules
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_seed_v2_endpoint(self, client, db, admin_token):
|
||
"""seed-v2 接口应能初始化所有 V2 规则。"""
|
||
from tests.conftest import auth_header
|
||
resp = await client.post(
|
||
"/api/v1/audit-rules/seed-v2",
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert isinstance(data, list)
|
||
assert len(data) >= 40 # 应创建至少40条规则
|
||
|
||
# 再次调用应跳过已存在的规则
|
||
resp2 = await client.post(
|
||
"/api/v1/audit-rules/seed-v2",
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp2.status_code == 200
|
||
data2 = resp2.json()
|
||
assert len(data2) == 0 # 都已存在，不再创建
|
||
|
||
|
||
class TestAllRulesRegistered:
|
||
"""验证所有要求的规则都已注册。"""
|
||
|
||
def test_ddl_rules_registered(self):
|
||
"""DDL 规则应全部注册。"""
|
||
from app.services.audit_engine_v2 import get_all_registered_rules
|
||
rules = get_all_registered_rules()
|
||
codes = [r.code for r in rules]
|
||
for i in range(1, 13):
|
||
assert f"DDL-{i:03d}" in codes, f"DDL-{i:03d} 未注册"
|
||
|
||
def test_dml_rules_registered(self):
|
||
"""DML 规则应全部注册。"""
|
||
from app.services.audit_engine_v2 import get_all_registered_rules
|
||
rules = get_all_registered_rules()
|
||
codes = [r.code for r in rules]
|
||
for i in range(1, 9):
|
||
assert f"DML-{i:03d}" in codes, f"DML-{i:03d} 未注册"
|
||
|
||
def test_dql_rules_registered(self):
|
||
"""DQL 规则应全部注册。"""
|
||
from app.services.audit_engine_v2 import get_all_registered_rules
|
||
rules = get_all_registered_rules()
|
||
codes = [r.code for r in rules]
|
||
for i in range(1, 6):
|
||
assert f"DQL-{i:03d}" in codes, f"DQL-{i:03d} 未注册"
|
||
|
||
def test_naming_rules_registered(self):
|
||
"""命名规范规则应全部注册。"""
|
||
from app.services.audit_engine_v2 import get_all_registered_rules
|
||
rules = get_all_registered_rules()
|
||
codes = [r.code for r in rules]
|
||
for i in range(1, 6):
|
||
assert f"NAME-{i:03d}" in codes, f"NAME-{i:03d} 未注册"
|
||
|
||
def test_perf_rules_registered(self):
|
||
"""性能规则应全部注册。"""
|
||
from app.services.audit_engine_v2 import get_all_registered_rules
|
||
rules = get_all_registered_rules()
|
||
codes = [r.code for r in rules]
|
||
for i in range(1, 9):
|
||
assert f"PERF-{i:03d}" in codes, f"PERF-{i:03d} 未注册"
|
||
|
||
def test_security_rules_registered(self):
|
||
"""安全规则应全部注册。"""
|
||
from app.services.audit_engine_v2 import get_all_registered_rules
|
||
rules = get_all_registered_rules()
|
||
codes = [r.code for r in rules]
|
||
for i in range(1, 6):
|
||
assert f"SEC-{i:03d}" in codes, f"SEC-{i:03d} 未注册"
|