114 lines
4.9 KiB
Python
114 lines
4.9 KiB
Python
|
# -*- 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]}"
|