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

"""管理类权限扩展测试（管理位挂角色，方案B）。

覆盖：
- 角色管理位（分类/机构/用户编辑）对相应写接口的放行与拒绝；
- 管理范围锚点（角色 managedOrgId）的子树收敛（机构/用户增删改与用户列表）；
- 锚点管理员不可将用户迁出管理子树；
- 用户接口不再接受管理位字段、角色锚点无效与权限位非布尔的契约拒绝；
- ADMIN 降级 USER 后角色关联保留、管理位字段跟随生效角色；
- 锚点 "SELF"（落库哨兵 0）跟随管理员自身部门，无部门时无管理范围；
- 内置 admin 不受权限位与范围约束（由全量测试套以 perm 全 0 的 admin 隐含验证）。
"""

from __future__ import annotations

import pytest

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


ADMIN_INITIAL_PASSWORD = "Initial@2026"


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


def _login(client, password: str, *, username: str) -> str:
    response = client.post(
        "/api/v1/auth/login",
        json={"username": username, "password": password, "keepSignedIn": False},
    )
    assert response.status_code == 200, response.get_json()
    return response.get_json()["data"]["accessToken"]


def _org_id(app, code: str) -> str:
    with app.app_context():
        org = db.session.scalar(
            db.select(Organization).where(Organization.org_code == code)
        )
        assert org is not None, f"organization {code} not found"
        return str(org.id)


def _user_id(app, username: str) -> str:
    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 str(user.id)


def _role_payload(suffix: str) -> dict:
    return {
        "roleName": f"管理测试-{suffix}",
        "sortNo": 90,
        "status": "ENABLED",
        "remark": None,
        "isDefault": False,
        "manageCategoryEdit": False,
        "manageOrgEdit": False,
        "manageUserEdit": False,
        "managedOrgId": None,
    }


def _create_admin(
    client,
    token: str,
    *,
    username: str,
    organization_id: str | None,
    perm_category: bool = False,
    perm_org: bool = False,
    perm_user: bool = False,
    managed_org_id: str | None = None,
):
    """由内置 admin 先建带管理位的角色，再建 ADMIN 类别账户挂该角色。"""
    role_id = None
    if perm_category or perm_org or perm_user or managed_org_id is not None:
        role_payload = _role_payload(username)
        role_payload.update(
            {
                "manageCategoryEdit": perm_category,
                "manageOrgEdit": perm_org,
                "manageUserEdit": perm_user,
                "managedOrgId": managed_org_id,
            }
        )
        resp = client.post(
            "/api/v1/roles", json=role_payload, headers=_auth(token)
        )
        assert resp.status_code == 201, resp.get_json()
        role_id = resp.get_json()["data"]["id"]
    response = client.post(
        "/api/v1/users",
        json={
            "username": username,
            "realName": username,
            "initialPassword": ADMIN_INITIAL_PASSWORD,
            "organizationId": organization_id,
            "roleCode": "ADMIN",
            "roleId": role_id,
            "securityLevel": "TOP_SECRET",
            "status": "ENABLED",
        },
        headers=_auth(token),
    )
    assert response.status_code == 201, response.get_json()
    return response.get_json()["data"]


def _category_payload(code: str) -> dict:
    return {
        "categoryCode": code,
        "categoryName": f"权限测试{code}",
        "categoryType": "SCENE",
        "parentId": None,
        "sortNo": 10,
        "flatMode": False,
        "typeDisplayName": None,
    }


def _org_payload(code: str, *, parent_id: str | None) -> dict:
    return {
        "orgCode": code,
        "orgName": f"权限测试{code}",
        "parentId": parent_id,
        "sortNo": 10,
        "status": "ENABLED",
    }


def _user_payload(username: str, *, organization_id: str | None) -> dict:
    return {
        "username": username,
        "realName": username,
        "initialPassword": ADMIN_INITIAL_PASSWORD,
        "organizationId": organization_id,
        "roleCode": "USER",
        "roleId": None,
        "securityLevel": "INTERNAL",
        "status": "ENABLED",
    }


# =====================================================================
# 权限位开关
# =====================================================================


