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

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

覆盖：
- 上传默认取创建者当前部门；显式 organizationId 可指定其他启用部门
- 无效/停用部门 → 404 RESOURCE_NOT_FOUND
- PUT /documents/{id} 修改归属部门
- 组织改名级联 Document.organization_name 冗余字段
- 用户调岗不联动历史文档归属
- 附件库条目输出 organizationName
"""

from __future__ import annotations

import io
import json

import pytest
from docx import Document as WordDocument

from dms.extensions import db
from dms.models import Document, Organization, 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 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",
            visibility_type="CUSTOM",
            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