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

"""契约1.27 机构/用户/角色列表拖动重排测试。

覆盖：
- 机构：同级槽位置换（步长10重编号 sort_no）、范围拒绝、父级不存在、权限
- 用户：全体槽位置换、列表顺序联动、重复提交幂等、无效ID、权限
- 角色：创建后槽位置换、无效ID、权限
- 审计：REORDER_ORGANIZATION / REORDER_USER / REORDER_ROLE 落库与明细
"""

from __future__ import annotations

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


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


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


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


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


# ===== 机构 =====


def test_reorder_organizations_within_parent(b3_app, b3_client, login_token):
    root_id = _org_id(b3_app, "ORG_ROOT")
    ops_id = _org_id(b3_app, "ORG_OPS")
    comms_id = _org_id(b3_app, "ORG_COMMS")
    response = b3_client.put(
        "/api/v1/organizations/reorder",
        json={
            "parentId": str(root_id),
            "orderedIds": [str(comms_id), str(ops_id)],
        },
        headers=_auth(login_token),
    )
    assert response.status_code == 200
    # 槽位置换：原次序 OPS(20)/COMMS(30)，置换后 COMMS 落槽位1(=10)、
    # OPS 落槽位2(=20，与原值相同)，实际发生变化的仅 COMMS 一行。
    assert response.get_json()["data"] == {"reorderedCount": 1}
    with b3_app.app_context():
        ops = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_OPS")
        )
        comms = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_COMMS")
        )
        assert comms.sort_no == 10
        assert ops.sort_no == 20


def test_reorder_organizations_rejects_cross_scope(b3_app, b3_client, login_token):
    ops_id = _org_id(b3_app, "ORG_OPS")
    comms_id = _org_id(b3_app, "ORG_COMMS")
    response = b3_client.put(
        "/api/v1/organizations/reorder",
        json={"parentId": None, "orderedIds": [str(ops_id), str(comms_id)]},
        headers=_auth(login_token),
    )
    assert response.status_code == 400


def test_reorder_organizations_rejects_missing_parent(b3_client, login_token):
    response = b3_client.put(
        "/api/v1/organizations/reorder",
        json={"parentId": "999999", "orderedIds": ["1", "2"]},
        headers=_auth(login_token),
    )
    assert response.status_code == 404


def test_reorder_organizations_requires_admin(b3_client, token_for):
    response = b3_client.put(
        "/api/v1/organizations/reorder",
        json={"parentId": None, "orderedIds": ["1", "2"]},
        headers=_auth(token_for("user")),
    )
    assert response.status_code == 403


# ===== 用户 =====


def test_reorder_users_slot_swap_and_idempotent(b3_app, b3_client, login_token):
    admin_id = _user_id(b3_app, "admin")
    user_id = _user_id(b3_app, "user")
    response = b3_client.put(
        "/api/v1/users/reorder",
        json={"orderedIds": [str(user_id), str(admin_id)]},
        headers=_auth(login_token),
    )
    assert response.status_code == 200
    # 种子用户 sort_no 全空：[admin, user, disabled] 置换为 [user, admin, disabled]
    assert response.get_json()["data"] == {"reorderedCount": 3}
    listing = b3_client.get("/api/v1/users", headers=_auth(login_token))
    assert listing.status_code == 200
    usernames = [
        item["username"] for item in listing.get_json()["data"]["items"]
    ]
    assert usernames[:3] == ["user", "admin", "disabled"]
    # 相同顺序重复提交：无变更
    repeat = b3_client.put(
        "/api/v1/users/reorder",
        json={"orderedIds": [str(user_id), str(admin_id)]},
        headers=_auth(login_token),
    )
    assert repeat.status_code == 200
    assert repeat.get_json()["data"] == {"reorderedCount": 0}


def test_reorder_users_rejects_unknown_id(b3_client, login_token):
    response = b3_client.put(
        "/api/v1/users/reorder",
        json={"orderedIds": ["999999", "888888"]},
        headers=_auth(login_token),
    )
    assert response.status_code == 404


def test_reorder_users_requires_admin(b3_client, token_for):
    response = b3_client.put(
        "/api/v1/users/reorder",
        json={"orderedIds": ["999999", "888888"]},
        headers=_auth(token_for("user")),
    )
    assert response.status_code == 403


# ===== 角色 =====


def test_reorder_roles_slot_swap(b2_app, b2_client, login_token, role_cleanup):
    first = _create_role(b2_client, login_token, "资料员", sort_no=10)
    second = _create_role(b2_client, login_token, "档案员", sort_no=20)
    assert first.status_code == 201 and second.status_code == 201
    first_id = first.get_json()["data"]["id"]
    second_id = second.get_json()["data"]["id"]
    response = b2_client.put(
        "/api/v1/roles/reorder",
        json={"orderedIds": [second_id, first_id]},
        headers=_auth(login_token),
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {"reorderedCount": 2}
    with b2_app.app_context():
        rows = {
            row.role_name: row.sort_no
            for row in db.session.scalars(db.select(Role)).all()
        }
        assert rows == {"资料员": 20, "档案员": 10}


def test_reorder_roles_rejects_unknown_id(
    b2_client, login_token, role_cleanup
):
    created = _create_role(b2_client, login_token, "资料员")
    assert created.status_code == 201
    response = b2_client.put(
        "/api/v1/roles/reorder",
        json={"orderedIds": ["999999", "888888"]},
        headers=_auth(login_token),
    )
    assert response.status_code == 400


# ===== 审计 =====


def test_reorder_writes_three_audit_actions(
    b3_app, b3_client, login_token, role_cleanup
):
    root_id = _org_id(b3_app, "ORG_ROOT")
    ops_id = _org_id(b3_app, "ORG_OPS")
    comms_id = _org_id(b3_app, "ORG_COMMS")
    org_response = b3_client.put(
        "/api/v1/organizations/reorder",
        json={
            "parentId": str(root_id),
            "orderedIds": [str(comms_id), str(ops_id)],
        },
        headers=_auth(login_token),
    )
    assert org_response.status_code == 200

    admin_id = _user_id(b3_app, "admin")
    user_id = _user_id(b3_app, "user")
    user_response = b3_client.put(
        "/api/v1/users/reorder",
        json={"orderedIds": [str(user_id), str(admin_id)]},
        headers=_auth(login_token),
    )
    assert user_response.status_code == 200

    first_role = _create_role(b3_client, login_token, "资料员")
    second_role = _create_role(b3_client, login_token, "档案员")
    assert first_role.status_code == 201 and second_role.status_code == 201
    role_response = b3_client.put(
        "/api/v1/roles/reorder",
        json={
            "orderedIds": [
                second_role.get_json()["data"]["id"],
                first_role.get_json()["data"]["id"],
            ]
        },
        headers=_auth(login_token),
    )
    assert role_response.status_code == 200

    with b3_app.app_context():
        rows = db.session.scalars(
            db.select(AuditLog).where(
                AuditLog.action_type.in_(
                    [
                        "REORDER_ORGANIZATION",
                        "REORDER_USER",
                        "REORDER_ROLE",
                    ]
                )
            )
        ).all()
        by_action = {row.action_type: row for row in rows}
        assert set(by_action) == {
            "REORDER_ORGANIZATION",
            "REORDER_USER",
            "REORDER_ROLE",
        }
        assert by_action["REORDER_ORGANIZATION"].detail["reorderedCount"] == 1
        assert by_action["REORDER_USER"].detail["reorderedCount"] == 3
        assert by_action["REORDER_ROLE"].detail["reorderedCount"] == 2