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

"""文档归属部门行为测试（T1.2）。

覆盖：
- 上传默认取创建者当前部门；显式 organizationId 可指定其他启用部门
- 无效/停用部门 → 404 RESOURCE_NOT_FOUND
- PUT /documents/{id} 修改归属部门
- 编辑缺省 organizationId 沿用现有归属，不被编辑人部门覆盖
- 非管理员原样回传其他部门归属视为未变更；改挂其他部门 → 400 INVALID_ARGUMENT
- 组织改名级联 Document.organization_name 冗余字段
- 用户调岗不联动历史文档归属
- 附件库条目输出 organizationName
"""

from __future__ import annotations

import io
import json

import pytest
from docx import Document as WordDocument
from sqlalchemy import update

from dms.extensions import db
from dms.models import Document, Organization, Role, RolePermission, User


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


def _docx_bytes() -> bytes:
    stream = io.BytesIO()
    document = WordDocument()
    document.add_paragraph("DO组织行为验证")
    document.save(stream)
    return stream.getvalue()


def _upload(client, token: str, path: str, metadata: dict):
    name = metadata["documentName"]
    return client.post(
        f"/api/v1/{path}",
        data={
            "file": (io.BytesIO(_docx_bytes()), f"{name}.docx"),
            "metadata": json.dumps(metadata, ensure_ascii=False),
        },
        headers=_auth(token),
        content_type="multipart/form-data",
    )


def _leaf_category_id(app) -> str:
    from dms.models import Category

    with app.app_context():
        category = db.session.scalar(
            db.select(Category).where(Category.category_code == "CHILD_A1")
        )
        assert category is not None
        return str(category.id)


def _org_by_code(app, code: str):
    with app.app_context():
        org = db.session.scalar(
            db.select(Organization).where(Organization.org_code == code)
        )
        assert org is not None
        return org.id, org.org_name, org.parent_id, org.row_version


def _main_metadata(leaf_id: str, name: str, **extra) -> dict:
    return {
        "documentName": name,
        "documentType": "MAIN",
        "summary": "DO摘要",
        "categoryId": leaf_id,
        "securityLevel": "INTERNAL",
        "tags": ["DO"],
        **extra,
    }


@pytest.fixture()
def do_app(b3_app):
    with b3_app.app_context():
        root = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_ROOT")
        )
        assert root is not None
        org_a = Organization(
            org_code="ORG_DO_A",
            org_name="DO甲部门",
            org_path=f"{root.org_path}/DO甲部门",
            parent_id=root.id,
            sort_no=80,
            status="ENABLED",
        )
        org_disabled = Organization(
            org_code="ORG_DO_DISABLED",
            org_name="DO停用部门",
            org_path=f"{root.org_path}/DO停用部门",
            parent_id=root.id,
            sort_no=81,
            status="DISABLED",
        )
        db.session.add_all([org_a, org_disabled])
        db.session.commit()
        b3_app.config["DO_IDS"] = {
            "org_a": org_a.id,
            "org_disabled": org_disabled.id,
        }
    return b3_app


@pytest.fixture()
def do_client(do_app):
    return do_app.test_client()


def test_upload_defaults_to_creator_organization(do_app, do_client, token_for):
    response = _upload(
        do_client,
        token_for("admin"),
        "documents",
        _main_metadata(_leaf_category_id(do_app), "DO_TEST_默认部门主案"),
    )
    assert response.status_code == 201
    item = response.get_json()["data"]
    with do_app.app_context():
        admin = db.session.scalar(db.select(User).where(User.username == "admin"))
        assert item["organizationId"] == str(admin.organization_id)
        assert item["organizationName"] == admin.organization_name


def test_upload_with_explicit_enabled_organization(do_app, do_client, token_for):
    org_a_id = do_app.config["DO_IDS"]["org_a"]
    response = _upload(
        do_client,
        token_for("admin"),
        "documents",
        _main_metadata(
            _leaf_category_id(do_app),
            "DO_TEST_指定部门主案",
            organizationId=str(org_a_id),
        ),
    )
    assert response.status_code == 201
    item = response.get_json()["data"]
    assert item["organizationId"] == str(org_a_id)
    assert item["organizationName"] == "DO甲部门"


@pytest.mark.parametrize(
    "org_code",
    ["missing", "ORG_DO_DISABLED"],
)
def test_upload_rejects_invalid_or_disabled_organization(
    do_app, do_client, token_for, org_code
):
    raw_id = (
        "99999999"
        if org_code == "missing"
        else str(do_app.config["DO_IDS"][org_code])
    )
    response = _upload(
        do_client,
        token_for("admin"),
        "documents",
        _main_metadata(
            _leaf_category_id(do_app),
            f"DO_TEST_非法部门主案_{org_code}",
            organizationId=raw_id,
        ),
    )
    assert response.status_code == 404
    assert response.get_json()["code"] == "RESOURCE_NOT_FOUND"


