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

"""全局搜索与高级检索测试。

覆盖检索条件矩阵（四条件 AND 叠加）：
- 简单关键词：仅匹配标题；主案聚合模式下命中子案/附件标题也召回主案
- 正文关键词：匹配 contentText / searchText，命中时输出 contentSnippet
- 密级多选：勾选项内 OR
- 部门关键词：模糊匹配 organization_name
- 聚合命中徽章：titleHit / subHitCount / attachmentHitCount（不可见子案不计入）
"""

from __future__ import annotations

import pytest


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


def data(response):
    return response.get_json()["data"]


def names(response) -> list[str]:
    return [item["documentName"] for item in data(response)["items"]]


def by_name(response, name: str):
    return next(item for item in data(response)["items"] if item["documentName"] == name)


@pytest.fixture()
def gs_app(b4_app):
    from dms.extensions import db
    from dms.models import AttachmentBinding, Document, Organization

    with b4_app.app_context():
        root = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_ROOT")
        )
        assert root is not None
        # 追加三个 GS 专用组织，便于用 organizationKeyword=GS 圈定数据集。
        gs_ops = Organization(
            org_code="ORG_GS_OPS",
            org_name="GS作战组",
            org_path=f"{root.org_path}/GS作战组",
            parent_id=root.id,
            sort_no=50,
            status="ENABLED",
        )
        gs_comms = Organization(
            org_code="ORG_GS_COMMS",
            org_name="GS通信组",
            org_path=f"{root.org_path}/GS通信组",
            parent_id=root.id,
            sort_no=60,
            status="ENABLED",
        )
        gs_base = Organization(
            org_code="ORG_GS_BASE",
            org_name="GS检索部",
            org_path=f"{root.org_path}/GS检索部",
            parent_id=root.id,
            sort_no=70,
            status="ENABLED",
        )
        db.session.add_all([gs_ops, gs_comms, gs_base])
        db.session.flush()

        def document(
            name: str,
            document_type: str,
            *,
            security: str = "INTERNAL",
            visibility: str = "ALL_AUTHENTICATED",
            parent: Document | None = None,
            organization: Organization,
            content_text: str | None = None,
            attachment_type: str | None = None,
        ) -> Document:
            item = Document(
                document_name=name,
                summary=f"{name}摘要",
                document_type=document_type,
                document_status="PUBLISHED",
                security_level=security,
                visibility_type=visibility,
                attachment_type=attachment_type,
                organization_id=organization.id,
                organization_name=organization.org_name,
                parent_document_id=parent.id if parent else None,
                root_document_id=parent.id if parent else None,
                tags=["GS"],
                original_file_name=f"{name}.docx",
                file_relative_path=f"original/{name}.docx",
                file_extension="docx",
                mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
                file_size=64,
                file_hash="b" * 64,
                search_text=name,
                content_text=content_text,
                created_by_name="系统管理员",
                updated_by_name="系统管理员",
            )
            db.session.add(item)
            db.session.flush()
            return item

        alpha = document(
            "GS_TEST_阿尔法主案",
            "MAIN",
            security="INTERNAL",
            organization=gs_ops,
            content_text="阿尔法主案正文，包含全域搜索唯一词。",
        )
        alpha_sub = document(
            "GS_TEST_阿尔法雷达子案",
            "SUB_PLAN",
            security="INTERNAL",
            parent=alpha,
            organization=gs_ops,
            content_text="阿尔法子案正文专属词。",
        )
        document(
            "GS_TEST_阿尔法隐藏子案",
            "SUB_PLAN",
            security="SECRET",
            visibility="CUSTOM",
            parent=alpha,
            organization=gs_ops,
        )
        alpha_att = document(
            "GS_TEST_阿尔法挂载附件",
            "ATTACHMENT",
            security="PUBLIC",
            organization=gs_ops,
            content_text="附件正文检索词。",
            attachment_type="OTHER",
        )
        beta = document(
            "GS_TEST_贝塔主案",
            "MAIN",
            security="SECRET",
            organization=gs_comms,
        )
        document(
            "GS_TEST_贝塔雷达子案",
            "SUB_PLAN",
            security="SECRET",
            parent=beta,
            organization=gs_comms,
            content_text="雷达正文唯一词。",
        )
        gamma = document(
            "GS_TEST_伽马主案",
            "MAIN",
            security="TOP_SECRET",
            organization=gs_base,
        )
        document(
            "GS_TEST_伽马雷达子案",
            "SUB_PLAN",
            security="TOP_SECRET",
            parent=gamma,
            organization=gs_base,
        )
        db.session.add(
            AttachmentBinding(
                main_document_id=alpha.id,
                attachment_document_id=alpha_att.id,
                sort_no=10,
            )
        )
        alpha.attachment_count = 1
        alpha.child_count = 2
        beta.child_count = 1
        gamma.child_count = 1
        db.session.commit()
        b4_app.config["GS_IDS"] = {
            "alpha": alpha.id,
            "alpha_sub": alpha_sub.id,
            "alpha_att": alpha_att.id,
            "beta": beta.id,
            "gamma": gamma.id,
            "gs_ops": gs_ops.id,
        }
    return b4_app


