document-management/backend/tests/dms/test_role_management.py

"""需求16 角色管理与权限矩阵测试。

覆盖：
- 角色 CRUD、角色名唯一、默认角色全局唯一（先清后设）
- row_version 乐观锁
- ROLE_IN_USE 删除守卫与解除后可删
- 权限矩阵 10 行整存校验与持久化
- 用户关联（仅普通用户、auth_version 递增、解除）
- 非 ADMIN 访问 403
- 审计记录（CREATE_ROLE/EDIT_ROLE/CHANGE_ROLE_PERMISSION/CHANGE_ROLE_USER/DELETE_ROLE）
"""

from __future__ import annotations

import pytest

from dms.extensions import db
from dms.models import AuditLog, User


LEVELS = ("PUBLIC", "INTERNAL", "SECRET", "CONFIDENTIAL", "TOP_SECRET")
SCOPES = ("OWN", "OTHER")
ACTION_KEYS = (
    "canView",
    "canDownload",
    "canEdit",
    "canDelete",
    "canConfigPermission",
    "canBindAttachment",
    "canUnbindAttachment",
)


def _auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def _create_role(
    client,
    token: str,
    *,
    name: str,
    sort_no: int = 100,
    status: str = "ENABLED",
    remark: str | None = None,
    is_default: bool = False,
):
    return client.post(
        "/api/v1/roles",
        json={
            "roleName": name,
            "sortNo": sort_no,
            "status": status,
            "remark": remark,
            "isDefault": is_default,
        },
        headers=_auth(token),
    )


def _update_role(client, token, role_id, payload):
    return client.put(
        f"/api/v1/roles/{role_id}", json=payload, headers=_auth(token)
    )


def _matrix(overrides: dict | None = None) -> list[dict]:
    overrides = overrides or {}
    rows = []
    for level in LEVELS:
        for scope in SCOPES:
            row: dict[str, object] = {
                "securityLevel": level,
                "deptScope": scope,
            }
            row.update({key: False for key in ACTION_KEYS})
            row.update(overrides.get((level, scope), {}))
            rows.append(row)
    return rows


def _row_payload(role, *, is_default=None, name=None) -> dict:
    return {
        "roleName": name or role["roleName"],
        "sortNo": role["sortNo"],
        "status": role["status"],
        "remark": role["remark"],
        "isDefault": role["isDefault"] if is_default is None else is_default,
        "rowVersion": role["rowVersion"],
    }


def _user_id(app, username: str) -> int:
    with app.app_context():
        user = db.session.scalar(
            db.select(User).where(User.username == username)
        )
        assert user is not None, f"user {username} not found"
        return user.id


@pytest.mark.integration
def test_create_role_seeds_matrix_and_lists(
    b2_app, b2_client, login_token, role_cleanup
):
    response = _create_role(b2_client, login_token, name="资料员", sort_no=10)
    assert response.status_code == 201, response.get_json()
    role = response.get_json()["data"]
    assert role["roleName"] == "资料员"
    assert role["memberCount"] == 0
    assert role["isDefault"] is False
    assert role["rowVersion"] == 0

    listed = b2_client.get(
        "/api/v1/roles", headers=_auth(login_token)
    ).get_json()["data"]
    assert listed["total"] >= 1
    assert any(item["roleName"] == "资料员" for item in listed["items"])

    matrix = b2_client.get(
        f"/api/v1/roles/{role['id']}/permissions", headers=_auth(login_token)
    ).get_json()["data"]
    assert len(matrix) == 10
    assert {(row["securityLevel"], row["deptScope"]) for row in matrix} == {
        (level, scope) for level in LEVELS for scope in SCOPES
    }
    assert all(row[key] is False for row in matrix for key in ACTION_KEYS)


@pytest.mark.integration
def test_create_role_rejects_duplicate_name(
    b2_client, login_token, role_cleanup
):
    first = _create_role(b2_client, login_token, name="重复角色")
    assert first.status_code == 201
    second = _create_role(b2_client, login_token, name="重复角色")
    assert second.status_code == 409
    assert second.get_json()["code"] == "DATA_VERSION_CONFLICT"


