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

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

覆盖：
- 角色管理位（分类/机构/用户编辑）对相应写接口的放行与拒绝；
- 管理范围锚点（角色 managedOrgId）的子树收敛（机构/用户增删改与用户列表）；
- 锚点管理员不可将用户迁出管理子树；
- 用户接口不再接受管理位字段、角色锚点无效与权限位非布尔的契约拒绝；
- ADMIN 降级 USER 后角色关联保留、管理位字段跟随生效角色；
- 锚点 "SELF"（落库哨兵 0）跟随管理员自身部门，无部门时无管理范围；
- 用户级管理范围三态（契约1.30）：角色锚点 SELF 时回落用户级设置；
- 内置 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,
        "manageSystemManage": False,
        "manageAuditLog": False,
        "manageRoleEdit": 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,
    perm_role: bool = False,
    managed_org_id: str | None = None,
):
    """由内置 admin 先建带管理位的角色，再建 ADMIN 类别账户挂该角色。"""
    role_id = None
    if perm_category or perm_org or perm_user or perm_role or managed_org_id is not None:
        role_payload = _role_payload(username)
        role_payload.update(
            {
                "manageCategoryEdit": perm_category,
                "manageOrgEdit": perm_org,
                "manageUserEdit": perm_user,
                "manageRoleEdit": perm_role,
                "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

    # 角色写接口（契约1.25）：无角色编辑位的管理员建角色同样拒绝
    resp = b2_client.post(
        "/api/v1/roles",
        json=_role_payload("plain_admin"),
        headers=_auth(token),
    )
    assert resp.status_code == 403
    assert resp.get_json()["code"] == "FORBIDDEN"


@pytest.mark.integration
def test_role_edit_flag_grants_role_writes(
    b2_app, b2_client, login_token, b2_password
):
    """角色编辑位（契约1.25）：有位的管理员可建/改角色，位本身随载荷落库。"""
    _create_admin(
        b2_client,
        login_token,
        username="role_admin",
        organization_id=_org_id(b2_app, "ORG_ROOT"),
        perm_role=True,
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="role_admin")

    created = b2_client.post(
        "/api/v1/roles",
        json=_role_payload("role_admin_child"),
        headers=_auth(token),
    )
    assert created.status_code == 201, created.get_json()
    role_id = created.get_json()["data"]["id"]
    assert created.get_json()["data"]["manageRoleEdit"] is False

    updated = b2_client.put(
        f"/api/v1/roles/{role_id}",
        json={
            **_role_payload("role_admin_child"),
            "roleName": "管理测试-role_admin_child-改",
            "manageUserEdit": True,
            "rowVersion": 0,
        },
        headers=_auth(token),
    )
    assert updated.status_code == 200, updated.get_json()
    assert updated.get_json()["data"]["manageUserEdit"] is True


@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_user_list_rejects_org_outside_managed_scope(
    b2_app, b2_client, login_token, b2_password
):
    """显式请求管理范围外机构：403 无权限，而非空页（避免误显示"机构无用户"）。"""
    ops = _org_id(b2_app, "ORG_OPS")
    comms = _org_id(b2_app, "ORG_COMMS")
    _create_admin(
        b2_client,
        login_token,
        username="ops_viewer2",
        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_viewer2")

    # 范围外机构（不带/带 includeDescendants）→ 403
    for query in (
        f"/api/v1/users?organizationId={comms}",
        f"/api/v1/users?organizationId={comms}&includeDescendants=true",
    ):
        resp = b2_client.get(query, headers=_auth(scoped_token))
        assert resp.status_code == 403
        assert resp.get_json()["code"] == "FORBIDDEN"

    # 范围内机构（含后代展开）仍正常返回
    resp = b2_client.get(
        f"/api/v1/users?organizationId={ops}&includeDescendants=true",
        headers=_auth(scoped_token),
    )
    assert resp.status_code == 200
    assert resp.get_json()["data"]["total"] == 1

    # 全局管理员（内置 admin）不受范围约束
    resp = b2_client.get(
        f"/api/v1/users?organizationId={comms}", headers=_auth(login_token)
    )
    assert resp.status_code == 200


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


@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
    # managedOrgId 为用户级三态（契约1.30③）：创建未提交缺省落 SELF，不再取生效角色锚点
    assert data["managedOrgId"] == "SELF"


@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


def _update_user(client, token: str, user_id: str, *, row_version: int, **overrides):
    """PUT /users/{id} 全量载荷；overrides 追加或覆盖可选字段（如 managedOrgId）。"""
    payload = {
        "realName": overrides.pop("realName", "scope-user"),
        "organizationId": overrides.pop("organizationId", None),
        "roleCode": overrides.pop("roleCode", "USER"),
        "roleId": overrides.pop("roleId", None),
        "securityLevel": overrides.pop("securityLevel", "INTERNAL"),
        "status": overrides.pop("status", "ENABLED"),
        "rowVersion": row_version,
    }
    payload.update(overrides)
    return client.put(f"/api/v1/users/{user_id}", json=payload, headers=_auth(token))


@pytest.mark.integration
def test_user_level_managed_org_field(b2_app, b2_client, login_token):
    """用户级管理范围三态字段（契约1.30②）：create 显式/缺省、update 省略保持与显式置空。"""
    ops = _org_id(b2_app, "ORG_OPS")

    # create 显式机构 ID → 摘要原样返回
    payload = _user_payload("userlvl_anchor", organization_id=ops)
    payload["managedOrgId"] = ops
    resp = b2_client.post("/api/v1/users", json=payload, headers=_auth(login_token))
    assert resp.status_code == 201, resp.get_json()
    data = resp.get_json()["data"]
    assert data["managedOrgId"] == ops
    user_id = data["id"]

    # create 缺省 → 落 SELF（哨兵 0 序列化还原）
    payload = _user_payload("userlvl_self", organization_id=ops)
    resp = b2_client.post("/api/v1/users", json=payload, headers=_auth(login_token))
    assert resp.status_code == 201, resp.get_json()
    assert resp.get_json()["data"]["managedOrgId"] == "SELF"

    # 非法值契约拒绝：非数字、哨兵 "0"、不存在的机构
    for bad in ("abc", "0", "999999"):
        payload = _user_payload(f"userlvl_bad_{bad}", organization_id=ops)
        payload["managedOrgId"] = bad
        resp = b2_client.post(
            "/api/v1/users", json=payload, headers=_auth(login_token)
        )
        assert resp.status_code == 400, (bad, resp.get_json())
        assert resp.get_json()["code"] == "INVALID_ARGUMENT"

    # update 省略 → 保持原值（仍是 ops）
    resp = _update_user(b2_client, login_token, user_id, row_version=0)
    assert resp.status_code == 200, resp.get_json()
    assert resp.get_json()["data"]["managedOrgId"] == ops

    # update 显式 null → 全局；再置 "SELF" → 跟随所在机构
    resp = _update_user(
        b2_client, login_token, user_id, row_version=1, managedOrgId=None
    )
    assert resp.status_code == 200, resp.get_json()
    assert resp.get_json()["data"]["managedOrgId"] is None
    resp = _update_user(
        b2_client, login_token, user_id, row_version=2, managedOrgId="SELF"
    )
    assert resp.status_code == 200, resp.get_json()
    assert resp.get_json()["data"]["managedOrgId"] == "SELF"

    # 列表摘要与详情同口径
    resp = b2_client.get(
        f"/api/v1/users?keyword=userlvl_anchor&page=1&pageSize=10",
        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"]


@pytest.mark.integration
def test_role_self_anchor_falls_back_to_user_level(
    b2_app, b2_client, login_token
):
    """角色锚点 SELF 回落用户级设置（契约1.30④）：null=全局、机构ID=该子树、SELF=所在机构。"""
    ops = _org_id(b2_app, "ORG_OPS")
    created = _create_admin(
        b2_client,
        login_token,
        username="userlvl_admin",
        organization_id=None,
        perm_org=True,
        managed_org_id="SELF",
    )
    user_id = created["id"]
    assert created["managedOrgId"] == "SELF"
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="userlvl_admin")

    # 用户级 SELF + 本人无部门 → 空集：OPS 子机构创建被拒
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_USERLVL_EMPTY", parent_id=ops),
        headers=_auth(token),
    )
    assert resp.status_code == 403

    # 用户级设为 OPS 固定锚点 → 管 OPS 子树；顶级仍不可建
    resp = _update_user(
        b2_client,
        login_token,
        user_id,
        row_version=0,
        realName="userlvl_admin",
        organizationId=None,
        roleCode="ADMIN",
        roleId=created["roleId"],
        securityLevel="TOP_SECRET",
        managedOrgId=ops,
    )
    assert resp.status_code == 200, resp.get_json()
    assert resp.get_json()["data"]["managedOrgId"] == ops
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_USERLVL_OPS_CHILD", parent_id=ops),
        headers=_auth(token),
    )
    assert resp.status_code == 201, resp.get_json()
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_USERLVL_TOP", parent_id=None),
        headers=_auth(token),
    )
    assert resp.status_code == 403

    # 用户级设为全局（null）→ 顶级机构放行；范围调整不 bump authVersion（旧 token 仍有效）
    resp = _update_user(
        b2_client,
        login_token,
        user_id,
        row_version=1,
        realName="userlvl_admin",
        organizationId=None,
        roleCode="ADMIN",
        roleId=created["roleId"],
        securityLevel="TOP_SECRET",
        managedOrgId=None,
    )
    assert resp.status_code == 200, resp.get_json()
    assert resp.get_json()["data"]["managedOrgId"] is None
    resp = b2_client.post(
        "/api/v1/organizations",
        json=_org_payload("ORG_USERLVL_GLOBAL_TOP", parent_id=None),
        headers=_auth(token),
    )
    assert resp.status_code == 201, resp.get_json()