def test_update_document_organization(do_app, do_client, token_for):
    token = token_for("admin")
    org_a_id = do_app.config["DO_IDS"]["org_a"]
    created = _upload(
        do_client,
        token,
        "documents",
        _main_metadata(_leaf_category_id(do_app), "DO_TEST_修改部门主案"),
    ).get_json()["data"]
    assert created["organizationName"] != "DO甲部门"

    updated = do_client.put(
        f"/api/v1/documents/{created['id']}",
        json={
            "documentName": created["documentName"],
            "summary": created["summary"],
            "categoryId": created["categoryId"],
            "securityLevel": created["securityLevel"],
            "tags": created["tags"],
            "rowVersion": created["rowVersion"],
            "organizationId": str(org_a_id),
        },
        headers=_auth(token),
    )
    assert updated.status_code == 200
    assert updated.get_json()["data"]["organizationName"] == "DO甲部门"

    listed = do_client.get(
        "/api/v1/documents",
        query_string={
            "keyword": "DO_TEST_修改部门主案",
            "organizationKeyword": "DO甲部",
        },
        headers=_auth(token),
    ).get_json()["data"]
    assert [item["documentName"] for item in listed["items"]] == [
        "DO_TEST_修改部门主案"
    ]


def _edit_payload(item: dict, **extra) -> dict:
    return {
        "documentName": item["documentName"],
        "summary": item["summary"],
        "categoryId": item["categoryId"],
        "securityLevel": item["securityLevel"],
        "tags": item["tags"],
        "rowVersion": item["rowVersion"],
        **extra,
    }


def test_update_without_organization_keeps_current(
    do_app, do_client, token_for
):
    """编辑未携带 organizationId 时沿用文档现有归属，不被编辑人部门覆盖。"""
    token = token_for("admin")
    org_a_id = do_app.config["DO_IDS"]["org_a"]
    created = _upload(
        do_client,
        token,
        "documents",
        _main_metadata(
            _leaf_category_id(do_app),
            "DO_TEST_编辑不改部门主案",
            organizationId=str(org_a_id),
        ),
    ).get_json()["data"]

    updated = do_client.put(
        f"/api/v1/documents/{created['id']}",
        json=_edit_payload(created),
        headers=_auth(token),
    )
    assert updated.status_code == 200
    data = updated.get_json()["data"]
    assert data["organizationId"] == str(org_a_id)
    assert data["organizationName"] == "DO甲部门"


def test_non_admin_can_resubmit_other_organization_but_not_reassign(
    do_app, do_client, token_for
):
    """非管理员原样回传其他部门归属视为未变更；改挂其他部门被拒绝。"""
    admin_token = token_for("admin")
    user_token = token_for("user")
    org_a_id = do_app.config["DO_IDS"]["org_a"]
    comms_id, _, _, _ = _org_by_code(do_app, "ORG_COMMS")

    created = _upload(
        do_client,
        admin_token,
        "documents",
        _main_metadata(
            _leaf_category_id(do_app),
            "DO_TEST_普通用户编辑他部门主案",
            organizationId=str(org_a_id),
        ),
    ).get_json()["data"]
    with do_app.app_context():
        user_id = db.session.scalar(
            db.select(User.id).where(User.username == "user")
        )

    # 借阅语义下EDIT仅来自角色权限矩阵：为用户挂载INTERNAL×OTHER
    # 可查看可编辑的角色（该主案归属DO甲部门，对用户属于OTHER范围）。
    with do_app.app_context():
        role = Role(role_name="DO编辑角色", sort_no=90)
        db.session.add(role)
        db.session.flush()
        db.session.add(
            RolePermission(
                role_id=role.id,
                security_level="INTERNAL",
                dept_scope="OTHER",
                can_view=True,
                can_edit=True,
            )
        )
        db.session.execute(
            update(User).where(User.id == user_id).values(role_id=role.id)
        )
        db.session.commit()

    # 保存权限会推进rowVersion，重新读取详情
    latest = do_client.get(
        f"/api/v1/documents/{created['id']}", headers=_auth(user_token)
    ).get_json()["data"]
    assert "EDIT" in latest["allowedActions"]

    # 原样回传其他部门（DO甲部门）归属：视为未变更，允许保存
    unchanged = do_client.put(
        f"/api/v1/documents/{created['id']}",
        json=_edit_payload(latest, organizationId=str(org_a_id)),
        headers=_auth(user_token),
    )
    assert unchanged.status_code == 200, unchanged.get_json()
    assert unchanged.get_json()["data"]["organizationId"] == str(org_a_id)

    # 改挂到通信部：非管理员被拒绝
    reassigned = do_client.put(
        f"/api/v1/documents/{created['id']}",
        json=_edit_payload(
            unchanged.get_json()["data"], organizationId=str(comms_id)
        ),
        headers=_auth(user_token),
    )
    assert reassigned.status_code == 400
    body = reassigned.get_json()
    assert body["code"] == "INVALID_ARGUMENT"
    assert body["message"] == "仅管理员可将文档调整到其他部门"



