server-core/tests/test_operator_rbac.py

# -*- coding: utf-8 -*-
"""运营端（operator 端口）子角色权限与账号管理测试。"""
from __future__ import annotations

from fastapi.testclient import TestClient


def login(client: TestClient, username: str, password: str = "123456") -> str:
    res = client.post("/auth/login", json={"username": username, "password": password})
    assert res.status_code == 200, res.text
    return res.json()["token"]


def auth(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


def user_id_by_username(client: TestClient, token: str, username: str) -> str:
    users = client.get("/admin/users", headers=auth(token)).json()
    return next(u["id"] for u in users if u["username"] == username)


def select_identity(client: TestClient, token: str, port: str) -> str:
    """把中性令牌切换到指定端口身份，返回绑定身份的令牌。"""
    body = client.post("/auth/login", json={"username": "pine", "password": "123456"}).json()
    ident = next(i for i in body["identities"] if i["port"] == port)["id"]
    sel = client.post(
        "/auth/select-identity", headers=auth(token), json={"identity_id": ident}
    ).json()
    return sel["token"]


def test_operator_admin_cannot_escalate_to_super_admin(client):
    # 管理员不可创建/提升 超管/管理员（特权角色）
    admin_t = login(client, "op_admin")
    res = client.post(
        "/admin/users",
        headers=auth(admin_t),
        json={"username": "13800001111", "password": "123456",
              "role": "operator", "sub_role": "op_super_admin"},
    )
    assert res.status_code == 403
    # 管理员可创建普通运营账号（客服）
    ok = client.post(
        "/admin/users",
        headers=auth(admin_t),
        json={"username": "13800002222", "password": "123456",
              "role": "operator", "sub_role": "op_customer_service"},
    )
    assert ok.status_code == 200
    # 管理员把某账号提升为超管 → 403
    target_id = user_id_by_username(client, admin_t, "13800002222")
    up = client.post(
        f"/admin/users/{target_id}/role",
        headers=auth(admin_t),
        json={"role": "operator", "sub_role": "op_super_admin"},
    )
    assert up.status_code == 403


def test_operator_cs_cannot_change_config(client):
    # 客服可建任务/服务商，但不可改系统配置
    cs_t = login(client, "op_cs")
    assert (
        client.put("/admin/config/platform.fee.task",
                   headers=auth(cs_t), json={"value": "0.01"}).status_code == 403
    )
    assert client.post("/admin/tasks", headers=auth(cs_t),
                       json={"title": "客服建任务"}).status_code == 200


def test_operator_finance_cannot_manage_tasks(client):
    # 财务可看结算，但不可建任务
    fin_t = login(client, "op_fin")
    assert (
        client.post("/admin/tasks", headers=auth(fin_t),
                    json={"title": "财务建任务"}).status_code == 403
    )
    assert client.get("/operator/settlements", headers=auth(fin_t)).status_code == 200


def test_operator_invalid_role_combo_rejected(client):
    # 非法/跨界角色组合被白名单拒绝
    admin_t = login(client, "op_admin")
    res = client.post(
        "/admin/users",
        headers=auth(admin_t),
        json={"username": "13800003333", "password": "123456",
              "role": "operator", "sub_role": "gov_province"},
    )
    assert res.status_code == 400


def test_operator_grant_perm_only_super_admin(client):
    # 配置角色权限仅超管可用
    admin_t = login(client, "op_admin")
    assert (
        client.post("/admin/roles/operator%7Cop_admin/permissions",
                    headers=auth(admin_t), json={"permissions": []}).status_code == 403
    )
    # pine 以超管身份可配置
    neutral = login(client, "pine")
    sa_t = select_identity(client, neutral, "operator")
    ok = client.post(
        "/admin/roles/operator%7Cop_analyst/permissions",
        headers=auth(sa_t), json={"permissions": ["menu:audit", "action:data.export"]},
    )
    assert ok.status_code == 200
    assert "menu:audit" in ok.json()["permissions"]


def test_operator_role_change_takes_effect(client):
    # 改角色后：令牌失效 + 身份同步，重新登录体现新角色权限
    admin_t = login(client, "op_admin")
    created = client.post(
        "/admin/users",
        headers=auth(admin_t),
        json={"username": "13800004444", "password": "123456",
              "role": "operator", "sub_role": "op_customer_service"},
    ).json()
    # 该账号现可建任务（客服有 task.manage）
    cs_t = login(client, "13800004444")
    assert client.post("/admin/tasks", headers=auth(cs_t),
                       json={"title": "t"}).status_code == 200
    # 管理员将其改为分析师（无 task.manage）
    target_id = user_id_by_username(client, admin_t, "13800004444")
    up = client.post(
        f"/admin/users/{target_id}/role",
        headers=auth(admin_t),
        json={"role": "operator", "sub_role": "op_analyst"},
    )
    assert up.status_code == 200
    # 旧令牌已失效
    assert client.get("/admin/users", headers=auth(cs_t)).status_code == 401
    # 新登录为分析师：不可建任务
    an_t = login(client, "13800004444")
    assert client.post("/admin/tasks", headers=auth(an_t),
                       json={"title": "t2"}).status_code == 403