@pytest.mark.integration
def test_plain_admin_without_flags_is_readonly_for_directory_writes(
    b2_app, b2_client, login_token, b2_password
):
    """未挂管理位角色的管理员（默认角色管理位全 0）不能写分类/机构/用户。"""
    _create_admin(
        b2_client,
        login_token,
        username="plain_admin",
        organization_id=_org_id(b2_app, "ORG_ROOT"),
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="plain_admin")

    resp = b2_client.post(
        "/api/v1/categories",
        json=_category_payload("PERM_CAT"),
        headers=_auth(token),
    )
    assert resp.status_code == 403
    assert resp.get_json()["code"] == "FORBIDDEN"

    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_PERM_TOP", parent_id=None),
        headers=_auth(token),
    )
    assert resp.status_code == 403

    resp = b2_client.post(
        "/api/v1/users",
        json=_user_payload("perm_target_user", organization_id=None),
        headers=_auth(token),
    )
    assert resp.status_code == 403


@pytest.mark.integration
def test_category_edit_flag_grants_global_category_writes(
    b2_app, b2_client, login_token, b2_password
):
    """分类编辑权限不受锚点范围约束：有权限位即可编辑所有分类。"""
    # 锚定OPS，但分类编辑仍是全局能力
    _create_admin(
        b2_client,
        login_token,
        username="cat_admin",
        organization_id=_org_id(b2_app, "ORG_OPS"),
        perm_category=True,
        managed_org_id=_org_id(b2_app, "ORG_OPS"),
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="cat_admin")
    resp = b2_client.post(
        "/api/v1/categories",
        json=_category_payload("PERM_CAT_OK"),
        headers=_auth(token),
    )
    assert resp.status_code == 201, resp.get_json()
    assert resp.get_json()["data"]["categoryCode"] == "PERM_CAT_OK"


# =====================================================================
# 机构编辑：锚点子树收敛
# =====================================================================


@pytest.mark.integration
def test_anchored_org_editor_scope(
    b2_app, b2_client, login_token, b2_password
):
    """处于 ROOT 但角色锚定 OPS 的机构编辑员：可写 OPS 子树，越范围与顶级一律拒绝。"""
    ops = _org_id(b2_app, "ORG_OPS")
    comms = _org_id(b2_app, "ORG_COMMS")
    root = _org_id(b2_app, "ORG_ROOT")
    _create_admin(
        b2_client,
        login_token,
        username="ops_admin",
        organization_id=root,
        perm_org=True,
        managed_org_id=ops,
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="ops_admin")

    # 子机构创建：范围内 → 201
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_OPS_CHILD", parent_id=ops),
        headers=_auth(token),
    )
    assert resp.status_code == 201, resp.get_json()

    # 编辑范围外机构（COMMS）→ 403
    resp = b2_client.put(
        f"/api/v1/organizations/{comms}",
        json={
            "orgCode": "ORG_COMMS",
            "orgName": "通信部",
            "parentId": root,
            "sortNo": 10,
            "status": "ENABLED",
            "rowVersion": 0,
        },
        headers=_auth(token),
    )
    assert resp.status_code == 403
    assert resp.get_json()["code"] == "FORBIDDEN"

    # 创建顶级机构（parentId 为 null）→ 403（仅全局管理员可建顶级）
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_PERM_TOP", parent_id=None),
        headers=_auth(token),
    )
    assert resp.status_code == 403


@pytest.mark.integration
def test_global_org_editor_can_create_top_level(
    b2_app, b2_client, login_token, b2_password
):
    """角色无锚点（managedOrgId 为 null）的机构编辑员为全局范围，可建顶级机构。"""
    _create_admin(
        b2_client,
        login_token,
        username="global_org_admin",
        organization_id=_org_id(b2_app, "ORG_ROOT"),
        perm_org=True,
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="global_org_admin")
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_PERM_TOP", parent_id=None),
        headers=_auth(token),
    )
    assert resp.status_code == 201, resp.get_json()


# =====================================================================
# 用户编辑：锚点子树收敛 + 列表收敛
# =====================================================================


