script-build-and-framework-tmp/visualization-old/backend/tests/test_retrieval_detail_projection.py

from app.retrieval_detail_projection import (
    project_retrieval_event,
    summarize_retrieval_output,
)


def _event(event_id, name, input_value, output_value, **overrides):
    event = {
        "id": event_id,
        "event_type": "tool_call",
        "event_name": name,
        "status": "ok",
        "ended_at": "2026-07-14T20:00:00",
        "input": {"contentType": "json", "content": input_value},
        "output": {"contentType": "json", "content": output_value},
    }
    event.update(overrides)
    return event


def test_case_hit_preserves_conditions_and_complete_case_content():
    detail = project_retrieval_event(
        _event(
            4607,
            "search_script_decode_case",
            {"keyword": "毛驴效应", "top_k": 3, "return_field": "主脉络"},
            {
                "success": True,
                "count": 2,
                "results": [
                    {
                        "post_id": "case-a",
                        "account": "每天心理学",
                        "score": 0.7032,
                        "return_field": "主脉络",
                        "data": [{"id": "段落1", "详细描述": "完整解构内容"}],
                    },
                    {
                        "post_id": "case-b",
                        "account": "每天心理学",
                        "score": 0.68,
                        "return_field": "主脉络",
                        "data": [{"id": "段落1", "详细描述": "第二条"}],
                    },
                ],
            },
        )
    )

    assert detail["detailKind"] == "retrieval-event"
    assert detail["sourceKind"] == "agent-query"
    assert detail["outcome"] == {
        "state": "hit",
        "count": 2,
        "message": "查询成功，返回 2 条结果。",
    }
    assert [item["title"] for item in detail["items"]] == [
        "每天心理学 · Case 1",
        "每天心理学 · Case 2",
    ]
    assert detail["items"][0]["sections"][0]["value"][0]["详细描述"] == "完整解构内容"
    assert [field["label"] for field in detail["queryConditions"]] == ["关键词", "最多返回", "返回内容"]


def test_successful_empty_is_not_a_failure_and_keeps_query_conditions():
    detail = project_retrieval_event(
        _event(
            4603,
            "search_script_decode_case",
            {"keyword": "毛驴效应", "account_name": "每天一点心理学", "top_k": 3},
            {"success": True, "count": 0, "results": []},
        )
    )

    assert detail["outcome"] == {
        "state": "empty",
        "count": 0,
        "message": "查询执行成功，没有匹配数据。",
    }
    assert detail["items"] == []
    assert {item["label"] for item in detail["queryConditions"]} == {"关键词", "账号", "最多返回"}


def test_text_failure_exposes_readable_parameter_error():
    detail = project_retrieval_event(
        _event(
            4066,
            "search_script_decode_case",
            {"return_field": "写法"},
            "错误: return_field 仅支持 ['主脉络', '主题']",
            status="error",
        )
    )

    assert detail["outcome"]["state"] == "failure"
    assert detail["outcome"]["errorKind"] == "parameter"
    assert "return_field 仅支持" in detail["outcome"]["message"]


def test_knowledge_list_count_and_content_are_projected():
    detail = project_retrieval_event(
        _event(
            4574,
            "search_knowledge",
            {"keyword": "决策瘫痪", "max_count": 3},
            [
                {"id": 1, "title": "知识 A", "purpose": "用于核对", "content": "正文 A"},
                {"id": 2, "title": "知识 B", "purpose": "用于行动建议", "content": "正文 B"},
            ],
        )
    )

    assert detail["outcome"]["count"] == 2
    assert detail["outcome"]["state"] == "hit"
    assert detail["items"][1]["sections"] == [{"title": "完整内容", "value": "正文 B"}]


def test_direct_tool_has_direct_source_and_structured_output():
    detail = project_retrieval_event(
        _event(
            4650,
            "get_script_snapshot",
            {"script_build_id": 444},
            {"script_build": {"id": 444}, "paragraphs": [{"id": 2284}], "elements": []},
        )
    )

    assert detail["sourceKind"] == "direct-tool"
    assert detail["outcome"]["state"] == "hit"
    assert [section["title"] for section in detail["items"][0]["sections"]] == ["脚本概况", "段落"]


def test_terminal_run_marks_unfinished_event_interrupted_before_other_rules():
    detail = project_retrieval_event(
        _event(
            5000,
            "search_knowledge",
            {"keyword": "决策"},
            [{"title": "还未完成的结果"}],
            status="running",
            ended_at=None,
        ),
        run_status="failed",
    )

    assert detail["outcome"]["state"] == "interrupted"
    assert detail["items"] == []


def test_light_summary_does_not_include_full_results():
    summary = summarize_retrieval_output(
        "search_script_decode_case",
        raw_status="ok",
        ended_at="2026-07-14T20:00:00",
        output={"success": True, "count": 1, "results": [{"data": "x" * 1000}]},
    )

    assert summary == {"state": "hit", "count": 1, "error": None}
    assert "results" not in summary