"""契约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