"""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