@pytest.mark.integration
def test_anchored_user_editor_scope(
    b2_app, b2_client, login_token, b2_password
):
    """角色锚定 OPS 的用户编辑员：可增改 OPS 子树用户，越范围一律拒绝。"""
    ops = _org_id(b2_app, "ORG_OPS")
    comms = _org_id(b2_app, "ORG_COMMS")
    _create_admin(
        b2_client,
        login_token,
        username="user_editor",
        organization_id=_org_id(b2_app, "ORG_OPS"),
        perm_user=True,
        managed_org_id=ops,
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="user_editor")

    # 编辑范围内用户（user 在 OPS）→ 200，且详情管理位字段跟随其生效角色
    resp = b2_client.put(
        f"/api/v1/users/{_user_id(b2_app, 'user')}",
        json={
            "realName": "普通用户",
            "organizationId": ops,
            "roleCode": "USER",
            "roleId": None,
            "securityLevel": "INTERNAL",
            "status": "ENABLED",
            "rowVersion": 0,
        },
        headers=_auth(token),
    )
    assert resp.status_code == 200, resp.get_json()
    data = resp.get_json()["data"]
    assert data["permCategoryEdit"] is False
    assert data["permOrgEdit"] is False
    assert data["permUserEdit"] is False
    assert data["managedOrgId"] is None

    # 编辑范围外用户（disabled 在 COMMS）→ 403
    resp = b2_client.put(
        f"/api/v1/users/{_user_id(b2_app, 'disabled')}",
        json={
            "realName": "停用用户",
            "organizationId": comms,
            "roleCode": "USER",
            "roleId": None,
            "securityLevel": "INTERNAL",
            "status": "DISABLED",
            "rowVersion": 0,
        },
        headers=_auth(token),
    )
    assert resp.status_code == 403

    # 创建范围内用户 → 201；创建范围外用户 → 403
    resp = b2_client.post(
        "/api/v1/users",
        json=_user_payload("ops_new_user", organization_id=ops),
        headers=_auth(token),
    )
    assert resp.status_code == 201, resp.get_json()
    resp = b2_client.post(
        "/api/v1/users",
        json=_user_payload("comms_new_user", organization_id=comms),
        headers=_auth(token),
    )
    assert resp.status_code == 403


@pytest.mark.integration
def test_anchored_user_editor_cannot_move_user_out_of_scope(
    b2_app, b2_client, login_token, b2_password
):
    """锚点管理员不可将范围内用户迁出到范围外部门。"""
    ops = _org_id(b2_app, "ORG_OPS")
    comms = _org_id(b2_app, "ORG_COMMS")
    _create_admin(
        b2_client,
        login_token,
        username="user_editor",
        organization_id=ops,
        perm_user=True,
        managed_org_id=ops,
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="user_editor")
    resp = b2_client.put(
        f"/api/v1/users/{_user_id(b2_app, 'user')}",
        json={
            "realName": "普通用户",
            "organizationId": comms,
            "roleCode": "USER",
            "roleId": None,
            "securityLevel": "INTERNAL",
            "status": "ENABLED",
            "rowVersion": 0,
        },
        headers=_auth(token),
    )
    assert resp.status_code == 403
    assert resp.get_json()["code"] == "FORBIDDEN"


@pytest.mark.integration
def test_user_list_converges_to_managed_scope(
    b2_app, b2_client, login_token, b2_password
):
    """用户列表按锚点子树收敛；全局管理员与内置 admin 不受限。"""
    ops = _org_id(b2_app, "ORG_OPS")
    # 管理员身属 ROOT、角色锚定 OPS（身处 A 部门管理 B 部门）：
    # 其本人不在 OPS 子树内，列表应仅见被管子树用户
    _create_admin(
        b2_client,
        login_token,
        username="ops_viewer",
        organization_id=_org_id(b2_app, "ORG_ROOT"),
        perm_user=True,
        managed_org_id=ops,
    )
    scoped_token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="ops_viewer")

    resp = b2_client.get(
        "/api/v1/users?page=1&pageSize=50", headers=_auth(scoped_token)
    )
    page = resp.get_json()["data"]
    # OPS 子树内仅有 conftest 的 user（admin 在 ROOT、disabled 在 COMMS 均不可见）
    assert page["total"] == 1
    assert [item["username"] for item in page["items"]] == ["user"]

    # 内置 admin（超管）不受范围约束，可见全部用户
    resp = b2_client.get(
        "/api/v1/users?page=1&pageSize=50", headers=_auth(login_token)
    )
    usernames = {item["username"] for item in resp.get_json()["data"]["items"]}
    assert {"admin", "user", "disabled", "ops_viewer"} <= usernames


# =====================================================================
# 契约校验
# =====================================================================


@pytest.mark.integration
def test_admin_perm_contract_rejections(b2_app, b2_client, login_token):
    """用户接口不接受管理位字段；角色锚点无效、权限位非布尔均 400。"""
    root = _org_id(b2_app, "ORG_ROOT")

    # 用户 payload 携带管理位字段（管理位已移至角色）→ 400（字段集外）
    payload = _user_payload("perm_user_cat", organization_id=root)
    payload["permCategoryEdit"] = True
    resp = b2_client.post("/api/v1/users", json=payload, headers=_auth(login_token))
    assert resp.status_code == 400
    assert resp.get_json()["code"] == "INVALID_ARGUMENT"

    # 角色锚点指向不存在机构 → 400
    role_payload = _role_payload("BAD_ANCHOR")
    role_payload["managedOrgId"] = "999999"
    resp = b2_client.post(
        "/api/v1/roles", json=role_payload, headers=_auth(login_token)
    )
    assert resp.status_code == 400

    # 角色权限位非布尔 → 400
    role_payload = _role_payload("BAD_FLAG")
    role_payload["manageUserEdit"] = "yes"
    resp = b2_client.post(
        "/api/v1/roles", json=role_payload, headers=_auth(login_token)
    )
    assert resp.status_code == 400