@pytest.mark.integration
def test_default_role_switches_exclusively(
    b2_app, b2_client, login_token, role_cleanup
):
    first = _create_role(
        b2_client, login_token, name="默认甲", is_default=True
    ).get_json()["data"]
    second = _create_role(
        b2_client, login_token, name="默认乙", is_default=False
    ).get_json()["data"]

    switched = _update_role(
        b2_client,
        login_token,
        second["id"],
        _row_payload(second, is_default=True),
    )
    assert switched.status_code == 200, switched.get_json()
    assert switched.get_json()["data"]["isDefault"] is True

    items = b2_client.get(
        "/api/v1/roles", headers=_auth(login_token)
    ).get_json()["data"]["items"]
    by_name = {item["roleName"]: item for item in items}
    assert by_name["默认甲"]["isDefault"] is False
    assert by_name["默认乙"]["isDefault"] is True
    # 旧默认被系统调整后 rowVersion 递增
    assert by_name["默认甲"]["rowVersion"] == first["rowVersion"] + 1


@pytest.mark.integration
def test_update_role_optimistic_lock(b2_client, login_token, role_cleanup):
    role = _create_role(b2_client, login_token, name="乐观锁角色").get_json()[
        "data"
    ]
    stale = _row_payload(role, name="新名字")
    stale["rowVersion"] = role["rowVersion"] + 5
    response = _update_role(b2_client, login_token, role["id"], stale)
    assert response.status_code == 409
    body = response.get_json()
    assert body["code"] == "DATA_VERSION_CONFLICT"
    assert body["details"]["currentRowVersion"] == 0


@pytest.mark.integration
def test_delete_role_blocked_while_in_use_and_succeeds_after_unbind(
    b2_app, b2_client, login_token, role_cleanup
):
    user_id = _user_id(b2_app, "user")
    role = _create_role(b2_client, login_token, name="在用角色").get_json()[
        "data"
    ]
    bound = b2_client.put(
        f"/api/v1/roles/{role['id']}/users",
        json={"userIds": [str(user_id)], "rowVersion": role["rowVersion"]},
        headers=_auth(login_token),
    )
    assert bound.status_code == 200, bound.get_json()

    blocked = b2_client.delete(
        f"/api/v1/roles/{role['id']}?rowVersion={bound.get_json()['data']['rowVersion']}",
        headers=_auth(login_token),
    )
    assert blocked.status_code == 409
    assert blocked.get_json()["code"] == "ROLE_IN_USE"

    current_version = bound.get_json()["data"]["rowVersion"]
    unbound = b2_client.put(
        f"/api/v1/roles/{role['id']}/users",
        json={"userIds": [], "rowVersion": current_version},
        headers=_auth(login_token),
    )
    assert unbound.status_code == 200

    deleted = b2_client.delete(
        f"/api/v1/roles/{role['id']}?rowVersion={unbound.get_json()['data']['rowVersion']}",
        headers=_auth(login_token),
    )
    assert deleted.status_code == 200
    items = b2_client.get(
        "/api/v1/roles", headers=_auth(login_token)
    ).get_json()["data"]["items"]
    assert all(item["id"] != role["id"] for item in items)


@pytest.mark.integration
def test_matrix_validation_rejects_malformed_rows(
    b2_client, login_token, role_cleanup
):
    role = _create_role(b2_client, login_token, name="矩阵校验角色").get_json()[
        "data"
    ]

    short = b2_client.put(
        f"/api/v1/roles/{role['id']}/permissions",
        json={"matrix": _matrix()[:9], "rowVersion": 0},
        headers=_auth(login_token),
    )
    assert short.status_code == 400

    duplicated = _matrix()
    duplicated[1]["securityLevel"] = duplicated[0]["securityLevel"]
    duplicated[1]["deptScope"] = duplicated[0]["deptScope"]
    dup = b2_client.put(
        f"/api/v1/roles/{role['id']}/permissions",
        json={"matrix": duplicated, "rowVersion": 0},
        headers=_auth(login_token),
    )
    assert dup.status_code == 400

    typed = _matrix()
    typed[0]["canView"] = 1
    bad_type = b2_client.put(
        f"/api/v1/roles/{role['id']}/permissions",
        json={"matrix": typed, "rowVersion": 0},
        headers=_auth(login_token),
    )
    assert bad_type.status_code == 400
    assert bad_type.get_json()["code"] == "INVALID_ARGUMENT"