def test_organization_rename_cascades_document_name(do_app, do_client, token_for):
    token = token_for("admin")
    org_a_id, _, parent_id, row_version = _org_by_code(do_app, "ORG_DO_A")
    created = _upload(
        do_client,
        token,
        "documents",
        _main_metadata(
            _leaf_category_id(do_app),
            "DO_TEST_改名级联主案",
            organizationId=str(org_a_id),
        ),
    ).get_json()["data"]
    assert created["organizationName"] == "DO甲部门"

    def rename(name: str, version: int) -> int:
        response = do_client.put(
            f"/api/v1/organizations/{org_a_id}",
            json={
                "orgName": name,
                "parentId": str(parent_id),
                "sortNo": 80,
                "status": "ENABLED",
                "rowVersion": version,
            },
            headers=_auth(token),
        )
        assert response.status_code == 200, response.get_json()
        return response.get_json()["data"]["rowVersion"]

    try:
        row_version = rename("DO甲部门新名", row_version)
        listed = do_client.get(
            "/api/v1/documents",
            query_string={"keyword": "DO_TEST_改名级联主案"},
            headers=_auth(token),
        ).get_json()["data"]
        assert listed["items"][0]["organizationName"] == "DO甲部门新名"
    finally:
        rename("DO甲部门", row_version)


def test_user_transfer_does_not_cascade_documents(do_app, do_client, token_for):
    token = token_for("admin")
    org_a_id = do_app.config["DO_IDS"]["org_a"]
    comms_id, comms_name, _, _ = _org_by_code(do_app, "ORG_COMMS")

    created_user = do_client.post(
        "/api/v1/users",
        json={
            "username": "do_transfer_user",
            "realName": "调岗验证用户",
            "initialPassword": "Initial@2026",
            "organizationId": str(org_a_id),
            "roleCode": "USER",
            "securityLevel": "INTERNAL",
            "status": "ENABLED",
        },
        headers=_auth(token),
    )
    assert created_user.status_code == 201, created_user.get_json()
    user_id = created_user.get_json()["data"]["id"]

    with do_app.app_context():
        leaf = db.session.scalar(
            db.select(Document).where(
                Document.document_name == "DO_TEST_修改部门主案",
                Document.is_deleted.is_(False),
            )
        )
        assert leaf is not None
        document = Document(
            document_name="DO_TEST_调岗不联动主案",
            summary="调岗不联动",
            document_type="MAIN",
            document_status="PUBLISHED",
            security_level="INTERNAL",
            organization_id=org_a_id,
            organization_name="DO甲部门",
            tags=["DO"],
            original_file_name="DO_TEST_调岗不联动主案.docx",
            file_relative_path="original/DO_TEST_调岗不联动主案.docx",
            file_extension="docx",
            mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
            file_size=64,
            file_hash="d" * 64,
            search_text="DO_TEST_调岗不联动主案",
            created_by=int(user_id),
            updated_by=int(user_id),
            created_by_name="调岗验证用户",
            updated_by_name="调岗验证用户",
            category_id=leaf.category_id,
        )
        db.session.add(document)
        user = db.session.get(User, int(user_id))
        row_version = user.row_version
        db.session.commit()

    transferred = do_client.put(
        f"/api/v1/users/{user_id}",
        json={
            "realName": "调岗验证用户",
            "organizationId": str(comms_id),
            "roleCode": "USER",
            "securityLevel": "INTERNAL",
            "status": "ENABLED",
            "rowVersion": row_version,
        },
        headers=_auth(token),
    )
    assert transferred.status_code == 200, transferred.get_json()
    assert transferred.get_json()["data"]["organizationName"] == comms_name

    with do_app.app_context():
        current = db.session.scalar(
            db.select(Document).where(
                Document.document_name == "DO_TEST_调岗不联动主案",
                Document.is_deleted.is_(False),
            )
        )
        assert current.organization_id == org_a_id
        assert current.organization_name == "DO甲部门"


def test_attachment_library_exposes_organization_name(
    do_app, do_client, token_for
):
    token = token_for("admin")
    created = _upload(
        do_client,
        token,
        "attachments",
        {
            "documentName": "DO_TEST_部门附件",
            "attachmentType": "OTHER",
            "summary": "DO附件",
            "tags": ["DO"],
        },
    )
    assert created.status_code == 201
    listed = do_client.get(
        "/api/v1/attachments",
        query_string={"keyword": "DO_TEST_部门附件"},
        headers=_auth(token),
    ).get_json()["data"]
    assert len(listed["items"]) == 1
    with do_app.app_context():
        admin = db.session.scalar(db.select(User).where(User.username == "admin"))
        assert listed["items"][0]["organizationId"] == str(admin.organization_id)
        assert listed["items"][0]["organizationName"] == admin.organization_name