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

"""跨页移动排序：文档 move 与挂载附件 move 的行为契约测试。"""

from __future__ import annotations

import pytest

from dms.extensions import db
from dms.models import AuditLog, AttachmentBinding, Category, Document, User


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


def _ids(app):
    return app.config["B4_IDS"]


def _category(app, code: str) -> Category:
    with app.app_context():
        category = db.session.scalar(
            db.select(Category).where(Category.category_code == code)
        )
        assert category is not None
        db.session.expunge(category)
        return category


def _main_document(name: str, category: Category) -> Document:
    return Document(
        document_name=name,
        document_type="MAIN",
        document_status="PUBLISHED",
        security_level="INTERNAL",
        category_id=category.id,
        category_name=category.category_name,
        category_path=category.category_path,
        original_file_name=f"{name}.docx",
        file_relative_path=f"original/MOVE_{name}.docx",
        file_extension="docx",
        file_hash="b" * 64,
    )


def _attachment_document(name: str, attachment_type: str) -> Document:
    return Document(
        document_name=name,
        document_type="ATTACHMENT",
        document_status="PUBLISHED",
        security_level="PUBLIC",
        attachment_type=attachment_type,
        original_file_name=f"{name}.pdf",
        file_relative_path=f"original/MOVE_{name}.pdf",
        file_extension="pdf",
        file_hash="c" * 64,
    )


def _move(client, document_id, payload, token):
    return client.put(
        f"/api/v1/documents/{document_id}/move",
        json=payload,
        headers=_auth(token),
    )


def _main_order(app) -> list[int]:
    """全量主案按 (sort_no, id) 升序的 id 序列。"""
    with app.app_context():
        rows = db.session.scalars(
            db.select(Document).where(
                Document.document_type == "MAIN",
                Document.is_deleted.is_(False),
            )
        ).all()
        return sorted(rows, key=lambda item: (item.sort_no, item.id))


def test_move_requires_authentication(b6_client):
    response = b6_client.put("/api/v1/documents/1/move", json={"target": "TOP"})
    assert response.status_code == 401


@pytest.mark.parametrize(
    "payload",
    [
        {},
        {"targetPosition": 1, "target": "TOP"},
        {"targetPosition": 0},
        {"target": "MIDDLE"},
        {"targetPosition": "3"},
        {"targetPosition": 1, "unknown": 1},
    ],
)
def test_move_rejects_invalid_payload(b6_client, login_token, payload):
    response = _move(b6_client, "1", payload, login_token)
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


def test_move_missing_document_returns_404(b6_client, login_token):
    response = _move(b6_client, "999999", {"target": "TOP"}, login_token)
    assert response.status_code == 404
    assert response.get_json()["code"] == "RESOURCE_NOT_FOUND"