# =====================================================================
# 防提权（契约1.30修订）：目标 ADMIN 生效管理范围不得超出操作者
# =====================================================================


def _create_role_with_flags(
    client, token: str, *, suffix: str, managed_org_id, **flags
) -> str:
    payload = _role_payload(suffix)
    payload.update(flags)
    payload["managedOrgId"] = managed_org_id
    resp = client.post("/api/v1/roles", json=payload, headers=_auth(token))
    assert resp.status_code == 201, resp.get_json()
    return resp.get_json()["data"]["id"]


def _admin_user_payload(username: str, *, role_id, organization_id: str) -> dict:
    return {
        "username": username,
        "realName": username,
        "initialPassword": ADMIN_INITIAL_PASSWORD,
        "organizationId": organization_id,
        "roleCode": "ADMIN",
        "roleId": role_id,
        "securityLevel": "INTERNAL",
        "status": "ENABLED",
    }


@pytest.mark.integration
def test_user_editor_cannot_grant_scope_beyond_own_scope(
    b2_app, b2_client, login_token
):
    """受限用户编辑员（锚点 OPS）不能创建生效范围超出自己的 ADMIN。

    仅当目标 ADMIN 的生效角色具备机构/用户编辑位时校验；
    无管理位角色与 USER 类别不受限。
    """
    ops = _org_id(b2_app, "ORG_OPS")
    comms = _org_id(b2_app, "ORG_COMMS")

    # 预建四枚角色：SELF 锚点 / 全局锚点 / 范围外锚点 / 无管理位
    role_self = _create_role_with_flags(
        b2_client,
        login_token,
        suffix="grant_self",
        managed_org_id="SELF",
        manageUserEdit=True,
    )
    role_global = _create_role_with_flags(
        b2_client,
        login_token,
        suffix="grant_global",
        managed_org_id=None,
        manageUserEdit=True,
    )
    role_comms = _create_role_with_flags(
        b2_client,
        login_token,
        suffix="grant_comms",
        managed_org_id=comms,
        manageUserEdit=True,
    )
    role_plain = _create_role_with_flags(
        b2_client,
        login_token,
        suffix="grant_plain",
        managed_org_id=None,
    )

    _create_admin(
        b2_client,
        login_token,
        username="grant_editor",
        organization_id=ops,
        perm_user=True,
        managed_org_id=ops,
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="grant_editor")

    def _create(name, *, role_id, managed_org_id=...):
        payload = _admin_user_payload(name, role_id=role_id, organization_id=ops)
        if managed_org_id is not ...:
            payload["managedOrgId"] = managed_org_id
        return b2_client.post(
            "/api/v1/users", json=payload, headers=_auth(token)
        )

    # 角色锚点 SELF + 用户级全局（null）→ 提权，403
    resp = _create("esc_global", role_id=role_self, managed_org_id=None)
    assert resp.status_code == 403
    assert resp.get_json()["code"] == "FORBIDDEN"

    # 生效角色锚点为全局 → 提权，403
    resp = _create("esc_role_global", role_id=role_global)
    assert resp.status_code == 403

    # 角色锚点 SELF + 用户级范围外机构（COMMS）→ 403
    resp = _create("esc_comms", role_id=role_self, managed_org_id=comms)
    assert resp.status_code == 403

    # 生效角色锚点范围外（COMMS）→ 403
    resp = _create("esc_role_comms", role_id=role_comms)
    assert resp.status_code == 403

    # 合法：角色锚点 SELF + 用户级 SELF（缺省）→ 201
    resp = _create("esc_ok_self", role_id=role_self)
    assert resp.status_code == 201, resp.get_json()

    # 合法：角色锚点 SELF + 用户级 OPS → 201
    resp = _create("esc_ok_ops", role_id=role_self, managed_org_id=ops)
    assert resp.status_code == 201, resp.get_json()

    # 合法：ADMIN + 无管理位角色（用户级全局不构成提权）→ 201
    resp = _create("esc_ok_plain", role_id=role_plain, managed_org_id=None)
    assert resp.status_code == 201, resp.get_json()

    # 合法：USER 类别不受限 → 201
    resp = b2_client.post(
        "/api/v1/users",
        json=_user_payload("esc_ok_user", organization_id=ops),
        headers=_auth(token),
    )
    assert resp.status_code == 201, resp.get_json()


