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

"""B9 机构与用户维护写操作测试。

覆盖本期新增/重构的逻辑：
- 自我保护（admin 不能编辑自己角色/状态、删除、重置密码）
- 最后一个活跃管理员守卫（统一覆盖删除/降级/停用）
- row_version 乐观锁冲突
- 机构循环检测（parentId 指向后代）
- 机构占用守卫（存在子机构或挂载用户）
- 级联路径更新 + 路径长度预校验
- 软删除后同名重建（active_marker 复合唯一索引）
- auth_version 在角色变更/停用/重置密码场景下的递增
"""

from __future__ import annotations

import pytest

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


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


def _login(client, password: str, *, username: str = "admin") -> 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) -> int:
    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 org.id


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


def _create_org(
    client,
    token: str,
    *,
    code: str,
    name: str,
    parent_id: str | None = None,
    sort_no: int = 100,
    status: str = "ENABLED",
):
    return client.post(
        "/api/v1/organizations",
        json={
            "orgCode": code,
            "orgName": name,
            "parentId": parent_id,
            "sortNo": sort_no,
            "status": status,
        },
        headers=_auth(token),
    )


def _update_org(
    client,
    token: str,
    org_id: str,
    *,
    name: str,
    parent_id: str | None,
    sort_no: int,
    status: str,
    row_version: int,
):
    return client.put(
        f"/api/v1/organizations/{org_id}",
        json={
            "orgName": name,
            "parentId": parent_id,
            "sortNo": sort_no,
            "status": status,
            "rowVersion": row_version,
        },
        headers=_auth(token),
    )


def _delete_org(client, token: str, org_id: str, row_version: int):
    return client.delete(
        f"/api/v1/organizations/{org_id}?rowVersion={row_version}",
        headers=_auth(token),
    )


def _create_user(
    client,
    token: str,
    *,
    username: str,
    real_name: str,
    password: str = "Initial@2026",
    organization_id: str | None = None,
    role_code: str = "USER",
    security_level: str = "INTERNAL",
    status: str = "ENABLED",
):
    return client.post(
        "/api/v1/users",
        json={
            "username": username,
            "realName": real_name,
            "initialPassword": password,
            "organizationId": organization_id,
            "roleCode": role_code,
            "roleId": None,
            "securityLevel": security_level,
            "status": status,
        },
        headers=_auth(token),
    )


def _update_user(
    client,
    token: str,
    user_id: str,
    *,
    real_name: str,
    organization_id: str | None,
    role_code: str,
    security_level: str,
    status: str,
    row_version: int,
):
    return client.put(
        f"/api/v1/users/{user_id}",
        json={
            "realName": real_name,
            "organizationId": organization_id,
            "roleCode": role_code,
            "roleId": None,
            "securityLevel": security_level,
            "status": status,
            "rowVersion": row_version,
        },
        headers=_auth(token),
    )


def _reset_password(
    client, token: str, user_id: str, *, new_password: str, row_version: int
):
    return client.post(
        f"/api/v1/users/{user_id}/reset-password",
        json={"newPassword": new_password, "rowVersion": row_version},
        headers=_auth(token),
    )


# =====================================================================
# 机构生命周期 + 软删除 + 复合唯一索引
# =====================================================================


@pytest.mark.integration
def test_create_organization_success_writes_audit(b2_app, b2_client, login_token):
    response = _create_org(
        b2_client,
        login_token,
        code="ORG_B9_NEW",
        name="B9新增机构",
        parent_id=str(_org_id(b2_app, "ORG_ROOT")),
        sort_no=99,
    )
    assert response.status_code == 201, response.get_json()
    payload = response.get_json()["data"]
    assert payload["orgCode"] == "ORG_B9_NEW"
    assert payload["orgPath"] == "/机关/B9新增机构"
    assert payload["parentId"] == str(_org_id(b2_app, "ORG_ROOT"))
    with b2_app.app_context():
        audit = db.session.scalar(
            db.select(AuditLog).where(AuditLog.action_type == "CREATE_ORGANIZATION")
        )
        assert audit is not None
        assert audit.target_id == int(payload["id"])


