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

from __future__ import annotations

from uuid import uuid4

from acquisition.domain import Query, QueryBatch
from acquisition.queries.builder import persist_query_batch


class FakeRepo:
    def __init__(self) -> None:
        self.batch_kwargs = None
        self.queries = []

    def create_query_batch(self, **kwargs):
        self.batch_kwargs = kwargs
        return QueryBatch(id=uuid4(), **kwargs)

    def add_query(self, **kwargs):
        self.queries.append(kwargs)
        return Query(id=uuid4(), **kwargs)


def test_persist_query_batch_writes_formal_batch_and_query_contract():
    repo = FakeRepo()
    generated = {
        "metadata": {"query_filter_prompt_version": "abc123"},
        "families": [
            {
                "key": "f1",
                "name": "实质 × 模态 × 业务阶段 × 知识类型",
                "axes": ["实质", "模态"],
                "items": [
                    {
                        "query": "反转 视频 脚本 怎么做",
                        "parts": {"实质": "反转", "模态": "视频"},
                        "keep": True,
                        "valid": 8,
                        "relevant": True,
                        "reason": "可搜创作方法",
                    }
                ],
            }
        ],
    }

    batch, count = persist_query_batch(repo, generated, name="formal-demo")

    assert batch.name == "formal-demo"
    assert count == 1
    assert repo.batch_kwargs["status"] == "ready"
    assert repo.batch_kwargs["target_platforms"] == ["xiaohongshu", "weixin", "douyin"]
    row = repo.queries[0]
    assert row["batch_id"] == batch.id
    assert row["query_text"] == "反转 视频 脚本 怎么做"
    assert row["axes"] == {"实质": "反转", "模态": "视频"}
    assert row["keep"] is True
    assert row["filter_reason"] == "可搜创作方法"
    assert row["metadata"]["family_key"] == "f1"
    assert row["metadata"]["valid"] == 8
    assert row["metadata"]["query_filter_prompt_version"] == "abc123"