Create-knowledge-find-decode/tests/test_topic_table_query_generation.py

from __future__ import annotations

from typing import Any

import pytest
from fastapi.testclient import TestClient

from app.api import app
from app.routes import query_generation
from query_planning.sources.pattern_topic_build import (
    PatternTopicBuildConfig,
    PatternTopicBuildSource,
)
from query_planning.topic_table import (
    ROUTES,
    TopicTableGenerationError,
    generate_topic_table_preview,
    preview_to_dict,
)


def _topic_payload() -> dict[str, Any]:
    return {
        "success": True,
        "source_url": "https://pattern.aiddit.com/api/pattern/topic_builds/1229",
        "build": {
            "id": 1229,
            "demand": "为小羊糊涂啊创作新选题",
            "demand_constraints": None,
            "strategies_config": {"always_on": []},
            "status": "success",
        },
        "topics": [
            {
                "id": 1392,
                "status": "mature",
                "result": "樱花树下低角度拍少女与小羊，传递治愈感并分享穿搭。",
                "topic_direction": {
                    "account_name": "小羊糊涂啊",
                    "persona_dimensions": {
                        "实质偏好": "樱花与穿搭",
                        "形式偏好": "低角度和动态摆拍",
                        "意图偏好": "治愈和种草",
                    },
                },
                "points": [
                    {
                        "id": 8371,
                        "point_type": "灵感点",
                        "point_result": "低角度仰拍小羊和彩虹光晕",
                        "is_active": True,
                        "item_ids": [25684, 25688],
                    }
                ],
                "composition_items": [
                    {
                        "id": 25684,
                        "item_level": "element",
                        "dimension": "形式",
                        "point_type": "灵感点",
                        "element_name": "低角度仰拍",
                        "category_path": "/呈现/视觉/低角度",
                        "reason": "增强视觉张力",
                        "is_active": True,
                        "sources": [
                            {
                                "derivation_type": "external_search",
                                "source_type": "search_case",
                                "source_reference_id": "case-1",
                                "reason": "草坪案例证明低角度有动感",
                                "dataset_from": "",
                                "source_reference_data": {"large": "must not enter prompt"},
                            }
                        ],
                    },
                    {
                        "id": 25688,
                        "item_level": "element",
                        "dimension": "实质",
                        "point_type": "灵感点",
                        "element_name": "羊",
                        "reason": "与账号名形成双关",
                        "is_active": True,
                        "sources": [],
                    },
                    {
                        "id": 25700,
                        "item_level": "element",
                        "dimension": "意图",
                        "point_type": "目的点",
                        "element_name": "治愈感",
                        "reason": "满足粉丝对美好生活的向往",
                        "is_active": True,
                        "sources": [],
                    },
                ],
                "item_relations": [
                    {
                        "source_item_id": 25684,
                        "target_item_id": 25688,
                        "reason": "低角度用于突出人与小羊互动",
                    }
                ],
            }
        ],
    }


def _llm_output() -> dict[str, Any]:
    needs = []
    for index, (route, config) in enumerate(ROUTES.items()):
        needs.append(
            {
                "need_key": f"need_{route}",
                "route": route,
                "decision_context": f"决定{config['label']}怎么做",
                "unknown_information": f"缺少{config['label']}的可执行方法",
                "priority": config["priority"],
                "source_ref": {"point_id": 8371, "item_ids": [25684, 25688]},
                "queries": [
                    f"{config['label']}有哪些可执行方法",
                    "逆光人像怎么拍出彩虹光晕" if index == 0 else f"{config['label']}常见失败原因",
                ],
            }
        )
    return {"knowledge_needs": needs}


def test_topic_source_calls_read_only_detail_endpoint():
    calls = []

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return _topic_payload()

    source = PatternTopicBuildSource(
        PatternTopicBuildConfig(base_url="https://pattern.example", timeout_seconds=8),
        http_get=lambda url, **kwargs: calls.append((url, kwargs)) or Response(),
    )

    payload = source.fetch(1229)

    assert calls == [
        ("https://pattern.example/api/pattern/topic_builds/1229", {"timeout": 8})
    ]
    assert payload["build"]["id"] == 1229
    assert payload["source_url"].endswith("/1229")


def test_topic_table_generation_uses_six_routes_and_unified_budget():
    seen = {}

    def chat(system, user):
        seen["system"] = system
        seen["user"] = user
        return _llm_output()

    preview = generate_topic_table_preview(
        _topic_payload(),
        topic_id=1392,
        max_queries=8,
        chat_fn=chat,
    )
    data = preview_to_dict(preview)

    assert data["generator_kind"] == "topic_table"
    assert len(data["knowledge_needs"]) == 6
    assert set(data["summary"]["route_counts"]) == set(ROUTES)
    assert data["summary"]["selected_count"] == 8
    assert data["summary"]["persisted"] is False
    assert data["summary"]["search_started"] is False
    assert data["topic"]["dimensions"] == {
        "实质": ["羊"],
        "形式": ["低角度仰拍"],
        "意图": ["治愈感"],
    }
    assert "source_reference_data" not in seen["user"]
    assert "正好有 6 项" in seen["user"]
    assert "废弃" not in seen["user"]
    assert "What、How、Why" in seen["system"]
    assert data["queries"][0]["route"] == "unique_hook"


def test_topic_table_generation_rejects_missing_or_duplicate_routes():
    malformed = _llm_output()
    malformed["knowledge_needs"][-1]["route"] = "unique_hook"

    with pytest.raises(TopicTableGenerationError, match="不完整或有重复"):
        generate_topic_table_preview(
            _topic_payload(),
            topic_id=1392,
            max_queries=18,
            chat_fn=lambda _system, _user: malformed,
        )


def test_topic_table_prompt_and_preview_api_do_not_persist_or_search(monkeypatch):
    class Source:
        def fetch(self, topic_build_id):
            assert topic_build_id == 1229
            return _topic_payload()

    monkeypatch.setattr(query_generation, "_topic_source", lambda _env: Source())
    monkeypatch.setattr(
        query_generation,
        "_topic_query_chat",
        lambda system, user, settings: _llm_output(),
    )
    monkeypatch.setattr(
        query_generation.Settings,
        "from_env",
        classmethod(lambda cls, _env: object()),
    )
    client = TestClient(app)

    prompt = client.get("/api/query-generation/topic-table/prompt")
    assert prompt.status_code == 200
    assert prompt.json()["example_request"]["topic_build_id"] == 1229
    assert len(prompt.json()["routes"]) == 6

    response = client.post(
        "/api/query-generation/topic-table/preview",
        json={"topic_build_id": 1229, "topic_id": 1392, "max_queries": 10},
    )
    assert response.status_code == 200
    body = response.json()
    assert body["source"]["topic_id"] == 1392
    assert body["summary"]["selected_count"] == 10
    assert body["summary"]["persisted"] is False
    assert body["summary"]["search_started"] is False