logan
2984f11181
- 删除16个旧Python测试文件(970行→400行) - 重写conftest.py适配真实种子用户(admin/test123/test234/testtest/logan) - 5个精简测试文件覆盖核心API: - test_auth.py: 登录/鉴权/种子用户验证 (6 tests) - test_instances.py: 实例列表/详情/连接测试 (6 tests) - test_tickets.py: 工单CRUD/只读SQL拦截/auditor权限 (7 tests) - test_workflow.py: E2E完整审批流/拒绝/撤销 (3 tests) - test_permissions.py: 角色权限/执行权限/workflow权限 (6 tests) - 新增 cleanup_test_data.py 清理测试脏数据(支持--dry-run) - 全量29测试3.27s通过
64 lines
2.8 KiB
Python
64 lines
2.8 KiB
Python
"""Permission tests — verify role-based access control"""
|
||
import pytest
|
||
from httpx import AsyncClient
|
||
def auth_header(token: str) -> dict:
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
class TestTicketPermissions:
|
||
"""验证各角色对工单操作的权限"""
|
||
|
||
async def test_developer_can_see_approver_tickets(self, client: AsyncClient,
|
||
developer_token, approver1_token):
|
||
"""test123(developer) 作为审批人能看到待审工单"""
|
||
# test123 查看工单列表
|
||
resp = await client.get("/api/v1/tickets", headers=auth_header(approver1_token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
# 应能看到工单（自己的 + 作为审批人的）
|
||
assert "items" in data
|
||
|
||
async def test_only_admin_dba_can_execute(self, client: AsyncClient,
|
||
developer_token, admin_token, mysql_instance):
|
||
"""developer 不能执行工单"""
|
||
# 创建工单
|
||
resp = await client.post("/api/v1/tickets", headers=auth_header(developer_token), json={
|
||
"title": "权限测试-执行",
|
||
"db_type": "mysql",
|
||
"instance_id": mysql_instance["id"],
|
||
"target_database": "opsxsql_test",
|
||
"sql_content": "CREATE TABLE IF NOT EXISTS _perm_test (id INT)",
|
||
"workflow_id": 4,
|
||
})
|
||
assert resp.status_code in (200, 201)
|
||
tid = resp.json()["id"]
|
||
|
||
# developer 尝试执行 → 应被拒绝
|
||
resp = await client.post(f"/api/v1/tickets/{tid}/execute", headers=auth_header(developer_token), json={})
|
||
assert resp.status_code in (403, 400)
|
||
|
||
|
||
class TestWorkflowPermissions:
|
||
"""审批流程管理权限"""
|
||
|
||
async def test_admin_can_list_workflows(self, client: AsyncClient, admin_token):
|
||
resp = await client.get("/api/v1/workflows", headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
|
||
async def test_developer_can_list_workflows(self, client: AsyncClient, developer_token):
|
||
"""developer 应该能看到 workflow 列表（前端提交时需要选 workflow）"""
|
||
resp = await client.get("/api/v1/workflows", headers=auth_header(developer_token))
|
||
assert resp.status_code == 200
|
||
|
||
|
||
class TestDashboardPermissions:
|
||
"""Dashboard 统计数据权限"""
|
||
|
||
async def test_admin_statistics(self, client: AsyncClient, admin_token):
|
||
resp = await client.get("/api/v1/tickets/statistics", headers=auth_header(admin_token))
|
||
assert resp.status_code == 200
|
||
|
||
async def test_developer_statistics(self, client: AsyncClient, developer_token):
|
||
resp = await client.get("/api/v1/tickets/statistics", headers=auth_header(developer_token))
|
||
assert resp.status_code == 200
|