feat: Phase2影响分析 — EXPLAIN预检+大表检测(简化版)+执行策略+风险评估table_sizes支持 (1141测试)

新增: - explain_preview.py: EXPLAIN预检服务(MySQL/PG), 影响行数预估, 全表扫描检测 - ticket.executor: 三种执行策略(stop_on_error/continue/transaction) - risk_evaluator: table_sizes参数, 大表ALTER/UPDATE/DELETE自动升级high - test_phase2_features.py: 22个测试(EXPLAIN+执行策略+大表检测) 简化: - 砍掉CTX-001上下文感知规则(复杂二次扫描), 合并大表检测到risk_evaluator - 审核引擎保持纯静态分析, 不再导入_RULE_REGISTRY做二次扫描模型: - Ticket加exec_strategy字段 - TicketSqlItem模型 - schema加sql_type/exec_strategy/risk_level/workflow_id
2026-05-25 11:02:51 +08:00
1 parent 7be0bf1a7c
commit f6f9142ce8
@@ -0,0 +1,189 @@
+"""Phase 2 影响分析测试 — EXPLAIN预检 + 部分失败策略 + 上下文感知规则。"""
+
+import pytest
+from unittest.mock import MagicMock, AsyncMock, patch
+
+
+# ===== EXPLAIN 预检: 表名提取 =====
+
+class TestExtractTableNames:
+    """SQL 表名提取测试。"""
+
+    def test_update_table(self):
+        from app.services.explain_preview import _extract_table_names
+        assert "users" in _extract_table_names("UPDATE users SET x=1")
+
+    def test_delete_from_table(self):
+        from app.services.explain_preview import _extract_table_names
+        assert "users" in _extract_table_names("DELETE FROM users WHERE id=1")
+
+    def test_alter_table(self):
+        from app.services.explain_preview import _extract_table_names
+        assert "orders" in _extract_table_names("ALTER TABLE orders ADD COLUMN price DECIMAL")
+
+    def test_drop_table(self):
+        from app.services.explain_preview import _extract_table_names
+        assert "temp" in _extract_table_names("DROP TABLE temp")
+
+    def test_drop_table_if_exists(self):
+        from app.services.explain_preview import _extract_table_names
+        assert "temp" in _extract_table_names("DROP TABLE IF EXISTS temp")
+
+    def test_insert_into(self):
+        from app.services.explain_preview import _extract_table_names
+        assert "logs" in _extract_table_names("INSERT INTO logs (msg) VALUES ('hi')")
+
+    def test_select_from(self):
+        from app.services.explain_preview import _extract_table_names
+        names = _extract_table_names("SELECT * FROM users WHERE id=1")
+        assert "users" in names
+
+    def test_join_tables(self):
+        from app.services.explain_preview import _extract_table_names
+        names = _extract_table_names("SELECT * FROM users JOIN orders ON users.id = orders.user_id")
+        assert "users" in names
+        assert "orders" in names
+
+    def test_create_table(self):
+        from app.services.explain_preview import _extract_table_names
+        assert "new_table" in _extract_table_names("CREATE TABLE new_table (id INT)")
+
+    def test_create_table_if_not_exists(self):
+        from app.services.explain_preview import _extract_table_names
+        assert "new_table" in _extract_table_names("CREATE TABLE IF NOT EXISTS new_table (id INT)")
+
+    def test_deduplication(self):
+        from app.services.explain_preview import _extract_table_names
+        names = _extract_table_names("SELECT * FROM users JOIN users AS u2 ON users.id = u2.id")
+        assert names.count("users") == 1
+
+    def test_empty_sql(self):
+        from app.services.explain_preview import _extract_table_names
+        assert _extract_table_names("") == []
+
+
+# ===== ExplainResult 测试 =====
+
+class TestExplainResult:
+    """EXPLAIN 结果数据结构测试。"""
+
+    def test_to_dict(self):
+        from app.services.explain_preview import ExplainResult
+        r = ExplainResult(
+            sql="SELECT * FROM users",
+            sql_type="DQL",
+            estimated_rows=1000,
+            full_table_scan=True,
+            warnings=["Using filesort"],
+        )
+        d = r.to_dict()
+        assert d["sql_type"] == "DQL"
+        assert d["estimated_rows"] == 1000
+        assert d["full_table_scan"] is True
+        assert "Using filesort" in d["warnings"]
+
+    def test_error_field(self):
+        from app.services.explain_preview import ExplainResult
+        r = ExplainResult(error="Connection refused")
+        d = r.to_dict()
+        assert d["error"] == "Connection refused"
+
+
+# ===== 执行策略测试 =====
+
+class TestExecStrategy:
+    """部分失败策略测试。"""
+
+    def test_schema_default_strategy(self):
+        """TicketCreate schema 默认策略是 stop_on_error。"""
+        from app.schemas.ticket import TicketCreate
+        data = TicketCreate(
+            title="测试工单12345",
+            db_type="mysql",
+            instance_id=1,
+            target_database="test_db",
+            sql_content="SELECT 1",
+            reason="测试原因abcde",
+        )
+        assert data.exec_strategy == "stop_on_error"
+
+    def test_schema_continue_strategy(self):
+        from app.schemas.ticket import TicketCreate
+        data = TicketCreate(
+            title="测试工单12345",
+            db_type="mysql",
+            instance_id=1,
+            target_database="test_db",
+            sql_content="SELECT 1",
+            reason="测试原因abcde",
+            exec_strategy="continue",
+        )
+        assert data.exec_strategy == "continue"
+
+    def test_schema_transaction_strategy(self):
+        from app.schemas.ticket import TicketCreate
+        data = TicketCreate(
+            title="测试工单12345",
+            db_type="mysql",
+            instance_id=1,
+            target_database="test_db",
+            sql_content="SELECT 1",
+            reason="测试原因abcde",
+            exec_strategy="transaction",
+        )
+        assert data.exec_strategy == "transaction"
+
+
+# ===== 上下文感知规则测试 =====
+
+class TestRiskEvaluatorTableSizes:
+    """risk_evaluator 大表检测测试（合并自原 CTX-001）。"""
+
+    def test_alter_large_table_upgraded_to_high(self):
+        """大表 ALTER 升级为 high。"""
+        from app.services.risk_evaluator import evaluate_risk_level
+        level = evaluate_risk_level(
+            "ALTER TABLE big_table ADD COLUMN status INT",
+            db_type="mysql",
+            table_sizes={"big_table": 500000},
+        )
+        assert level == "high"
+
+    def test_alter_small_table_not_upgraded(self):
+        """小表 ALTER 保持 medium。"""
+        from app.services.risk_evaluator import evaluate_risk_level
+        level = evaluate_risk_level(
+            "ALTER TABLE users ADD COLUMN age INT",
+            db_type="mysql",
+            table_sizes={"users": 100},
+        )
+        assert level == "medium"
+
+    def test_update_large_table_upgraded_to_high(self):
+        """大表 UPDATE 无 WHERE 升级为 high。"""
+        from app.services.risk_evaluator import evaluate_risk_level
+        level = evaluate_risk_level(
+            "UPDATE big_table SET status = 1",
+            db_type="mysql",
+            table_sizes={"big_table": 200000},
+        )
+        assert level == "high"
+
+    def test_no_table_sizes_no_upgrade(self):
+        """无 table_sizes 时不升级。"""
+        from app.services.risk_evaluator import evaluate_risk_level
+        level = evaluate_risk_level(
+            "ALTER TABLE users ADD COLUMN age INT",
+            db_type="mysql",
+        )
+        assert level == "medium"
+
+    def test_select_large_table_stays_low(self):
+        """SELECT 大表不升级。"""
+        from app.services.risk_evaluator import evaluate_risk_level
+        level = evaluate_risk_level(
+            "SELECT * FROM big_table",
+            db_type="mysql",
+            table_sizes={"big_table": 500000},
+        )
+        assert level == "low"