def test_move_main_plan_to_top_rewrites_sort_numbers(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    # 种子主案 sort_no 全 0：顺序按 id 即 全部/组织/自定义/草稿/绝密/无授权
    assert [item.id for item in _main_order(b6_app)].index(ids["main_custom"]) == 2

    response = _move(
        b6_client, ids["main_custom"], {"target": "TOP"}, login_token
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {"position": 1, "total": 6}

    order = [item.id for item in _main_order(b6_app)]
    assert order.index(ids["main_custom"]) == 0
    # 其余主案相对顺序保持，且全量按步长10重编号
    assert [item.sort_no for item in _main_order(b6_app)] == [
        10, 20, 30, 40, 50, 60,
    ]
    with b6_app.app_context():
        audit = db.session.scalar(
            db.select(AuditLog).where(AuditLog.action_type == "MOVE_DOCUMENT")
        )
        assert audit is not None
        assert audit.target_id == ids["main_custom"]
        detail = audit.operation_detail
        assert detail["fromPosition"] == 3
        assert detail["toPosition"] == 1
        assert detail["scope"] == "GLOBAL"
        assert detail["scopeTotal"] == 6


def test_move_clamps_out_of_range_position_to_bottom(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    response = _move(
        b6_client, ids["main_all"], {"targetPosition": 999}, login_token
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {"position": 6, "total": 6}
    assert [item.id for item in _main_order(b6_app)].index(ids["main_all"]) == 5


def test_move_to_current_position_is_noop_without_audit(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    response = _move(
        b6_client, ids["main_custom"], {"targetPosition": 3}, login_token
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {"position": 3, "total": 6}
    with b6_app.app_context():
        assert (
            db.session.scalar(
                db.select(AuditLog).where(AuditLog.action_type == "MOVE_DOCUMENT")
            )
            is None
        )
        assert [item.sort_no for item in _main_order(b6_app)] == [0] * 6


def test_move_with_category_scope_keeps_other_category_relative_order(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    root_a = _category(b6_app, "ROOT_A")
    root_b = _category(b6_app, "ROOT_B")
    with b6_app.app_context():
        admin = db.session.scalar(db.select(User).where(User.username == "admin"))
        extra_b1 = _main_document("MOVE_TEST_范围B甲", root_b)
        extra_b2 = _main_document("MOVE_TEST_范围B乙", root_b)
        db.session.add_all([extra_b1, extra_b2])
        db.session.flush()
        binding_before = [
            item.id for item in _main_order(b6_app)
            if item.id in {extra_b1.id, extra_b2.id}
        ]
        assert binding_before == [extra_b1.id, extra_b2.id]

    response = _move(
        b6_client,
        ids["main_custom"],
        {"target": "TOP", "scopeCategoryId": str(root_a.id)},
        login_token,
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {"position": 1, "total": 6}

    order = [item.id for item in _main_order(b6_app)]
    assert order.index(ids["main_custom"]) == 0
    # 范围外（ROOT_B）主案相对顺序保持
    assert order.index(extra_b1.id) < order.index(extra_b2.id)
    with b6_app.app_context():
        audit = db.session.scalar(
            db.select(AuditLog).where(AuditLog.action_type == "MOVE_DOCUMENT")
        )
        assert audit.operation_detail["scopeCategoryId"] == str(root_a.id)


def test_move_rejects_document_outside_category_scope(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    root_b = _category(b6_app, "ROOT_B")
    response = _move(
        b6_client,
        ids["main_custom"],
        {"target": "TOP", "scopeCategoryId": str(root_b.id)},
        login_token,
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "DOCUMENT_NOT_IN_MOVE_SCOPE"


def test_move_rejects_scope_category_on_sub_plan(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    root_a = _category(b6_app, "ROOT_A")
    response = _move(
        b6_client,
        ids["sub"],
        {"target": "TOP", "scopeCategoryId": str(root_a.id)},
        login_token,
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


def test_move_sub_plan_stays_within_parent_scope(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    root_a = _category(b6_app, "ROOT_A")
    with b6_app.app_context():
        parent = db.session.get(Document, ids["main_custom"])
        second_sub = Document(
            document_name="MOVE_TEST_第二子案",
            document_type="SUB_PLAN",
            document_status="PUBLISHED",
            security_level="SECRET",
            category_id=root_a.id,
            category_name=root_a.category_name,
            category_path=root_a.category_path,
            parent_document_id=parent.id,
            root_document_id=parent.id,
            original_file_name="MOVE_TEST_第二子案.docx",
            file_relative_path="original/MOVE_TEST_第二子案.docx",
            file_extension="docx",
            file_hash="d" * 64,
        )
        db.session.add(second_sub)
        db.session.commit()
        second_sub_id = second_sub.id

    response = _move(b6_client, second_sub_id, {"target": "TOP"}, login_token)
    assert response.status_code == 200
    assert response.get_json()["data"] == {"position": 1, "total": 2}

    with b6_app.app_context():
        rows = db.session.scalars(
            db.select(Document).where(
                Document.document_type == "SUB_PLAN",
                Document.parent_document_id == ids["main_custom"],
                Document.is_deleted.is_(False),
            )
        ).all()
        by_sort = sorted(rows, key=lambda item: (item.sort_no, item.id))
        assert [item.id for item in by_sort] == [second_sub_id, ids["sub"]]
        # 子案移动不影响主案排序
        assert [item.sort_no for item in _main_order(b6_app)] == [0] * 6
        audit = db.session.scalar(
            db.select(AuditLog).where(AuditLog.action_type == "MOVE_DOCUMENT")
        )
        assert audit.operation_detail["scope"] == str(ids["main_custom"])


def _binding_id(app, main_id: int, attachment_id: int) -> int:
    with app.app_context():
        binding = db.session.scalar(
            db.select(AttachmentBinding).where(
                AttachmentBinding.main_document_id == main_id,
                AttachmentBinding.attachment_document_id == attachment_id,
                AttachmentBinding.is_deleted.is_(False),
            )
        )
        assert binding is not None
        return binding.id


def _binding_moves(app, main_id: int) -> list[tuple[int, int]]:
    """主案下有效挂载按 (sort_no, id) 升序的 (binding_id, sort_no)。"""
    with app.app_context():
        rows = db.session.scalars(
            db.select(AttachmentBinding).where(
                AttachmentBinding.main_document_id == main_id,
                AttachmentBinding.is_deleted.is_(False),
            )
        ).all()
        return [
            (item.id, item.sort_no)
            for item in sorted(rows, key=lambda row: (row.sort_no, row.id))
        ]


def test_move_attachment_binding_to_top_bumps_row_version(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    # main_all 挂载：表格附件(10) 在前、规范附件(20) 在后
    binding_one = _binding_id(b6_app, ids["main_all"], ids["attachment_one"])
    response = b6_client.put(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{binding_one}/move",
        json={"target": "TOP", "mainPlanRowVersion": 0},
        headers=_auth(login_token),
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {"position": 1, "total": 2, "mainPlanRowVersion": 1}

    moves = {binding: sort for binding, sort in _binding_moves(b6_app, ids["main_all"])}
    binding_two = _binding_id(b6_app, ids["main_all"], ids["attachment_two"])
    assert moves[binding_one] < moves[binding_two]
    with b6_app.app_context():
        main = db.session.get(Document, ids["main_all"])
        assert main.row_version == 1
        audit = db.session.scalar(
            db.select(AuditLog).where(AuditLog.action_type == "MOVE_ATTACHMENT")
        )
        assert audit is not None
        assert audit.target_id == binding_one
        detail = audit.operation_detail
        assert detail["fromPosition"] == 2
        assert detail["toPosition"] == 1
        assert detail["scopeTotal"] == 2


def test_move_attachment_within_type_scope_keeps_other_type_order(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    with b6_app.app_context():
        admin = db.session.scalar(db.select(User).where(User.username == "admin"))
        main = db.session.get(Document, ids["main_all"])
        third = _attachment_document("MOVE_TEST_训考乙", "TRAINING_MATERIAL")
        db.session.add(third)
        db.session.flush()
        db.session.add(
            AttachmentBinding(
                main_document_id=main.id,
                attachment_document_id=third.id,
                sort_no=30,
                created_by=admin.id,
                updated_by=admin.id,
            )
        )
        main.attachment_count = 3
        db.session.commit()
        third_id = third.id

    binding_one = _binding_id(b6_app, ids["main_all"], ids["attachment_one"])
    # 训考资料范围内：规范附件(20) 第1、训考乙(30) 第2，移动训考乙到第1
    response = b6_client.put(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/"
        f"{_binding_id(b6_app, ids['main_all'], third_id)}/move",
        json={
            "targetPosition": 1,
            "scopeAttachmentType": "TRAINING_MATERIAL",
            "mainPlanRowVersion": 0,
        },
        headers=_auth(login_token),
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {"position": 1, "total": 2, "mainPlanRowVersion": 1}

    moves = _binding_moves(b6_app, ids["main_all"])
    binding_two = _binding_id(b6_app, ids["main_all"], ids["attachment_two"])
    binding_third = _binding_id(b6_app, ids["main_all"], third_id)
    # 支撑成果(类型外)保持最前，训考乙越过规范附件
    assert [binding for binding, _ in moves] == [
        binding_two, binding_third, binding_one,
    ]


def test_move_attachment_rejects_type_scope_mismatch(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    # 规范附件是训考资料，按支撑成果口径移动应拒绝
    binding_one = _binding_id(b6_app, ids["main_all"], ids["attachment_one"])
    response = b6_client.put(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{binding_one}/move",
        json={
            "target": "TOP",
            "scopeAttachmentType": "SUPPORT_ACHIEVEMENT",
            "mainPlanRowVersion": 0,
        },
        headers=_auth(login_token),
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "DOCUMENT_NOT_IN_MOVE_SCOPE"


def test_move_attachment_rejects_stale_row_version(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    binding_one = _binding_id(b6_app, ids["main_all"], ids["attachment_one"])
    response = b6_client.put(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{binding_one}/move",
        json={"target": "TOP", "mainPlanRowVersion": 9},
        headers=_auth(login_token),
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "DATA_VERSION_CONFLICT"


@pytest.mark.parametrize(
    "payload",
    [
        {"target": "TOP"},
        {"mainPlanRowVersion": 0},
        {"targetPosition": 1, "target": "TOP", "mainPlanRowVersion": 0},
        {"target": "SIDE", "mainPlanRowVersion": 0},
        {"targetPosition": 1, "mainPlanRowVersion": -1},
        {"targetPosition": 1, "mainPlanRowVersion": 0, "extra": True},
    ],
)
def test_move_attachment_rejects_invalid_payload(
    b6_app, b6_client, login_token, payload
):
    ids = _ids(b6_app)
    binding_one = _binding_id(b6_app, ids["main_all"], ids["attachment_one"])
    response = b6_client.put(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{binding_one}/move",
        json=payload,
        headers=_auth(login_token),
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


def test_move_attachment_missing_binding_returns_404(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    response = b6_client.put(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/999999/move",
        json={"target": "TOP", "mainPlanRowVersion": 0},
        headers=_auth(login_token),
    )
    assert response.status_code == 404
    assert response.get_json()["code"] == "RESOURCE_NOT_FOUND"