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

from __future__ import annotations

import io
import json
from pathlib import Path

import fitz
import pytest
from docx import Document as WordDocument


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


def _category_id(app, code: str) -> str:
    from dms.extensions import db
    from dms.models import Category

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


def _docx_bytes(term: str = "Q3正文唯一词") -> bytes:
    stream = io.BytesIO()
    document = WordDocument()
    document.add_paragraph(f"段落 {term}")
    table = document.add_table(rows=1, cols=1)
    table.cell(0, 0).text = "表格唯一词"
    section = document.sections[0]
    section.header.paragraphs[0].text = "页眉唯一词"
    section.footer.paragraphs[0].text = "页脚唯一词"
    document.save(stream)
    return stream.getvalue()


def _pdf_bytes(term: str = "Q3PdfUniqueTerm") -> bytes:
    document = fitz.open()
    page = document.new_page()
    page.insert_text((72, 72), term)
    content = document.tobytes()
    document.close()
    return content


def _main_metadata(category_id: str, name: str = "Q3主案") -> dict:
    return {
        "documentName": name,
        "documentType": "MAIN",
        "summary": "Q3摘要",
        "categoryId": category_id,
        "securityLevel": "INTERNAL",
        "tags": ["Q3"],
    }


def _sub_metadata(parent_id: str, name: str = "Q3子方案") -> dict:
    return {
        "documentName": name,
        "documentType": "SUB_PLAN",
        "summary": "Q3子方案摘要",
        "parentDocumentId": parent_id,
        "securityLevel": "INTERNAL",
        "tags": ["Q3"],
    }


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


def _tree_node(nodes: list[dict], category_id: str) -> dict:
    pending = list(nodes)
    while pending:
        node = pending.pop()
        if node["id"] == category_id:
            return node
        pending.extend(node["children"])
    raise AssertionError(f"category {category_id} not found")


def test_business_timezone_default_and_invalid_fail_fast(monkeypatch):
    from dms.config import load_dms_config

    monkeypatch.delenv("DMS_BUSINESS_TIMEZONE", raising=False)
    assert load_dms_config()["DMS_BUSINESS_TIMEZONE"] == "Asia/Shanghai"
    monkeypatch.setenv("DMS_BUSINESS_TIMEZONE", "Not/A_Real_Zone")
    with pytest.raises(ValueError, match="DMS_BUSINESS_TIMEZONE"):
        load_dms_config()


def test_admin_modules_and_historical_auditor_are_invalid(
    b2_app, b2_client, b2_password, token_for
):
    from dms.extensions import db
    from dms.models import User
    from dms.security.jwt_tokens import issue_access_token

    admin = b2_client.get(
        "/api/v1/users/me", headers=_auth(token_for("admin"))
    )
    assert admin.get_json()["data"]["allowedModules"] == [
        "DOCUMENT_BROWSER",
        "BACKEND_MANAGEMENT",
        "AUDIT_LOG",
    ]
    failed = b2_client.post(
        "/api/v1/auth/login",
        json={
            "username": "auditor",
            "password": b2_password,
            "keepSignedIn": False,
        },
    )
    assert failed.status_code == 401
    with b2_app.app_context():
        historical = db.session.scalar(
            db.select(User).where(User.username == "auditor")
        )
        assert historical.role_code == "USER"
        assert historical.status == "DISABLED"
        assert historical.is_deleted
        old_token, _ = issue_access_token(historical, keep_signed_in=False)
    assert b2_client.get(
        "/api/v1/users/me", headers=_auth(old_token)
    ).status_code == 401


def test_leaf_rules_create_defaults_and_sub_plan_category_sync(
    b3_app, b3_client, token_for
):
    from dms.extensions import db
    from dms.models import Document

    token = token_for("admin")
    root_id = _category_id(b3_app, "ROOT_A")
    leaf_one = _category_id(b3_app, "CHILD_A1")
    leaf_two = _category_id(b3_app, "CHILD_A2")
    rejected = _upload(
        b3_client, token, _main_metadata(root_id), _docx_bytes(), "docx"
    )
    assert rejected.status_code == 409
    assert rejected.get_json()["code"] == "CATEGORY_NOT_LEAF"
    assert rejected.get_json()["details"] == {
        "categoryId": root_id,
        "childCategoryCount": 2,
    }

    created_response = _upload(
        b3_client,
        token,
        _main_metadata(leaf_one, "Q3叶子主案"),
        _docx_bytes(),
        "docx",
    )
    assert created_response.status_code == 201
    main = created_response.get_json()["data"]
    assert (main["status"], main["visibilityType"]) == (
        "PUBLISHED",
        "CUSTOM",
    )
    child_category = b3_client.post(
        "/api/v1/categories",
        json={
            "categoryCode": "Q3_CHILD_FORBIDDEN",
            "categoryName": "禁止子分类",
            "categoryType": "OTHER",
            "parentId": leaf_one,
            "sortNo": 99,
        },
        headers=_auth(token),
    )
    assert child_category.status_code == 409
    assert child_category.get_json()["code"] == "CATEGORY_NOT_LEAF"

    sub_response = _upload(
        b3_client,
        token,
        _sub_metadata(main["id"]),
        _docx_bytes("子方案正文"),
        "docx",
    )
    assert sub_response.status_code == 201
    sub = sub_response.get_json()["data"]
    assert sub["categoryId"] == leaf_one
    with b3_app.app_context():
        current = db.session.get(Document, int(main["id"]))
        current_version = current.row_version
    moved = b3_client.put(
        f"/api/v1/documents/{main['id']}",
        json={
            "documentName": main["documentName"],
            "summary": main["summary"],
            "categoryId": leaf_two,
            "securityLevel": main["securityLevel"],
            "tags": main["tags"],
            "rowVersion": current_version,
        },
        headers=_auth(token),
    )
    assert moved.status_code == 200
    with b3_app.app_context():
        persisted_sub = db.session.get(Document, int(sub["id"]))
        assert str(persisted_sub.category_id) == leaf_two
        assert persisted_sub.category_name == moved.get_json()["data"]["categoryName"]