@pytest.fixture()
def gs_client(gs_app):
    return gs_app.test_client()


def test_aggregate_main_title_hit(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?aggregateHits=true&keyword=阿尔法主案",
        headers=auth(token_for("user")),
    )
    assert response.status_code == 200
    payload = data(response)
    assert [item["documentName"] for item in payload["items"]] == [
        "GS_TEST_阿尔法主案"
    ]
    item = payload["items"][0]
    assert item["titleHit"] is True
    assert item["subHitCount"] == 0
    assert item["attachmentHitCount"] == 0


def test_aggregate_sub_hit_with_acl_filtered_count(gs_client, token_for):
    user_response = gs_client.get(
        "/api/v1/documents?aggregateHits=true&keyword=雷达",
        headers=auth(token_for("user")),
    )
    user_names = names(user_response)
    # 伽马主案 TOP_SECRET 超出 user 密级，不召回。
    assert user_names == ["GS_TEST_阿尔法主案", "GS_TEST_贝塔主案"]
    alpha_user = by_name(user_response, "GS_TEST_阿尔法主案")
    # 隐藏子案 CUSTOM 无授权，对 user 不可见，不计入计数。
    assert alpha_user["subHitCount"] == 1
    assert alpha_user["titleHit"] is False

    admin_response = gs_client.get(
        "/api/v1/documents?aggregateHits=true&keyword=雷达",
        headers=auth(token_for("admin")),
    )
    assert names(admin_response) == [
        "GS_TEST_阿尔法主案",
        "GS_TEST_贝塔主案",
        "GS_TEST_伽马主案",
    ]
    alpha_admin = by_name(admin_response, "GS_TEST_阿尔法主案")
    assert alpha_admin["subHitCount"] == 2


def test_aggregate_keyword_hits_only_one_sub_plan_title(gs_client, token_for):
    """回归：子案标题命中只召回所属主案，其余主案不得入列表。"""
    response = gs_client.get(
        "/api/v1/documents?aggregateHits=true&keyword=贝塔雷达",
        headers=auth(token_for("user")),
    )
    assert names(response) == ["GS_TEST_贝塔主案"]
    beta = data(response)["items"][0]
    assert beta["titleHit"] is False
    assert beta["subHitCount"] == 1
    assert beta["attachmentHitCount"] == 0


def test_aggregate_attachment_hit(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?aggregateHits=true&keyword=挂载附件",
        headers=auth(token_for("user")),
    )
    assert names(response) == ["GS_TEST_阿尔法主案"]
    item = data(response)["items"][0]
    assert item["titleHit"] is False
    assert item["subHitCount"] == 0
    assert item["attachmentHitCount"] == 1


def test_aggregate_mixed_hit(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?aggregateHits=true&keyword=阿尔法",
        headers=auth(token_for("user")),
    )
    assert names(response) == ["GS_TEST_阿尔法主案"]
    item = data(response)["items"][0]
    assert item["titleHit"] is True
    assert item["subHitCount"] == 1
    assert item["attachmentHitCount"] == 1


def test_aggregate_no_hit_returns_empty(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?aggregateHits=true&keyword=GS_TEST_不存在的词",
        headers=auth(token_for("admin")),
    )
    assert response.status_code == 200
    assert data(response)["total"] == 0
    assert data(response)["items"] == []


def test_aggregate_without_keyword_keeps_all(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?aggregateHits=true&organizationKeyword=GS",
        headers=auth(token_for("user")),
    )
    assert set(names(response)) == {
        "GS_TEST_阿尔法主案",
        "GS_TEST_贝塔主案",
    }
    item = by_name(response, "GS_TEST_贝塔主案")
    assert item["titleHit"] is True
    assert item["subHitCount"] == 0
    assert item["attachmentHitCount"] == 0


def test_advanced_content_keyword_filters_main_itself(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?aggregateHits=true"
        "&organizationKeyword=GS&contentKeyword=全域搜索唯一词",
        headers=auth(token_for("user")),
    )
    assert names(response) == ["GS_TEST_阿尔法主案"]
    # 高级正文命中时输出片段。
    item = data(response)["items"][0]
    assert "全域搜索唯一词" in item["contentSnippet"]


def test_advanced_content_keyword_excludes_unmatched(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?aggregateHits=true"
        "&organizationKeyword=GS&contentKeyword=雷达正文唯一词",
        headers=auth(token_for("user")),
    )
    # 正文词只存在于贝塔子案：高级条件只过滤主案本身，子案命中不召回主案。
    assert data(response)["total"] == 0


