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
103 lines
4.5 KiB
Python
103 lines
4.5 KiB
Python
"""
|
||
集成测试: 审核规则 + 审核策略 + SQL审计打分
|
||
"""
|
||
import pytest
|
||
|
||
|
||
def _auth(token: str) -> dict:
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# DDL规则 — 需要sql-audit API（Go后端未实现）
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
class TestDDLRulesEnforcement:
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_create_table_no_pk(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_drop_table_production(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_truncate_warning(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_float_type_warning(self, client, admin_token): pass
|
||
|
||
|
||
class TestDMLRulesEnforcement:
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_update_no_where(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_delete_no_where(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_insert_with_where_ok(self, client, admin_token): pass
|
||
|
||
|
||
class TestPerformanceRules:
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_select_star(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_clean_select(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_join_no_on(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_large_in_list(self, client, admin_token): pass
|
||
|
||
|
||
class TestRedisAuditRules:
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_flushall_critical(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_flushdb_critical(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_keys_pattern_high(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_get_low_risk(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_config_set_warning(self, client, admin_token): pass
|
||
|
||
|
||
class TestAuditScoring:
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_clean_sql_high_score(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_dangerous_sql_low_score(self, client, admin_token): pass
|
||
|
||
@pytest.mark.skip(reason="Go后端未实现 sql-audit 端点")
|
||
async def test_score_between_0_100(self, client, admin_token): pass
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 审核规则管理 — Go后端有 audit-rules 端点
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
class TestAuditRuleManagement:
|
||
async def test_list_rules(self, client, admin_token):
|
||
resp = await client.get("/api/v1/audit-rules", headers=_auth(admin_token))
|
||
assert resp.status_code == 200
|
||
|
||
async def test_create_and_toggle_rule(self, client, admin_token):
|
||
import uuid
|
||
# 创建
|
||
resp = await client.post("/api/v1/audit-rules", json={
|
||
"name": f"测试规则_{uuid.uuid4().hex[:8]}", "rule_type": "syntax",
|
||
"severity": "warning", "message": "测试消息",
|
||
}, headers=_auth(admin_token))
|
||
assert resp.status_code in (200, 201, 400, 404)
|
||
|
||
async def test_rules_stats(self, client, admin_token):
|
||
resp = await client.get("/api/v1/audit-rules/stats", headers=_auth(admin_token))
|
||
assert resp.status_code in (200, 404)
|