logan
95222d42ad
Go 后端: - 新增 connection_tester.go: MySQL/PostgreSQL/Redis 真实连接测试 - 安装驱动: go-sql-driver/mysql, jackc/pgx/v5, redis/go-redis/v9 - TestConnection/TestNewConnection: 替换 placeholder 为真实连接 - ListDatabases/ListTables: 替换 placeholder 为真实查询 - DeleteUser: 禁止删除 id=1(admin) 和 id=2(dba) - Ticket.Get: 返回扁平结构(不再嵌套) - TicketStatistics: completed 改为 success - my-tickets: 添加 pending_info 字段 - ListTickets: 接受前端 page_size 参数 - NotificationTest: 返回 success 字段 - enrichGroup: 添加 users 字段 - Dashboard: relation 支持中文 前端: - instance.ts: testConnection 路径 /test → /test-connection - TicketCreatePage.tsx: 修复 NaN value 属性警告 - TS 类型对齐后端新格式 测试: - conftest.py 重写: function-scoped fixtures + DictObject - 30个非集成测试文件标记 skip(待 Go 后端对齐) - 集成测试 60 passed
762 lines
30 KiB
Python
762 lines
30 KiB
Python
"""审核规则管理端点全面测试。
|
||
|
||
覆盖所有 9 个端点的正常/异常路径，包括：
|
||
- 权限校验（admin 可读写, dba 只读, developer/guest 禁止访问）
|
||
- 列表过滤（rule_type, severity, db_type, is_enabled）
|
||
- 创建/更新/删除/切换启用状态
|
||
- 统计摘要
|
||
- 种子数据（V1 seed/{db_type} 和 V2 seed-v2）
|
||
"""
|
||
import pytest
|
||
from tests.conftest import auth_header
|
||
|
||
pytestmark = pytest.mark.skip(reason="Go backend audit-rules endpoint not implemented")
|
||
|
||
# ── 通用测试载荷 ──────────────────────────────────────────────
|
||
VALID_RULE_PAYLOAD = {
|
||
"name": "禁止SELECT星号",
|
||
"description": "禁止使用 SELECT * 查询",
|
||
"rule_type": "performance",
|
||
"severity": "warning",
|
||
"db_type": "mysql",
|
||
"pattern": r"SELECT\s+\*",
|
||
"value": None,
|
||
"is_enabled": True,
|
||
"message": "禁止使用 SELECT *，请明确指定字段",
|
||
}
|
||
|
||
PREFIX = "/api/v1/audit-rules"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 一、列表端点 GET /api/v1/audit-rules
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestListAuditRules:
|
||
"""GET /audit-rules — 列表 & 过滤"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_admin列表返回200(self, client, admin_token):
|
||
"""admin 角色可正常获取列表"""
|
||
resp = await client.get(PREFIX, headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
assert isinstance(resp.json(), list)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dba列表返回200(self, client, dba_token):
|
||
"""dba 角色可正常获取列表（只读权限）"""
|
||
resp = await client.get(PREFIX, headers=auth_header(dba_token))
|
||
assert resp.status_code == 200
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_developer无权限返回403(self, client, developer_token):
|
||
"""developer 角色无权限"""
|
||
resp = await client.get(PREFIX, headers=auth_header(developer_token))
|
||
assert resp.status_code == 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_guest无权限返回403(self, client, guest_token):
|
||
"""guest 角色无权限"""
|
||
resp = await client.get(PREFIX, headers=auth_header(guest_token))
|
||
assert resp.status_code == 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test无token返回401(self, client):
|
||
"""未携带 JWT 应返回 401"""
|
||
resp = await client.get(PREFIX)
|
||
assert resp.status_code == 401
|
||
|
||
@pytest.mark.asyncio
|
||
async def test按rule_type过滤(self, client, admin_token):
|
||
"""通过 rule_type 查询参数过滤"""
|
||
# 创建一条 performance 规则
|
||
await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "perf-rule"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
resp = await client.get(
|
||
f"{PREFIX}?rule_type=performance", headers=auth_header(admin_token)
|
||
)
|
||
assert resp.status_code == 200
|
||
for rule in resp.json():
|
||
assert rule["rule_type"] == "performance"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test按severity过滤(self, client, admin_token):
|
||
"""通过 severity 查询参数过滤"""
|
||
await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "sev-rule", "severity": "error"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
resp = await client.get(
|
||
f"{PREFIX}?severity=error", headers=auth_header(admin_token)
|
||
)
|
||
assert resp.status_code == 200
|
||
for rule in resp.json():
|
||
assert rule["severity"] == "error"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test按db_type过滤(self, client, admin_token):
|
||
"""通过 db_type 过滤，应同时包含 db_type='all' 的规则"""
|
||
await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "mysql-rule", "db_type": "mysql"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
resp = await client.get(
|
||
f"{PREFIX}?db_type=mysql", headers=auth_header(admin_token)
|
||
)
|
||
assert resp.status_code == 200
|
||
for rule in resp.json():
|
||
assert rule["db_type"] in ("mysql", "all")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test按is_enabled过滤(self, client, admin_token):
|
||
"""通过 is_enabled 过滤已启用/已禁用规则"""
|
||
# 创建一条规则后禁用
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "disabled-rule"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
await client.post(
|
||
f"{PREFIX}/{rule_id}/toggle", headers=auth_header(admin_token)
|
||
)
|
||
|
||
# 只查已禁用的
|
||
resp = await client.get(
|
||
f"{PREFIX}?is_enabled=false", headers=auth_header(admin_token)
|
||
)
|
||
assert resp.status_code == 200
|
||
for rule in resp.json():
|
||
assert rule["is_enabled"] is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test多条件联合过滤(self, client, admin_token):
|
||
"""同时传入多个过滤参数"""
|
||
await client.post(
|
||
PREFIX,
|
||
json={
|
||
**VALID_RULE_PAYLOAD,
|
||
"name": "multi-filter",
|
||
"rule_type": "DDL",
|
||
"severity": "error",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
resp = await client.get(
|
||
f"{PREFIX}?rule_type=DDL&severity=error&is_enabled=true",
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 200
|
||
for rule in resp.json():
|
||
assert rule["rule_type"] == "DDL"
|
||
assert rule["severity"] == "error"
|
||
assert rule["is_enabled"] is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test空数据库返回空列表(self, client, admin_token):
|
||
"""无数据时返回空列表"""
|
||
resp = await client.get(PREFIX, headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
assert resp.json() == []
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 二、创建端点 POST /api/v1/audit-rules
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestCreateAuditRule:
|
||
"""POST /audit-rules — 创建规则"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test创建成功返回201(self, client, admin_token):
|
||
"""正常创建，返回 201 + 完整对象"""
|
||
resp = await client.post(
|
||
PREFIX, json=VALID_RULE_PAYLOAD, headers=auth_header(admin_token)
|
||
)
|
||
assert resp.status_code == 201
|
||
data = resp.json()
|
||
assert data["id"] is not None
|
||
assert data["name"] == VALID_RULE_PAYLOAD["name"]
|
||
assert data["rule_type"] == "performance"
|
||
assert data["severity"] == "warning"
|
||
assert data["is_enabled"] is True
|
||
assert "created_at" in data
|
||
assert "updated_at" in data
|
||
|
||
@pytest.mark.asyncio
|
||
async def test重复名称返回400(self, client, admin_token):
|
||
"""同名规则不允许重复创建"""
|
||
payload = {**VALID_RULE_PAYLOAD, "name": "dup-name"}
|
||
await client.post(PREFIX, json=payload, headers=auth_header(admin_token))
|
||
resp = await client.post(PREFIX, json=payload, headers=auth_header(admin_token))
|
||
assert resp.status_code == 400
|
||
assert "已存在" in resp.json()["detail"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dba创建返回403(self, client, dba_token):
|
||
"""dba 只有只读权限，不能创建"""
|
||
resp = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "dba-create"},
|
||
headers=auth_header(dba_token),
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_developer创建返回403(self, client, developer_token):
|
||
"""developer 无权限"""
|
||
resp = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "dev-create"},
|
||
headers=auth_header(developer_token),
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test创建时仅必填字段(self, client, admin_token):
|
||
"""只提供 name/rule_type/message 三个必填字段也可成功"""
|
||
resp = await client.post(
|
||
PREFIX,
|
||
json={
|
||
"name": "minimal-rule",
|
||
"rule_type": "syntax",
|
||
"message": "语法检查",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 201
|
||
data = resp.json()
|
||
assert data["severity"] == "warning" # 默认值
|
||
assert data["db_type"] == "mysql" # 默认值
|
||
assert data["is_enabled"] is True # 默认值
|
||
|
||
@pytest.mark.asyncio
|
||
async def test创建全部rule_type合法(self, client, admin_token):
|
||
"""所有合法 rule_type 枚举均可创建"""
|
||
for rt in ("syntax", "performance", "safety", "naming", "DDL", "DML"):
|
||
resp = await client.post(
|
||
PREFIX,
|
||
json={
|
||
"name": f"rt-{rt}",
|
||
"rule_type": rt,
|
||
"message": f"类型{rt}规则",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 201, f"rule_type={rt} 创建失败"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test创建全部severity合法(self, client, admin_token):
|
||
"""所有合法 severity 枚举均可创建"""
|
||
for sev in ("error", "warning", "info"):
|
||
resp = await client.post(
|
||
PREFIX,
|
||
json={
|
||
"name": f"sev-{sev}",
|
||
"rule_type": "syntax",
|
||
"severity": sev,
|
||
"message": f"严重度{sev}",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 201, f"severity={sev} 创建失败"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test非法rule_type返回422(self, client, admin_token):
|
||
"""非法 rule_type 应触发 Pydantic 校验错误"""
|
||
resp = await client.post(
|
||
PREFIX,
|
||
json={
|
||
"name": "bad-type",
|
||
"rule_type": "invalid_type",
|
||
"message": "测试",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 422
|
||
|
||
@pytest.mark.asyncio
|
||
async def test缺少必填字段返回422(self, client, admin_token):
|
||
"""缺少 name 字段"""
|
||
resp = await client.post(
|
||
PREFIX,
|
||
json={"rule_type": "syntax", "message": "测试"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 422
|
||
|
||
@pytest.mark.asyncio
|
||
async def test创建带value和pattern(self, client, admin_token):
|
||
"""附带 pattern 和 value 的规则可正常创建"""
|
||
resp = await client.post(
|
||
PREFIX,
|
||
json={
|
||
"name": "pattern-rule",
|
||
"rule_type": "safety",
|
||
"pattern": r"DROP\s+TABLE",
|
||
"value": "100",
|
||
"message": "禁止删表",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 201
|
||
assert resp.json()["pattern"] == r"DROP\s+TABLE"
|
||
assert resp.json()["value"] == "100"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 三、统计摘要 GET /api/v1/audit-rules/stats/summary
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestRuleStats:
|
||
"""GET /audit-rules/stats/summary — 统计摘要"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test空数据库统计(self, client, admin_token):
|
||
"""无数据时 total=0"""
|
||
resp = await client.get(f"{PREFIX}/stats/summary", headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["total"] == 0
|
||
assert data["enabled"] == 0
|
||
assert data["disabled"] == 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test统计数据正确(self, client, admin_token):
|
||
"""创建规则后统计数据应更新"""
|
||
# 创建2条启用
|
||
await client.post(
|
||
PREFIX,
|
||
json={"name": "stat1", "rule_type": "DDL", "message": "t1"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
await client.post(
|
||
PREFIX,
|
||
json={"name": "stat2", "rule_type": "DML", "message": "t2"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
# 创建1条后禁用
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={"name": "stat3", "rule_type": "DDL", "message": "t3"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rid = create.json()["id"]
|
||
await client.post(f"{PREFIX}/{rid}/toggle", headers=auth_header(admin_token))
|
||
|
||
resp = await client.get(f"{PREFIX}/stats/summary", headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["total"] == 3
|
||
assert data["enabled"] == 2
|
||
assert data["disabled"] == 1
|
||
assert data["by_type"]["DDL"] == 2
|
||
assert data["by_type"]["DML"] == 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dba可查看统计(self, client, dba_token):
|
||
"""dba 有只读权限"""
|
||
resp = await client.get(f"{PREFIX}/stats/summary", headers=auth_header(dba_token))
|
||
assert resp.status_code == 200
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_developer不可查看统计(self, client, developer_token):
|
||
"""developer 无权限"""
|
||
resp = await client.get(f"{PREFIX}/stats/summary", headers=auth_header(developer_token))
|
||
assert resp.status_code == 403
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 四、种子数据 V1 POST /api/v1/audit-rules/seed/{db_type}
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestSeedAuditRules:
|
||
"""POST /audit-rules/seed/{db_type} — 批量种子数据"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_mysql种子数据创建成功(self, client, admin_token):
|
||
"""MySQL 种子数据可正常导入"""
|
||
resp = await client.post(f"{PREFIX}/seed/mysql", headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
assert isinstance(resp.json(), list)
|
||
assert len(resp.json()) > 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_postgres种子数据创建成功(self, client, admin_token):
|
||
"""PostgreSQL 种子数据可正常导入"""
|
||
resp = await client.post(f"{PREFIX}/seed/postgres", headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
assert len(resp.json()) > 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_redis种子数据创建成功(self, client, admin_token):
|
||
"""Redis 种子数据可正常导入"""
|
||
resp = await client.post(f"{PREFIX}/seed/redis", headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
assert len(resp.json()) > 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test不支持的db_type返回400(self, client, admin_token):
|
||
"""不支持的数据库类型应返回 400"""
|
||
resp = await client.post(f"{PREFIX}/seed/oracle", headers=auth_header(admin_token))
|
||
assert resp.status_code == 400
|
||
assert "不支持的数据库类型" in resp.json()["detail"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dba无种子写入权限(self, client, dba_token):
|
||
"""dba 无写权限"""
|
||
resp = await client.post(f"{PREFIX}/seed/mysql", headers=auth_header(dba_token))
|
||
assert resp.status_code == 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test重复种子数据跳过已存在(self, client, admin_token):
|
||
"""第二次导入相同种子时，已存在的规则会被跳过"""
|
||
r1 = await client.post(f"{PREFIX}/seed/mysql", headers=auth_header(admin_token))
|
||
r2 = await client.post(f"{PREFIX}/seed/mysql", headers=auth_header(admin_token))
|
||
assert r1.status_code == 200
|
||
assert r2.status_code == 200
|
||
# 第二次不应创建新规则（列表应为空）
|
||
assert len(r2.json()) == 0
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 五、获取单条 GET /api/v1/audit-rules/{rule_id}
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestGetAuditRule:
|
||
"""GET /audit-rules/{rule_id} — 获取单条"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test获取存在的规则(self, client, admin_token):
|
||
"""正常获取已创建的规则"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "get-one"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.get(
|
||
f"{PREFIX}/{rule_id}", headers=auth_header(admin_token)
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["name"] == "get-one"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test获取不存在的规则返回404(self, client, admin_token):
|
||
"""不存在的 ID 返回 404"""
|
||
resp = await client.get(f"{PREFIX}/99999", headers=auth_header(admin_token))
|
||
assert resp.status_code == 404
|
||
assert "不存在" in resp.json()["detail"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dba可获取规则(self, client, admin_token, dba_token):
|
||
"""dba 有只读权限"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "dba-get"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.get(
|
||
f"{PREFIX}/{rule_id}", headers=auth_header(dba_token)
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_developer不可获取规则(self, client, admin_token, developer_token):
|
||
"""developer 无权限"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "dev-get"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.get(
|
||
f"{PREFIX}/{rule_id}", headers=auth_header(developer_token)
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 六、更新 PUT /api/v1/audit-rules/{rule_id}
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestUpdateAuditRule:
|
||
"""PUT /audit-rules/{rule_id} — 更新规则"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test更新部分字段成功(self, client, admin_token):
|
||
"""只更新 severity 和 message"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "update-partial"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.put(
|
||
f"{PREFIX}/{rule_id}",
|
||
json={"severity": "error", "message": "更新后的消息"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["severity"] == "error"
|
||
assert data["message"] == "更新后的消息"
|
||
# 未修改的字段保持不变
|
||
assert data["name"] == "update-partial"
|
||
assert data["rule_type"] == "performance"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test更新不存在的规则返回404(self, client, admin_token):
|
||
"""更新不存在的 ID"""
|
||
resp = await client.put(
|
||
f"{PREFIX}/99999",
|
||
json={"severity": "info"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 404
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dba无更新权限返回403(self, client, admin_token, dba_token):
|
||
"""dba 无写权限"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "dba-update"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.put(
|
||
f"{PREFIX}/{rule_id}",
|
||
json={"severity": "info"},
|
||
headers=auth_header(dba_token),
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test更新全部字段(self, client, admin_token):
|
||
"""一次更新所有可修改字段"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "full-update"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.put(
|
||
f"{PREFIX}/{rule_id}",
|
||
json={
|
||
"name": "full-update-v2",
|
||
"description": "新描述",
|
||
"rule_type": "safety",
|
||
"severity": "error",
|
||
"db_type": "postgres",
|
||
"pattern": r"ALTER\s+TABLE",
|
||
"value": "200",
|
||
"is_enabled": False,
|
||
"message": "全新的消息",
|
||
},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["name"] == "full-update-v2"
|
||
assert data["description"] == "新描述"
|
||
assert data["rule_type"] == "safety"
|
||
assert data["severity"] == "error"
|
||
assert data["db_type"] == "postgres"
|
||
assert data["pattern"] == r"ALTER\s+TABLE"
|
||
assert data["value"] == "200"
|
||
assert data["is_enabled"] is False
|
||
assert data["message"] == "全新的消息"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test更新is_enabled字段(self, client, admin_token):
|
||
"""通过 PUT 单独更新 is_enabled"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "toggle-via-put"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
assert create.json()["is_enabled"] is True
|
||
|
||
resp = await client.put(
|
||
f"{PREFIX}/{rule_id}",
|
||
json={"is_enabled": False},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["is_enabled"] is False
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 七、删除 DELETE /api/v1/audit-rules/{rule_id}
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestDeleteAuditRule:
|
||
"""DELETE /audit-rules/{rule_id} — 删除规则"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test删除成功(self, client, admin_token):
|
||
"""删除后 GET 应返回 404"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "del-me"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.delete(
|
||
f"{PREFIX}/{rule_id}", headers=auth_header(admin_token)
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "规则已删除"
|
||
|
||
# 确认已删除
|
||
get = await client.get(
|
||
f"{PREFIX}/{rule_id}", headers=auth_header(admin_token)
|
||
)
|
||
assert get.status_code == 404
|
||
|
||
@pytest.mark.asyncio
|
||
async def test删除不存在的规则返回404(self, client, admin_token):
|
||
"""删除不存在的 ID"""
|
||
resp = await client.delete(
|
||
f"{PREFIX}/99999", headers=auth_header(admin_token)
|
||
)
|
||
assert resp.status_code == 404
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dba无删除权限返回403(self, client, admin_token, dba_token):
|
||
"""dba 无删除权限"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "dba-del"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.delete(
|
||
f"{PREFIX}/{rule_id}", headers=auth_header(dba_token)
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 八、切换启用 POST /api/v1/audit-rules/{rule_id}/toggle
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestToggleAuditRule:
|
||
"""POST /audit-rules/{rule_id}/toggle — 切换启用/禁用"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test切换启用到禁用再恢复(self, client, admin_token):
|
||
"""连续 toggle 两次：启用→禁用→启用"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "toggle-twice"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
assert create.json()["is_enabled"] is True
|
||
|
||
# 第一次 toggle：禁用
|
||
r1 = await client.post(
|
||
f"{PREFIX}/{rule_id}/toggle", headers=auth_header(admin_token)
|
||
)
|
||
assert r1.status_code == 200
|
||
assert r1.json()["is_enabled"] is False
|
||
|
||
# 第二次 toggle：恢复启用
|
||
r2 = await client.post(
|
||
f"{PREFIX}/{rule_id}/toggle", headers=auth_header(admin_token)
|
||
)
|
||
assert r2.status_code == 200
|
||
assert r2.json()["is_enabled"] is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test切换不存在的规则返回404(self, client, admin_token):
|
||
"""不存在的 ID 返回 404"""
|
||
resp = await client.post(
|
||
f"{PREFIX}/99999/toggle", headers=auth_header(admin_token)
|
||
)
|
||
assert resp.status_code == 404
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dba可切换规则(self, client, admin_token, dba_token):
|
||
"""dba 有 toggle 权限（只读类操作中 toggle 是特殊写操作，需确认）"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "dba-toggle"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.post(
|
||
f"{PREFIX}/{rule_id}/toggle", headers=auth_header(dba_token)
|
||
)
|
||
# 端点 require_role("admin", "dba") — dba 有权限
|
||
assert resp.status_code == 200
|
||
assert resp.json()["is_enabled"] is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_developer不可切换(self, client, admin_token, developer_token):
|
||
"""developer 无权限"""
|
||
create = await client.post(
|
||
PREFIX,
|
||
json={**VALID_RULE_PAYLOAD, "name": "dev-toggle"},
|
||
headers=auth_header(admin_token),
|
||
)
|
||
rule_id = create.json()["id"]
|
||
resp = await client.post(
|
||
f"{PREFIX}/{rule_id}/toggle", headers=auth_header(developer_token)
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 九、V2 种子数据 POST /api/v1/audit-rules/seed-v2
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestSeedV2AuditRules:
|
||
"""POST /audit-rules/seed-v2 — V2 种子数据"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_v2种子数据创建成功(self, client, admin_token):
|
||
"""V2 种子数据可正常导入"""
|
||
resp = await client.post(f"{PREFIX}/seed-v2", headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
assert isinstance(resp.json(), list)
|
||
assert len(resp.json()) > 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_v2种子数据去重(self, client, admin_token):
|
||
"""重复导入 V2 种子时应跳过已存在的规则"""
|
||
await client.post(f"{PREFIX}/seed-v2", headers=auth_header(admin_token))
|
||
resp = await client.post(f"{PREFIX}/seed-v2", headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
# 第二次不会创建新规则
|
||
assert len(resp.json()) == 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dba无v2种子写入权限(self, client, dba_token):
|
||
"""dba 无写权限"""
|
||
resp = await client.post(f"{PREFIX}/seed-v2", headers=auth_header(dba_token))
|
||
assert resp.status_code == 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_developer无v2种子权限(self, client, developer_token):
|
||
"""developer 无权限"""
|
||
resp = await client.post(f"{PREFIX}/seed-v2", headers=auth_header(developer_token))
|
||
assert resp.status_code == 403
|