def test_advanced_security_levels_multiselect(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?organizationKeyword=GS"
        "&securityLevels=SECRET,TOP_SECRET",
        headers=auth(token_for("admin")),
    )
    assert set(names(response)) == {"GS_TEST_贝塔主案", "GS_TEST_伽马主案"}
    internal = gs_client.get(
        "/api/v1/documents?organizationKeyword=GS&securityLevels=INTERNAL",
        headers=auth(token_for("admin")),
    )
    assert names(internal) == ["GS_TEST_阿尔法主案"]


def test_advanced_organization_keyword(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?organizationKeyword=GS作战",
        headers=auth(token_for("user")),
    )
    assert names(response) == ["GS_TEST_阿尔法主案"]


def test_simple_and_advanced_are_anded(gs_client, token_for):
    mismatch = gs_client.get(
        "/api/v1/documents?aggregateHits=true"
        "&keyword=贝塔&organizationKeyword=GS作战",
        headers=auth(token_for("user")),
    )
    assert data(mismatch)["total"] == 0
    match = gs_client.get(
        "/api/v1/documents?aggregateHits=true"
        "&keyword=贝塔&securityLevels=SECRET&organizationKeyword=GS通信",
        headers=auth(token_for("user")),
    )
    assert names(match) == ["GS_TEST_贝塔主案"]


def test_keyword_matches_title_only_not_content(gs_client, token_for):
    response = gs_client.get(
        "/api/v1/documents?keyword=全域搜索唯一词",
        headers=auth(token_for("admin")),
    )
    assert data(response)["total"] == 0


def test_sub_plans_tab_applies_all_conditions(
    gs_app, gs_client, token_for
):
    alpha_id = gs_app.config["GS_IDS"]["alpha"]
    base = f"/api/v1/main-plans/{alpha_id}/sub-plans"

    user_all = gs_client.get(
        f"{base}?pageSize=100", headers=auth(token_for("user"))
    )
    assert names(user_all) == ["GS_TEST_阿尔法雷达子案"]

    admin_all = gs_client.get(
        f"{base}?pageSize=100", headers=auth(token_for("admin"))
    )
    assert set(names(admin_all)) == {
        "GS_TEST_阿尔法雷达子案",
        "GS_TEST_阿尔法隐藏子案",
    }

    by_keyword = gs_client.get(f"{base}?keyword=雷达", headers=auth(token_for("user")))
    assert names(by_keyword) == ["GS_TEST_阿尔法雷达子案"]

    by_content = gs_client.get(
        f"{base}?contentKeyword=专属词", headers=auth(token_for("user"))
    )
    assert names(by_content) == ["GS_TEST_阿尔法雷达子案"]
    item = data(by_content)["items"][0]
    assert "专属词" in item["contentSnippet"]

    by_levels = gs_client.get(
        f"{base}?securityLevels=SECRET", headers=auth(token_for("user"))
    )
    # 唯一 SECRET 子案对 user 不可见（CUSTOM 无授权）→ 完全空白。
    assert data(by_levels)["total"] == 0

    by_org = gs_client.get(
        f"{base}?organizationKeyword=GS通信", headers=auth(token_for("user"))
    )
    assert data(by_org)["total"] == 0


def test_mounted_attachments_tab_applies_all_conditions(
    gs_app, gs_client, token_for
):
    alpha_id = gs_app.config["GS_IDS"]["alpha"]
    base = f"/api/v1/main-plans/{alpha_id}/attachments"

    by_keyword = gs_client.get(f"{base}?keyword=挂载附件", headers=auth(token_for("user")))
    assert names(by_keyword) == ["GS_TEST_阿尔法挂载附件"]

    by_content = gs_client.get(
        f"{base}?contentKeyword=附件正文检索词", headers=auth(token_for("user"))
    )
    assert names(by_content) == ["GS_TEST_阿尔法挂载附件"]
    item = data(by_content)["items"][0]
    assert "附件正文检索词" in item["contentSnippet"]
    assert item["organizationName"] == "GS作战组"

    by_levels = gs_client.get(
        f"{base}?securityLevels=SECRET", headers=auth(token_for("user"))
    )
    assert data(by_levels)["total"] == 0

    by_org = gs_client.get(
        f"{base}?organizationKeyword=GS通信", headers=auth(token_for("user"))
    )
    assert data(by_org)["total"] == 0


@pytest.mark.parametrize(
    "query",
    [
        "securityLevels=BOGUS",
        "securityLevels=SECRET,BOGUS",
        "securityLevels=SECRET,SECRET",
        "aggregateHits=true&documentType=MAIN,SUB_PLAN",
        "aggregateHits=maybe",
    ],
)
def test_invalid_advanced_params(gs_client, token_for, query):
    response = gs_client.get(
        f"/api/v1/documents?{query}", headers=auth(token_for("admin"))
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"