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

"""B6借阅授权审批单（PDF凭证）上传/下载/清除的接口契约测试（契约1.19 12.3-12.5）。"""

from __future__ import annotations

import io

VOUCHER_PDF = b"%PDF-1.4\nB6 voucher original\n%%EOF"
REPLACED_PDF = b"%PDF-1.4\nB6 voucher replaced\n%%EOF"


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


def _doc_id(app, name: str) -> str:
    from dms.extensions import db
    from dms.models import Document

    with app.app_context():
        document_id = db.session.scalar(
            db.select(Document.id).where(Document.document_name == name)
        )
        assert document_id is not None
        return str(document_id)


def _entry_id(b6_client, login_token, document_id) -> str:
    response = b6_client.get(
        f"/api/v1/documents/{document_id}/permissions",
        headers=_headers(login_token),
    )
    assert response.status_code == 200
    entries = response.get_json()["data"]["entries"]
    assert entries
    return entries[0]["id"]


def _upload(b6_client, login_token, document_id, entry_id, content, name):
    return b6_client.post(
        f"/api/v1/documents/{document_id}/permissions/entries/{entry_id}"
        "/voucher",
        data={"file": (io.BytesIO(content), name)},
        headers=_headers(login_token),
        content_type="multipart/form-data",
    )


def test_voucher_upload_download_replace_delete_and_audit_flow(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import AuditLog

    main_id = _doc_id(b6_app, "B4_TEST_自定义主案")
    entry_id = _entry_id(b6_client, login_token, main_id)

    # 未上传凭证：下载404，详情凭证字段为空
    missing = b6_client.get(
        f"/api/v1/documents/{main_id}/permissions/entries/{entry_id}/voucher",
        headers=_headers(login_token),
    )
    assert missing.status_code == 404
    assert missing.get_json()["code"] == "RESOURCE_NOT_FOUND"

    uploaded = _upload(
        b6_client, login_token, main_id, entry_id, VOUCHER_PDF, "借阅审批单.pdf"
    )
    assert uploaded.status_code == 200, uploaded.get_json()
    data = uploaded.get_json()["data"]
    assert data["documentId"] == main_id
    assert data["entryId"] == entry_id
    assert data["voucherFileName"] == "借阅审批单.pdf"
    assert data["voucherFileSize"] == len(VOUCHER_PDF)
    assert data["voucherUploadedAt"]

    detail = b6_client.get(
        f"/api/v1/documents/{main_id}/permissions",
        headers=_headers(login_token),
    ).get_json()["data"]["entries"]
    assert [item["id"] for item in detail] == [entry_id]
    assert detail[0]["voucherFileName"] == "借阅审批单.pdf"
    assert detail[0]["voucherFileSize"] == len(VOUCHER_PDF)
    assert detail[0]["voucherUploadedAt"] == data["voucherUploadedAt"]

    downloaded = b6_client.get(
        f"/api/v1/documents/{main_id}/permissions/entries/{entry_id}/voucher",
        headers=_headers(login_token),
    )
    assert downloaded.status_code == 200
    assert downloaded.data == VOUCHER_PDF
    assert downloaded.mimetype == "application/pdf"
    assert "attachment" in downloaded.headers["Content-Disposition"]

    # 重复上传为替换语义：凭证字段覆盖，下载返回新内容
    replaced = _upload(
        b6_client, login_token, main_id, entry_id, REPLACED_PDF, "替换审批单.pdf"
    )
    assert replaced.status_code == 200, replaced.get_json()
    assert replaced.get_json()["data"]["voucherFileName"] == "替换审批单.pdf"
    again = b6_client.get(
        f"/api/v1/documents/{main_id}/permissions/entries/{entry_id}/voucher",
        headers=_headers(login_token),
    )
    assert again.data == REPLACED_PDF

    removed = b6_client.delete(
        f"/api/v1/documents/{main_id}/permissions/entries/{entry_id}/voucher",
        headers=_headers(login_token),
    )
    assert removed.status_code == 200
    assert removed.get_json()["data"] == {
        "documentId": main_id,
        "entryId": entry_id,
    }
    cleared = b6_client.get(
        f"/api/v1/documents/{main_id}/permissions/entries/{entry_id}/voucher",
        headers=_headers(login_token),
    )
    assert cleared.status_code == 404
    entries = b6_client.get(
        f"/api/v1/documents/{main_id}/permissions",
        headers=_headers(login_token),
    ).get_json()["data"]["entries"]
    assert entries[0]["voucherFileName"] is None
    assert entries[0]["voucherFileSize"] is None
    assert entries[0]["voucherUploadedAt"] is None

    with b6_app.app_context():
        actions = db.session.scalars(
            db.select(AuditLog.action_type)
            .where(AuditLog.target_id == int(main_id))
            .order_by(AuditLog.id.asc())
        ).all()
    assert "UPLOAD_PERMISSION_VOUCHER" in actions
    assert "DELETE_PERMISSION_VOUCHER" in actions


def test_voucher_upload_rejects_non_pdf(b6_app, b6_client, login_token):
    main_id = _doc_id(b6_app, "B4_TEST_自定义主案")
    entry_id = _entry_id(b6_client, login_token, main_id)

    wrong_type = _upload(
        b6_client, login_token, main_id, entry_id, b"plain text", "凭证.txt"
    )
    assert wrong_type.status_code == 415
    assert wrong_type.get_json()["code"] == "UNSUPPORTED_FILE_TYPE"

    no_extension = _upload(
        b6_client, login_token, main_id, entry_id, VOUCHER_PDF, "凭证"
    )
    assert no_extension.status_code == 415
    assert no_extension.get_json()["code"] == "UNSUPPORTED_FILE_TYPE"


def test_voucher_requires_config_permission(
    b6_app, b6_client, login_token, token_for
):
    main_id = _doc_id(b6_app, "B4_TEST_自定义主案")
    entry_id = _entry_id(b6_client, login_token, main_id)
    user_headers = _headers(token_for("user"))
    base = f"/api/v1/documents/{main_id}/permissions/entries/{entry_id}/voucher"
    assert (
        b6_client.post(
            base,
            data={"file": (io.BytesIO(VOUCHER_PDF), "审批单.pdf")},
            headers=user_headers,
            content_type="multipart/form-data",
        ).status_code
        == 403
    )
    assert b6_client.get(base, headers=user_headers).status_code == 403
    assert b6_client.delete(base, headers=user_headers).status_code == 403


def test_voucher_unknown_entry_and_inherited_sources(
    b6_app, b6_client, login_token
):
    main_id = _doc_id(b6_app, "B4_TEST_自定义主案")
    sub_id = _doc_id(b6_app, "B4_TEST_动态继承子案")
    attachment_id = _doc_id(b6_app, "B4_TEST_规范附件")

    unknown = b6_client.post(
        f"/api/v1/documents/{main_id}/permissions/entries/99999999/voucher",
        data={"file": (io.BytesIO(VOUCHER_PDF), "审批单.pdf")},
        headers=_headers(login_token),
        content_type="multipart/form-data",
    )
    assert unknown.status_code == 404
    assert unknown.get_json()["code"] == "RESOURCE_NOT_FOUND"

    sub = b6_client.get(
        f"/api/v1/documents/{sub_id}/permissions/entries/1/voucher",
        headers=_headers(login_token),
    )
    assert sub.status_code == 409
    assert sub.get_json()["code"] == "SUB_PLAN_PERMISSION_INHERITED"

    attachment = b6_client.get(
        f"/api/v1/documents/{attachment_id}/permissions/entries/1/voucher",
        headers=_headers(login_token),
    )
    assert attachment.status_code == 400
    assert attachment.get_json()["code"] == "ATTACHMENT_HAS_NO_ACL"