@pytest.mark.integration
def test_matrix_update_persists_flags(b2_client, login_token, role_cleanup):
    role = _create_role(b2_client, login_token, name="矩阵更新角色").get_json()[
        "data"
    ]
    matrix = _matrix({("SECRET", "OWN"): {"canView": True, "canDownload": True}})
    saved = b2_client.put(
        f"/api/v1/roles/{role['id']}/permissions",
        json={"matrix": matrix, "rowVersion": 0},
        headers=_auth(login_token),
    )
    assert saved.status_code == 200, saved.get_json()
    assert saved.get_json()["data"]["rowVersion"] == 1

    reloaded = b2_client.get(
        f"/api/v1/roles/{role['id']}/permissions", headers=_auth(login_token)
    ).get_json()["data"]
    by_key = {(row["securityLevel"], row["deptScope"]): row for row in reloaded}
    assert by_key[("SECRET", "OWN")]["canView"] is True
    assert by_key[("SECRET", "OWN")]["canDownload"] is True
    assert by_key[("SECRET", "OWN")]["canEdit"] is False
    assert by_key[("PUBLIC", "OTHER")]["canView"] is False


@pytest.mark.integration
def test_role_users_plain_user_only_and_auth_version_bumped(
    b2_app, b2_client, login_token, role_cleanup
):
    admin_id = _user_id(b2_app, "admin")
    user_id = _user_id(b2_app, "user")
    disabled_id = _user_id(b2_app, "disabled")
    role = _create_role(b2_client, login_token, name="用户关联角色").get_json()[
        "data"
    ]

    rejected = b2_client.put(
        f"/api/v1/roles/{role['id']}/users",
        json={"userIds": [str(admin_id)], "rowVersion": 0},
        headers=_auth(login_token),
    )
    assert rejected.status_code == 400
    assert rejected.get_json()["code"] == "INVALID_ARGUMENT"

    with b2_app.app_context():
        baseline = db.session.scalar(
            db.select(User.auth_version).where(User.username == "user")
        )
    saved = b2_client.put(
        f"/api/v1/roles/{role['id']}/users",
        json={
            "userIds": [str(user_id), str(disabled_id)],
            "rowVersion": 0,
        },
        headers=_auth(login_token),
    )
    assert saved.status_code == 200, saved.get_json()
    assert saved.get_json()["data"]["userIds"] == [
        str(disabled_id),
        str(user_id),
    ]

    listed = b2_client.get(
        f"/api/v1/roles/{role['id']}/users", headers=_auth(login_token)
    ).get_json()["data"]
    assert set(listed["userIds"]) == {str(user_id), str(disabled_id)}

    with b2_app.app_context():
        stored_role_id = db.session.scalar(
            db.select(User.role_id).where(User.username == "user")
        )
        auth_version = db.session.scalar(
            db.select(User.auth_version).where(User.username == "user")
        )
    assert stored_role_id == int(role["id"])
    assert auth_version == baseline + 1

    cleared = b2_client.put(
        f"/api/v1/roles/{role['id']}/users",
        json={"userIds": [], "rowVersion": 1},
        headers=_auth(login_token),
    )
    assert cleared.status_code == 200
    with b2_app.app_context():
        assert (
            db.session.scalar(
                db.select(User.role_id).where(User.username == "user")
            )
            is None
        )


@pytest.mark.integration
def test_role_routes_require_admin(b2_client, token_for):
    user_token = token_for("user")
    response = b2_client.get("/api/v1/roles", headers=_auth(user_token))
    assert response.status_code == 403


@pytest.mark.integration
def test_role_audit_trail_written(
    b2_app, b2_client, login_token, role_cleanup
):
    user_id = _user_id(b2_app, "user")
    role = _create_role(b2_client, login_token, name="审计角色").get_json()[
        "data"
    ]
    b2_client.put(
        f"/api/v1/roles/{role['id']}/permissions",
        json={
            "matrix": _matrix({("PUBLIC", "OWN"): {"canView": True}}),
            "rowVersion": 0,
        },
        headers=_auth(login_token),
    )
    b2_client.put(
        f"/api/v1/roles/{role['id']}/users",
        json={"userIds": [str(user_id)], "rowVersion": 1},
        headers=_auth(login_token),
    )
    b2_client.put(
        f"/api/v1/roles/{role['id']}/users",
        json={"userIds": [], "rowVersion": 2},
        headers=_auth(login_token),
    )
    b2_client.delete(
        f"/api/v1/roles/{role['id']}?rowVersion=3", headers=_auth(login_token)
    )

    with b2_app.app_context():
        actions = set(
            db.session.scalars(
                db.select(AuditLog.action_type).where(
                    AuditLog.target_id == int(role["id"]),
                    AuditLog.target_type == "ROLE",
                )
            ).all()
        )
    assert {
        "CREATE_ROLE",
        "CHANGE_ROLE_PERMISSION",
        "CHANGE_ROLE_USER",
        "DELETE_ROLE",
    } <= actions