@pytest.mark.integration
def test_create_organization_rejects_duplicate_code(b2_client, login_token):
    response = _create_org(
        b2_client,
        login_token,
        code="ORG_ROOT",
        name="重名机构",
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "DATA_VERSION_CONFLICT"


@pytest.mark.integration
def test_create_organization_rejects_disabled_parent(b2_client, login_token, b2_app):
    disabled_id = str(_org_id(b2_app, "ORG_DISABLED"))
    response = _create_org(
        b2_client,
        login_token,
        code="ORG_B9_UNDER_DISABLED",
        name="挂在停用机构下",
        parent_id=disabled_id,
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


@pytest.mark.integration
def test_soft_deleted_org_code_can_be_reused(b2_client, login_token, b2_app):
    """active_marker 复合唯一索引：软删除后允许重建同名条目。"""
    create_resp = _create_org(
        b2_client, login_token, code="ORG_B9_REVIVE", name="原机构"
    )
    assert create_resp.status_code == 201
    org_id = create_resp.get_json()["data"]["id"]
    row_version = create_resp.get_json()["data"]["rowVersion"]

    delete_resp = _delete_org(b2_client, login_token, org_id, row_version)
    assert delete_resp.status_code == 200

    rebuild_resp = _create_org(
        b2_client, login_token, code="ORG_B9_REVIVE", name="重建机构"
    )
    assert rebuild_resp.status_code == 201
    assert rebuild_resp.get_json()["data"]["orgName"] == "重建机构"


# =====================================================================
# 机构更新：row_version / 循环 / 级联路径 / 长度上限
# =====================================================================


@pytest.mark.integration
def test_update_organization_row_version_conflict(b2_client, login_token, b2_app):
    create_resp = _create_org(
        b2_client, login_token, code="ORG_B9_VC", name="版本冲突"
    )
    org_id = create_resp.get_json()["data"]["id"]

    response = _update_org(
        b2_client,
        login_token,
        org_id,
        name="版本冲突",
        parent_id=None,
        sort_no=10,
        status="ENABLED",
        row_version=99,
    )
    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_update_organization_rejects_cycle_to_descendant(
    b2_client, login_token, b2_app
):
    """parentId 不能指向自己的后代（否则会形成环）。"""
    grandparent_resp = _create_org(
        b2_client, login_token, code="ORG_B9_GP", name="祖父"
    )
    gp_id = grandparent_resp.get_json()["data"]["id"]
    parent_resp = _create_org(
        b2_client,
        login_token,
        code="ORG_B9_P",
        name="父",
        parent_id=gp_id,
    )
    p_id = parent_resp.get_json()["data"]["id"]
    child_resp = _create_org(
        b2_client,
        login_token,
        code="ORG_B9_C",
        name="子",
        parent_id=p_id,
    )
    c_id = child_resp.get_json()["data"]["id"]

    # 尝试把"父"挂到"子"下面 -> 应当拒绝
    response = _update_org(
        b2_client,
        login_token,
        p_id,
        name="父",
        parent_id=c_id,
        sort_no=10,
        status="ENABLED",
        row_version=0,
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


@pytest.mark.integration
def test_update_organization_cascades_descendant_paths(
    b2_client, login_token, b2_app
):
    """重命名中间节点时，所有后代的 org_path 必须同步刷新。"""
    a_resp = _create_org(b2_client, login_token, code="ORG_B9_CASCADE_A", name="原名A")
    a_id = a_resp.get_json()["data"]["id"]
    b_resp = _create_org(
        b2_client,
        login_token,
        code="ORG_B9_CASCADE_B",
        name="原名B",
        parent_id=a_id,
    )
    b_id = b_resp.get_json()["data"]["id"]
    c_resp = _create_org(
        b2_client,
        login_token,
        code="ORG_B9_CASCADE_C",
        name="原名C",
        parent_id=b_id,
    )
    c_id = c_resp.get_json()["data"]["id"]

    rename_resp = _update_org(
        b2_client,
        login_token,
        a_id,
        name="新名A",
        parent_id=None,
        sort_no=10,
        status="ENABLED",
        row_version=0,
    )
    assert rename_resp.status_code == 200

    with b2_app.app_context():
        a = db.session.get(Organization, int(a_id))
        b = db.session.get(Organization, int(b_id))
        c = db.session.get(Organization, int(c_id))
        assert a.org_path == "/新名A"
        assert b.org_path == "/新名A/原名B"
        assert c.org_path == "/新名A/原名B/原名C"
        # row_version 都被递增
        assert a.row_version == 1
        assert b.row_version == 1
        assert c.row_version == 1


@pytest.mark.integration
def test_update_organization_path_length_pre_validation(
    b2_client, login_token, b2_app
):
    """M-4：路径长度溢出时整个事务回滚，不留半成品状态。"""
    a_resp = _create_org(
        b2_client, login_token, code="ORG_B9_LEN_A", name="A"
    )
    a_id = a_resp.get_json()["data"]["id"]
    b_resp = _create_org(
        b2_client, login_token, code="ORG_B9_LEN_B", name="B", parent_id=a_id
    )
    b_id = b_resp.get_json()["data"]["id"]

    # 制造一个超长名（>1000 字符），更新 B 应当整体回滚
    long_name = "X" * 1100
    response = _update_org(
        b2_client,
        login_token,
        b_id,
        name=long_name,
        parent_id=a_id,
        sort_no=10,
        status="ENABLED",
        row_version=0,
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"

    # B 的原状态保持不变
    with b2_app.app_context():
        b = db.session.get(Organization, int(b_id))
        assert b.org_name == "B"
        assert b.org_path == "/A/B"
        assert b.row_version == 0


# =====================================================================
# 机构删除：占用守卫
# =====================================================================


@pytest.mark.integration
def test_delete_organization_blocks_when_has_children(b2_client, login_token):
    parent_resp = _create_org(
        b2_client, login_token, code="ORG_B9_DEL_PARENT", name="父"
    )
    p_id = parent_resp.get_json()["data"]["id"]
    _create_org(
        b2_client,
        login_token,
        code="ORG_B9_DEL_CHILD",
        name="子",
        parent_id=p_id,
    )

    response = _delete_org(b2_client, login_token, p_id, row_version=0)
    assert response.status_code == 409
    body = response.get_json()
    assert body["code"] == "ORGANIZATION_IN_USE"
    assert body["details"]["childOrganizationCount"] == 1
    assert body["details"]["userCount"] == 0


@pytest.mark.integration
def test_delete_organization_blocks_when_has_users(
    b2_client, login_token, b2_app
):
    org_resp = _create_org(
        b2_client, login_token, code="ORG_B9_DEL_WITH_USER", name="占用机构"
    )
    org_id = org_resp.get_json()["data"]["id"]
    user_resp = _create_user(
        b2_client,
        login_token,
        username="b9_user_on_org",
        real_name="占用用户",
        organization_id=org_id,
    )
    assert user_resp.status_code == 201

    response = _delete_org(b2_client, login_token, org_id, row_version=0)
    assert response.status_code == 409
    body = response.get_json()
    assert body["code"] == "ORGANIZATION_IN_USE"
    assert body["details"]["childOrganizationCount"] == 0
    assert body["details"]["userCount"] == 1


@pytest.mark.integration
def test_delete_organization_row_version_conflict(b2_client, login_token):
    org_resp = _create_org(
        b2_client, login_token, code="ORG_B9_DEL_VC", name="版本"
    )
    org_id = org_resp.get_json()["data"]["id"]

    response = _delete_org(b2_client, login_token, org_id, row_version=99)
    assert response.status_code == 409
    assert response.get_json()["code"] == "DATA_VERSION_CONFLICT"


# =====================================================================
# 用户生命周期 + 自我保护 + 最后管理员守卫
# =====================================================================


@pytest.mark.integration
def test_create_user_success_and_duplicate(b2_client, login_token, b2_app):
    org_id = str(_org_id(b2_app, "ORG_ROOT"))
    first = _create_user(
        b2_client,
        login_token,
        username="b9_dup",
        real_name="重名测试",
        organization_id=org_id,
    )
    assert first.status_code == 201

    second = _create_user(
        b2_client,
        login_token,
        username="b9_dup",
        real_name="重名测试2",
        organization_id=org_id,
    )
    assert second.status_code == 409
    assert second.get_json()["code"] == "DATA_VERSION_CONFLICT"


@pytest.mark.integration
def test_update_user_row_version_conflict(b2_client, login_token, b2_app):
    user_id = str(_user_id(b2_app, "user"))
    response = _update_user(
        b2_client,
        login_token,
        user_id,
        real_name="普通用户",
        organization_id=str(_org_id(b2_app, "ORG_OPS")),
        role_code="USER",
        security_level="SECRET",
        status="ENABLED",
        row_version=99,
    )
    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_self_protection_admin_cannot_edit_own_role_or_status(
    b2_app, b2_client, b2_password
):
    """admin 不能编辑自己的角色/状态（避免误锁自己）。"""
    token = _login(b2_client, b2_password, username="admin")
    admin_id = str(_user_id(b2_app, "admin"))

    # 试图改自己的角色
    role_resp = _update_user(
        b2_client,
        token,
        admin_id,
        real_name="系统管理员",
        organization_id=str(_org_id(b2_app, "ORG_ROOT")),
        role_code="USER",
        security_level="TOP_SECRET",
        status="ENABLED",
        row_version=0,
    )
    assert role_resp.status_code == 403
    assert role_resp.get_json()["code"] == "USER_SELF_PROTECTION"

    # 试图停用自己
    disable_resp = _update_user(
        b2_client,
        token,
        admin_id,
        real_name="系统管理员",
        organization_id=str(_org_id(b2_app, "ORG_ROOT")),
        role_code="ADMIN",
        security_level="TOP_SECRET",
        status="DISABLED",
        row_version=0,
    )
    assert disable_resp.status_code == 403
    assert disable_resp.get_json()["code"] == "USER_SELF_PROTECTION"


@pytest.mark.integration
def test_self_protection_admin_cannot_delete_self(b2_app, b2_client, b2_password):
    token = _login(b2_client, b2_password, username="admin")
    admin_id = str(_user_id(b2_app, "admin"))
    response = b2_client.delete(
        f"/api/v1/users/{admin_id}?rowVersion=0",
        headers=_auth(token),
    )
    assert response.status_code == 403
    assert response.get_json()["code"] == "USER_SELF_PROTECTION"


@pytest.mark.integration
def test_self_protection_admin_cannot_reset_own_password(
    b2_app, b2_client, b2_password
):
    token = _login(b2_client, b2_password, username="admin")
    admin_id = str(_user_id(b2_app, "admin"))
    response = _reset_password(
        b2_client, token, admin_id, new_password="NewPassword@2026", row_version=0
    )
    assert response.status_code == 403
    assert response.get_json()["code"] == "USER_SELF_PROTECTION"


@pytest.mark.integration
def test_last_admin_protection_blocks_delete(b2_client, login_token, b2_app):
    """只有一个活跃 admin 时，删除该 admin 应当被拒绝。"""
    admin_id = str(_user_id(b2_app, "admin"))
    response = b2_client.delete(
        f"/api/v1/users/{admin_id}?rowVersion=0",
        headers=_auth(login_token),
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "DATA_VERSION_CONFLICT"


@pytest.mark.integration
def test_last_admin_protection_blocks_downgrade(b2_client, login_token, b2_app):
    """只有一个活跃 admin 时，将其降级为 USER 应当被拒绝。"""
    admin_id = str(_user_id(b2_app, "admin"))
    response = _update_user(
        b2_client,
        login_token,
        admin_id,
        real_name="系统管理员",
        organization_id=str(_org_id(b2_app, "ORG_ROOT")),
        role_code="USER",
        security_level="TOP_SECRET",
        status="ENABLED",
        row_version=0,
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "DATA_VERSION_CONFLICT"


@pytest.mark.integration
def test_last_admin_protection_blocks_disable(b2_client, login_token, b2_app):
    """只有一个活跃 admin 时，停用该 admin 应当被拒绝。"""
    admin_id = str(_user_id(b2_app, "admin"))
    response = _update_user(
        b2_client,
        login_token,
        admin_id,
        real_name="系统管理员",
        organization_id=str(_org_id(b2_app, "ORG_ROOT")),
        role_code="ADMIN",
        security_level="TOP_SECRET",
        status="DISABLED",
        row_version=0,
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "DATA_VERSION_CONFLICT"


@pytest.mark.integration
def test_disabled_admin_can_be_deleted_after_losing_active_status(
    b2_client, login_token, b2_app
):
    """H-1 隐藏 bug 验证：先创建第二个 admin，再降级原 admin（合法），
    接着停用原 admin（此时它已不是活跃 admin），最后删除应当通过。
    这条测试保护 _ensure_not_last_active_admin 中"非活跃 admin 放行"的分支。
    """
    # 1) 新建第二个 admin
    second_resp = _create_user(
        b2_client,
        login_token,
        username="b9_second_admin",
        real_name="第二管理员",
        organization_id=str(_org_id(b2_app, "ORG_ROOT")),
        role_code="ADMIN",
        security_level="TOP_SECRET",
        status="ENABLED",
    )
    assert second_resp.status_code == 201
    second_id = second_resp.get_json()["data"]["id"]

    # 2) 现在可以把原 admin 降级（仍有第二 admin 守护）
    admin_id = str(_user_id(b2_app, "admin"))
    downgrade_resp = _update_user(
        b2_client,
        login_token,
        admin_id,
        real_name="系统管理员",
        organization_id=str(_org_id(b2_app, "ORG_ROOT")),
        role_code="USER",
        security_level="TOP_SECRET",
        status="ENABLED",
        row_version=0,
    )
    assert downgrade_resp.status_code == 200, downgrade_resp.get_json()
    new_version = downgrade_resp.get_json()["data"]["rowVersion"]

    # 3) 停用原 admin（此时它已经不是 admin，最后管理员守卫不应触发）
    disable_resp = _update_user(
        b2_client,
        login_token,
        admin_id,
        real_name="系统管理员",
        organization_id=str(_org_id(b2_app, "ORG_ROOT")),
        role_code="USER",
        security_level="TOP_SECRET",
        status="DISABLED",
        row_version=new_version,
    )
    assert disable_resp.status_code == 200, disable_resp.get_json()
    new_version = disable_resp.get_json()["data"]["rowVersion"]

    # 4) 最后删除原 admin —— _ensure_not_last_active_admin 必须放行
    delete_resp = b2_client.delete(
        f"/api/v1/users/{admin_id}?rowVersion={new_version}",
        headers=_auth(login_token),
    )
    assert delete_resp.status_code == 200, delete_resp.get_json()


# =====================================================================
# auth_version 在用户写操作中的递增策略
# =====================================================================


@pytest.mark.integration
def test_auth_version_bumped_on_role_change(b2_app, b2_client, login_token):
    user_id = str(_user_id(b2_app, "user"))
    with b2_app.app_context():
        before_version = db.session.scalar(
            db.select(User.auth_version).where(User.username == "user")
        )
    response = _update_user(
        b2_client,
        login_token,
        user_id,
        real_name="普通用户",
        organization_id=str(_org_id(b2_app, "ORG_OPS")),
        role_code="ADMIN",
        security_level="SECRET",
        status="ENABLED",
        row_version=0,
    )
    assert response.status_code == 200
    with b2_app.app_context():
        after_version = db.session.scalar(
            db.select(User.auth_version).where(User.username == "user")
        )
    assert after_version == before_version + 1


@pytest.mark.integration
def test_auth_version_bumped_on_disable(b2_app, b2_client, login_token):
    user_id = str(_user_id(b2_app, "user"))
    with b2_app.app_context():
        before_version = db.session.scalar(
            db.select(User.auth_version).where(User.username == "user")
        )
    response = _update_user(
        b2_client,
        login_token,
        user_id,
        real_name="普通用户",
        organization_id=str(_org_id(b2_app, "ORG_OPS")),
        role_code="USER",
        security_level="SECRET",
        status="DISABLED",
        row_version=0,
    )
    assert response.status_code == 200
    with b2_app.app_context():
        after_version = db.session.scalar(
            db.select(User.auth_version).where(User.username == "user")
        )
    assert after_version == before_version + 1


@pytest.mark.integration
def test_auth_version_not_bumped_on_realname_only(b2_app, b2_client, login_token):
    """只改 realName 不应让旧 token 失效。"""
    user_id = str(_user_id(b2_app, "user"))
    with b2_app.app_context():
        before_version = db.session.scalar(
            db.select(User.auth_version).where(User.username == "user")
        )
    response = _update_user(
        b2_client,
        login_token,
        user_id,
        real_name="普通用户改名",
        organization_id=str(_org_id(b2_app, "ORG_OPS")),
        role_code="USER",
        security_level="SECRET",
        status="ENABLED",
        row_version=0,
    )
    assert response.status_code == 200
    with b2_app.app_context():
        after_version = db.session.scalar(
            db.select(User.auth_version).where(User.username == "user")
        )
    assert after_version == before_version


@pytest.mark.integration
def test_auth_version_bumped_on_password_reset(b2_app, b2_client, login_token):
    user_id = str(_user_id(b2_app, "user"))
    with b2_app.app_context():
        before_version = db.session.scalar(
            db.select(User.auth_version).where(User.username == "user")
        )
    response = _reset_password(
        b2_client,
        login_token,
        user_id,
        new_password="BrandNew@2026",
        row_version=0,
    )
    assert response.status_code == 200
    assert response.get_json()["data"]["authVersion"] == before_version + 1