feat: Phase 4 - SQL AST analyzer, audit rules CRUD, Alembic, Docker (128 tests)

backend/tests/test_audit_rules.py

@@ -0,0 +1,165 @@
+"""Tests for Audit Rule CRUD API (Phase 4.2)."""
+import pytest
+from tests.conftest import auth_header
+
+
+class TestListAuditRules:
+    @pytest.mark.asyncio
+    async def test_list_rules_admin(self, client, admin_token):
+        resp = await client.get("/api/v1/audit-rules", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+        assert isinstance(resp.json(), list)
+
+    @pytest.mark.asyncio
+    async def test_list_rules_dba(self, client, dba_token):
+        resp = await client.get("/api/v1/audit-rules", headers=auth_header(dba_token))
+        assert resp.status_code == 200
+
+    @pytest.mark.asyncio
+    async def test_list_rules_developer_forbidden(self, client, developer_token):
+        resp = await client.get("/api/v1/audit-rules", headers=auth_header(developer_token))
+        assert resp.status_code == 403
+
+
+class TestCreateAuditRule:
+    @pytest.mark.asyncio
+    async def test_create_rule_success(self, client, admin_token):
+        resp = await client.post(
+            "/api/v1/audit-rules",
+            json={
+                "name": "no_select_star",
+                "description": "禁止 SELECT *",
+                "rule_type": "performance",
+                "severity": "warning",
+                "db_type": "mysql",
+                "pattern": r"SELECT\s+\*",
+                "message": "禁止使用 SELECT *，请明确指定字段",
+            },
+            headers=auth_header(admin_token),
+        )
+        assert resp.status_code == 201
+        data = resp.json()
+        assert data["name"] == "no_select_star"
+        assert data["rule_type"] == "performance"
+        assert data["is_enabled"] is True
+
+    @pytest.mark.asyncio
+    async def test_create_duplicate_name(self, client, admin_token):
+        payload = {
+            "name": "dup_rule",
+            "rule_type": "safety",
+            "severity": "error",
+            "message": "test",
+        }
+        await client.post("/api/v1/audit-rules", json=payload, headers=auth_header(admin_token))
+        resp = await client.post("/api/v1/audit-rules", json=payload, headers=auth_header(admin_token))
+        assert resp.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_create_rule_non_admin(self, client, dba_token):
+        resp = await client.post(
+            "/api/v1/audit-rules",
+            json={"name": "dba_rule", "rule_type": "syntax", "severity": "info", "message": "test"},
+            headers=auth_header(dba_token),
+        )
+        assert resp.status_code == 403
+
+
+class TestGetAuditRule:
+    @pytest.mark.asyncio
+    async def test_get_rule(self, client, admin_token):
+        create = await client.post(
+            "/api/v1/audit-rules",
+            json={"name": "get_test", "rule_type": "naming", "severity": "info", "message": "test"},
+            headers=auth_header(admin_token),
+        )
+        rule_id = create.json()["id"]
+        resp = await client.get(f"/api/v1/audit-rules/{rule_id}", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+        assert resp.json()["name"] == "get_test"
+
+    @pytest.mark.asyncio
+    async def test_get_nonexistent_rule(self, client, admin_token):
+        resp = await client.get("/api/v1/audit-rules/99999", headers=auth_header(admin_token))
+        assert resp.status_code == 404
+
+
+class TestUpdateAuditRule:
+    @pytest.mark.asyncio
+    async def test_update_rule(self, client, admin_token):
+        create = await client.post(
+            "/api/v1/audit-rules",
+            json={"name": "update_test", "rule_type": "safety", "severity": "warning", "message": "old msg"},
+            headers=auth_header(admin_token),
+        )
+        rule_id = create.json()["id"]
+        resp = await client.put(
+            f"/api/v1/audit-rules/{rule_id}",
+            json={"severity": "error", "message": "new msg"},
+            headers=auth_header(admin_token),
+        )
+        assert resp.status_code == 200
+        assert resp.json()["severity"] == "error"
+        assert resp.json()["message"] == "new msg"
+
+
+class TestDeleteAuditRule:
+    @pytest.mark.asyncio
+    async def test_delete_rule(self, client, admin_token):
+        create = await client.post(
+            "/api/v1/audit-rules",
+            json={"name": "delete_test", "rule_type": "syntax", "severity": "info", "message": "test"},
+            headers=auth_header(admin_token),
+        )
+        rule_id = create.json()["id"]
+        resp = await client.delete(f"/api/v1/audit-rules/{rule_id}", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+
+        # Verify deleted
+        get = await client.get(f"/api/v1/audit-rules/{rule_id}", headers=auth_header(admin_token))
+        assert get.status_code == 404
+
+
+class TestToggleAuditRule:
+    @pytest.mark.asyncio
+    async def test_toggle_rule(self, client, admin_token):
+        create = await client.post(
+            "/api/v1/audit-rules",
+            json={"name": "toggle_test", "rule_type": "safety", "severity": "warning", "message": "test"},
+            headers=auth_header(admin_token),
+        )
+        rule_id = create.json()["id"]
+        assert create.json()["is_enabled"] is True
+
+        # Toggle off
+        resp = await client.post(f"/api/v1/audit-rules/{rule_id}/toggle", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+        assert resp.json()["is_enabled"] is False
+
+        # Toggle back on
+        resp = await client.post(f"/api/v1/audit-rules/{rule_id}/toggle", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+        assert resp.json()["is_enabled"] is True
+
+
+class TestRuleStats:
+    @pytest.mark.asyncio
+    async def test_rule_stats(self, client, admin_token):
+        # Create some rules
+        await client.post(
+            "/api/v1/audit-rules",
+            json={"name": "stats_rule_1", "rule_type": "safety", "severity": "error", "message": "test"},
+            headers=auth_header(admin_token),
+        )
+        await client.post(
+            "/api/v1/audit-rules",
+            json={"name": "stats_rule_2", "rule_type": "performance", "severity": "warning", "message": "test"},
+            headers=auth_header(admin_token),
+        )
+
+        resp = await client.get("/api/v1/audit-rules/stats/summary", headers=auth_header(admin_token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["total"] >= 2
+        assert data["enabled"] >= 2
+        assert "by_type" in data