@pytest.mark.integration
def test_user_editor_cannot_escalate_via_update(
    b2_app, b2_client, login_token
):
    """受限用户编辑员编辑既有生效范围为全局的 ADMIN：未收敛管理范围即拒绝。"""
    ops = _org_id(b2_app, "ORG_OPS")

    role_self = _create_role_with_flags(
        b2_client,
        login_token,
        suffix="upd_self",
        managed_org_id="SELF",
        manageUserEdit=True,
    )
    role_global = _create_role_with_flags(
        b2_client,
        login_token,
        suffix="upd_global",
        managed_org_id=None,
        manageUserEdit=True,
    )

    # 内置 admin 预建生效范围为全局的 B（角色 SELF + 用户级 null）
    created = b2_client.post(
        "/api/v1/users",
        json={
            **_admin_user_payload("esc_target", role_id=role_self, organization_id=ops),
            "managedOrgId": None,
        },
        headers=_auth(login_token),
    )
    assert created.status_code == 201, created.get_json()
    target_id = created.get_json()["data"]["id"]

    _create_admin(
        b2_client,
        login_token,
        username="upd_editor",
        organization_id=ops,
        perm_user=True,
        managed_org_id=ops,
    )
    token = _login(b2_client, ADMIN_INITIAL_PASSWORD, username="upd_editor")

    # 未提交 managedOrgId：按保持原值（全局）判定 → 403
    resp = _update_user(
        b2_client,
        token,
        target_id,
        row_version=0,
        realName="esc_target",
        organizationId=ops,
        roleCode="ADMIN",
        roleId=role_self,
        securityLevel="INTERNAL",
        status="ENABLED",
    )
    assert resp.status_code == 403
    assert resp.get_json()["code"] == "FORBIDDEN"

    # 换全局锚点管理角色 → 403（校验先于二次认证触发）
    resp = _update_user(
        b2_client,
        token,
        target_id,
        row_version=0,
        realName="esc_target",
        organizationId=ops,
        roleCode="ADMIN",
        roleId=role_global,
        securityLevel="INTERNAL",
        status="ENABLED",
    )
    assert resp.status_code == 403
    assert resp.get_json()["code"] == "FORBIDDEN"

    # 收敛用户级范围到 SELF（跟随所在机构 OPS）→ 200
    resp = _update_user(
        b2_client,
        token,
        target_id,
        row_version=0,
        realName="esc_target",
        organizationId=ops,
        roleCode="ADMIN",
        roleId=role_self,
        securityLevel="INTERNAL",
        status="ENABLED",
        managedOrgId="SELF",
    )
    assert resp.status_code == 200, resp.get_json()
    assert resp.get_json()["data"]["managedOrgId"] == "SELF"