@pytest.mark.integration
def test_downgrade_admin_keeps_role_binding(
    b2_app, b2_client, login_token
):
    """ADMIN 降级为 USER 后角色关联保留，管理位字段继续跟随生效角色。"""
    root = _org_id(b2_app, "ORG_ROOT")
    ops = _org_id(b2_app, "ORG_OPS")
    created = _create_admin(
        b2_client,
        login_token,
        username="downgrade_admin",
        organization_id=root,
        perm_category=True,
        perm_user=True,
        managed_org_id=ops,
    )
    user_id = created["id"]
    resp = b2_client.put(
        f"/api/v1/users/{user_id}",
        json={
            "realName": "downgrade_admin",
            "organizationId": root,
            "roleCode": "USER",
            "roleId": created["roleId"],
            "securityLevel": "TOP_SECRET",
            "status": "ENABLED",
            "rowVersion": 0,
        },
        headers=_auth(login_token),
    )
    assert resp.status_code == 200, resp.get_json()
    data = resp.get_json()["data"]
    assert data["roleCode"] == "USER"
    # 管理位挂角色：降级不清零，字段继续反映生效角色
    assert data["permCategoryEdit"] is True
    assert data["permUserEdit"] is True
    assert data["managedOrgId"] == ops


@pytest.mark.integration
def test_self_anchor_follows_admin_own_org(
    b2_app, b2_client, login_token, b2_password
):
    """锚点 "SELF"（落库哨兵 0）跟随管理员自身部门：身在 OPS 即管 OPS 子树。"""
    ops = _org_id(b2_app, "ORG_OPS")
    comms = _org_id(b2_app, "ORG_COMMS")
    root = _org_id(b2_app, "ORG_ROOT")
    _create_admin(
        b2_client,
        login_token,
        username="self_admin",
        organization_id=ops,
        perm_org=True,
        managed_org_id="SELF",
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="self_admin")

    # 角色列表与用户详情均按 "SELF" 哨兵序列化
    resp = b2_client.get(
        "/api/v1/roles?keyword=管理测试-self_admin",
        headers=_auth(login_token),
    )
    assert resp.status_code == 200, resp.get_json()
    items = resp.get_json()["data"]["items"]
    assert [item["managedOrgId"] for item in items] == ["SELF"]
    resp = b2_client.get("/api/v1/users/me", headers=_auth(token))
    assert resp.status_code == 200, resp.get_json()
    assert resp.get_json()["data"]["managedOrgId"] == "SELF"

    # 身在 OPS：子机构创建（范围内）→ 201
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_OPS_SELF_CHILD", parent_id=ops),
        headers=_auth(token),
    )
    assert resp.status_code == 201, resp.get_json()

    # 编辑范围外机构（COMMS）→ 403
    resp = b2_client.put(
        f"/api/v1/organizations/{comms}",
        json={
            "orgCode": "ORG_COMMS",
            "orgName": "通信部",
            "parentId": root,
            "sortNo": 10,
            "status": "ENABLED",
            "rowVersion": 0,
        },
        headers=_auth(token),
    )
    assert resp.status_code == 403

    # 创建顶级机构 → 403（仅全局管理员可建顶级）
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_SELF_TOP", parent_id=None),
        headers=_auth(token),
    )
    assert resp.status_code == 403


@pytest.mark.integration
def test_self_anchor_without_org_has_no_scope(
    b2_app, b2_client, login_token, b2_password
):
    """锚点 "SELF" 且管理员自身无部门：管理范围为空，任何机构写操作拒绝。"""
    ops = _org_id(b2_app, "ORG_OPS")
    _create_admin(
        b2_client,
        login_token,
        username="orgless_admin",
        organization_id=None,
        perm_org=True,
        managed_org_id="SELF",
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="orgless_admin")
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_ORGLESS_CHILD", parent_id=ops),
        headers=_auth(token),
    )
    assert resp.status_code == 403