# -*- coding: utf-8 -*- """功能④ RBAC 细粒度角色权限 — 验真测试。 验真目标: 1. 角色/权限模型 + 种子数据是否真实存在(rbac_service / models / scripts) 2. require_role / require_permission 在 API 层是否真正生效 —— 期望:admin 可访问(200),非 admin 被拒(403) —— 实际:所有 /admin/users 端点返回 422(装饰器被误用为 Depends) 说明:本测试中的"期望行为"断言是正确的(符合 PRD/设计), 若断言失败,说明是"源码缺陷"而非"测试写错"。 """ from sqlalchemy import select from app.models.agent import Agent from app.models.role import Role from app.models.user_role import UserRole from app.services.rbac_service import ROLE_PERMISSIONS, check_permission from tests.conftest import create_test_agent # --------------------------------------------------------------------------- # 1. 角色/权限模型(单元测试):证明模型是真实的 # --------------------------------------------------------------------------- def test_rbac_role_permissions_model_is_real(): """ROLE_PERMISSIONS 包含 5 个角色,admin 使用通配符 *:*:all。""" expected_roles = {"user", "agent", "team_lead", "auditor", "admin"} assert expected_roles.issubset(set(ROLE_PERMISSIONS.keys())), ( f"缺少角色: {expected_roles - set(ROLE_PERMISSIONS.keys())}" ) # admin 使用通配符表示全权限 assert ("*", "*", "all") in ROLE_PERMISSIONS["admin"], "admin 缺少 *:*:all 通配符" def test_check_permission_returns_true_for_granted(): """已授权角色应返回 True。""" ok = check_permission( user_roles=["admin"], user_permissions={"admin": ["conversation:read:all"]}, required_resource="conversation", required_action="read", required_scope="all", ) assert ok is True def test_check_permission_returns_false_for_denied(): """未授权角色应返回 False。""" ok = check_permission( user_roles=["agent"], user_permissions={"agent": ["conversation:read:own"]}, required_resource="conversation", required_action="read", required_scope="all", ) assert ok is False # --------------------------------------------------------------------------- # 2. API 层 RBAC 生效性(集成测试):期望 admin=200 / 非 admin=403 # 若返回 422,则证明装饰器被误用为 Depends(源码缺陷) # --------------------------------------------------------------------------- async def _seed_role(db_session, role_name: str) -> str: role = (await db_session.execute(select(Role).where(Role.name == role_name))).scalars().first() if not role: role = Role(id=f"rbac-{role_name}", name=role_name, display_name=role_name, permissions=[]) db_session.add(role) await db_session.flush() return role.id async def _login(client, user_id, name): r = await client.post("/agents/login", json={"user_id": user_id, "name": name}) assert r.status_code == 200, f"登录失败: {r.text}" return r.json()["data"]["token"] async def test_admin_user_list_allows_admin(client, db_session): """期望:拥有 admin 角色的坐席访问 GET /admin/users 应返回 200。 实际结果若为 422,证明 require_role 装饰器被误用为 Depends(require_role(...))。 """ role_id = await _seed_role(db_session, "admin") admin = create_test_agent(user_id="rbac_admin", name="RBAC管理员") admin.role = "admin" db_session.add(admin) await db_session.flush() db_session.add(UserRole(id="rbac-ur-1", employee_id="rbac_admin", role_id=role_id, source="manual")) await db_session.flush() token = await _login(client, "rbac_admin", "RBAC管理员") r = await client.get("/admin/users", headers={"Authorization": f"Bearer {token}"}) # 期望 200;若源码正确,应得到 200。若为 422 则装饰器误用。 assert r.status_code == 200, f"期望200,实际 {r.status_code}: {r.text[:300]}" async def test_admin_user_list_denies_non_admin(client, db_session): """期望:无 admin 角色(仅 agent)访问 GET /admin/users 应返回 403。 实际结果若为 422,证明装饰器被误用(与权限无关地全部 422)。 """ role_id = await _seed_role(db_session, "agent") agent = create_test_agent(user_id="rbac_agent", name="RBAC坐席") agent.role = "agent" db_session.add(agent) await db_session.flush() db_session.add(UserRole(id="rbac-ur-2", employee_id="rbac_agent", role_id=role_id, source="manual")) await db_session.flush() token = await _login(client, "rbac_agent", "RBAC坐席") r = await client.get("/admin/users", headers={"Authorization": f"Bearer {token}"}) # 期望 403(权限不足);若源码正确。若为 422 则装饰器误用。 assert r.status_code == 403, f"期望403,实际 {r.status_code}: {r.text[:300]}"