def test_include_descendants_and_permission_aware_subtree_count(
    b3_app, b3_client, token_for
):
    from dms.extensions import db
    from dms.models import AuditLog, User

    admin_token = token_for("admin")
    user_token = token_for("user")
    root_id = _category_id(b3_app, "ROOT_A")
    leaf_id = _category_id(b3_app, "CHILD_A1")
    created = _upload(
        b3_client,
        admin_token,
        _main_metadata(leaf_id, "Q3递归主案"),
        _docx_bytes(),
        "docx",
    ).get_json()["data"]

    exact = b3_client.get(
        f"/api/v1/documents?categoryId={root_id}", headers=_auth(admin_token)
    ).get_json()["data"]
    recursive = b3_client.get(
        f"/api/v1/documents?categoryId={root_id}&includeDescendants=true",
        headers=_auth(admin_token),
    ).get_json()["data"]
    all_categories = b3_client.get(
        "/api/v1/documents", headers=_auth(admin_token)
    ).get_json()["data"]
    assert exact["total"] == 0
    assert recursive["total"] == 1
    assert all_categories["total"] == 1
    invalid = b3_client.get(
        f"/api/v1/documents?categoryId={root_id}&includeDescendants=yes",
        headers=_auth(admin_token),
    )
    assert invalid.status_code == 400

    admin_tree = b3_client.get(
        "/api/v1/categories/tree", headers=_auth(admin_token)
    ).get_json()["data"]
    user_tree = b3_client.get(
        "/api/v1/categories/tree", headers=_auth(user_token)
    ).get_json()["data"]
    assert _tree_node(admin_tree, root_id)["documentCount"] == 1
    assert _tree_node(user_tree, root_id)["documentCount"] == 0
    with b3_app.app_context():
        user = db.session.scalar(db.select(User).where(User.username == "user"))
        audit_before = db.session.scalar(
            db.select(db.func.count()).select_from(AuditLog)
        )
    saved = b3_client.put(
        f"/api/v1/documents/{created['id']}/permissions",
        json={
            "visibilityType": "CUSTOM",
            "documentRowVersion": created["rowVersion"],
            "entries": [
                {
                    "subjectType": "USER",
                    "subjectId": str(user.id),
                    "canView": True,
                    "canDownload": True,
                    "canEdit": False,
                    "canManagePermission": False,
                    "canDelete": False,
                }
            ],
        },
        headers=_auth(admin_token),
    )
    assert saved.status_code == 200
    keyword_tree = b3_client.get(
        "/api/v1/categories/tree?keyword=CHILD_A1",
        headers=_auth(user_token),
    ).get_json()["data"]
    assert _tree_node(keyword_tree, root_id)["documentCount"] == 1
    with b3_app.app_context():
        audit_after = db.session.scalar(
            db.select(db.func.count()).select_from(AuditLog)
        )
    assert audit_after == audit_before + 1  # 仅权限配置产生审计，树查询不产生。


def test_docx_pdf_content_search_and_original_preview(
    b3_app, b3_client, token_for
):
    from dms.extensions import db
    from dms.models import AuditLog, Document

    token = token_for("admin")
    leaf_id = _category_id(b3_app, "CHILD_A1")
    docx_content = _docx_bytes("只在DOCX正文中的检索词")
    docx_item = _upload(
        b3_client,
        token,
        _main_metadata(leaf_id, "Q3-DOCX"),
        docx_content,
        "docx",
    ).get_json()["data"]
    pdf_content = _pdf_bytes("OnlyInsidePdfBody")
    pdf_item = _upload(
        b3_client,
        token,
        _main_metadata(leaf_id, "Q3-PDF"),
        pdf_content,
        "pdf",
    ).get_json()["data"]

    for keyword, expected_id in (
        ("只在DOCX正文中的检索词", docx_item["id"]),
        ("表格唯一词", docx_item["id"]),
        ("页眉唯一词", docx_item["id"]),
        ("页脚唯一词", docx_item["id"]),
        ("OnlyInsidePdfBody", pdf_item["id"]),
    ):
        result = b3_client.get(
            f"/api/v1/documents?keyword={keyword}", headers=_auth(token)
        ).get_json()["data"]
        assert [item["id"] for item in result["items"]] == [expected_id]

    with b3_app.app_context():
        before_download = {
            item.id: item.download_count
            for item in db.session.scalars(
                db.select(Document).where(
                    Document.id.in_([int(docx_item["id"]), int(pdf_item["id"])])
                )
            )
        }
        before_audit = db.session.scalar(
            db.select(db.func.count(AuditLog.id)).where(
                AuditLog.action_type == "DOWNLOAD_DOCUMENT"
            )
        )
    for item, content, mime in (
        (
            docx_item,
            docx_content,
            "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        ),
        (pdf_item, pdf_content, "application/pdf"),
    ):
        response = b3_client.get(
            f"/api/v1/documents/{item['id']}/preview",
            headers={**_auth(token), "Origin": "http://127.0.0.1:9346"},
        )
        assert response.status_code == 200
        assert response.mimetype == mime
        assert response.data == content
        assert response.headers["Content-Disposition"].startswith("inline")
        assert "filename*=" in response.headers["Content-Disposition"]
        assert response.headers["X-Content-Type-Options"] == "nosniff"
        assert response.headers["Cache-Control"] == "private, no-store"
        assert "Content-Disposition" in response.headers[
            "Access-Control-Expose-Headers"
        ]
        response.close()
    with b3_app.app_context():
        after_download = {
            item.id: item.download_count
            for item in db.session.scalars(
                db.select(Document).where(
                    Document.id.in_([int(docx_item["id"]), int(pdf_item["id"])])
                )
            )
        }
        after_audit = db.session.scalar(
            db.select(db.func.count(AuditLog.id)).where(
                AuditLog.action_type == "DOWNLOAD_DOCUMENT"
            )
        )
    assert after_download == before_download
    assert after_audit == before_audit


def test_attachment_library_is_admin_only_and_user_cannot_guess_unmounted_id(
    b3_app, b3_client, token_for
):
    admin_token = token_for("admin")
    user_token = token_for("user")
    response = b3_client.post(
        "/api/v1/attachments",
        data={
            "file": (io.BytesIO(_docx_bytes()), "Q3附件.docx"),
            "metadata": json.dumps(
                {
                    "documentName": "Q3未挂载附件",
                    "attachmentType": "OTHER",
                    "summary": "未挂载",
                    "tags": ["Q3"],
                },
                ensure_ascii=False,
            ),
        },
        headers=_auth(admin_token),
        content_type="multipart/form-data",
    )
    attachment = response.get_json()["data"]
    assert b3_client.get(
        "/api/v1/attachments", headers=_auth(user_token)
    ).status_code == 403
    assert b3_client.get(
        f"/api/v1/attachments/{attachment['id']}", headers=_auth(user_token)
    ).status_code == 403
    assert b3_client.get(
        f"/api/v1/documents/{attachment['id']}/preview",
        headers=_auth(user_token),
    ).status_code == 403


def test_backfill_gate_requires_explicit_dms_test_url(monkeypatch, b2_app):
    from dms.backfill_document_content import _database_gate

    with b2_app.app_context():
        monkeypatch.delenv("DMS_DATABASE_URL", raising=False)
        with pytest.raises(RuntimeError, match="DMS_DATABASE_URL"):
            _database_gate()
        monkeypatch.setenv(
            "DMS_DATABASE_URL",
            "mysql+pymysql://hidden:hidden@localhost:3306/dms",
        )
        with pytest.raises(RuntimeError, match="dms_test"):
            _database_gate()


def test_q3_openapi_contract_has_no_new_preview_path():
    import yaml

    root = Path(__file__).resolve().parents[2] / "openapi"
    document = yaml.safe_load((root / "openapi.yaml").read_text("utf-8"))
    schemas = yaml.safe_load(
        (root / "components" / "schemas.yaml").read_text("utf-8")
    )
    paths = yaml.safe_load(
        (root / "paths" / "documents.yaml").read_text("utf-8")
    )
    assert document["info"]["version"] == "1.6.0"
    assert schemas["RoleCode"]["enum"] == ["USER", "ADMIN"]
    parameters = paths["Documents"]["get"]["parameters"]
    assert any(item.get("name") == "includeDescendants" for item in parameters)
    preview_content = paths["DocumentPreview"]["get"]["responses"]["200"][
        "content"
    ]
    assert set(preview_content) == {"application/pdf"}
    assert all("content-preview" not in path for path in document["paths"])