harness_zbt/backend/tests/test_seed_knowledge_command.py

from datetime import UTC, datetime
from pathlib import Path

from zbt.commands.seed_knowledge import seed_documents
from zbt.domains.knowledge.index import InMemoryKnowledgeIndex
from zbt.domains.knowledge.repository import InMemoryKnowledgeRepository
from zbt.domains.knowledge.service import KnowledgeService

NOW = datetime(2026, 7, 29, 10, tzinfo=UTC)


def test_seed_documents_rebuilds_index_for_published_mysql_snapshot(
    tmp_path: Path,
) -> None:
    source_name = "测试保险条款.md"
    (tmp_path / source_name).write_text(
        "疾病医疗责任的等待期为30天。",
        encoding="utf-8",
    )
    repository = InMemoryKnowledgeRepository()
    original_service = KnowledgeService(
        repository,
        lambda: NOW,
        InMemoryKnowledgeIndex(),
    )
    created = original_service.create_document(
        title="测试保险条款",
        document_type="INSURANCE_TERMS",
        source_name=source_name,
        product_code="MED-TEST",
        content="疾病医疗责任的等待期为30天。",
        created_by="SYSTEM_SEED",
    )
    document_id = str(created["document_id"])
    original_service.index_document(document_id)
    original_service.test_search(document_id, "等待期")
    original_service.publish_document(document_id)

    fresh_index = InMemoryKnowledgeIndex()
    restored_service = KnowledgeService(repository, lambda: NOW, fresh_index)
    seed_documents(
        restored_service,
        repository,
        tmp_path,
        (
            {
                "file_name": source_name,
                "title": "测试保险条款",
                "document_type": "INSURANCE_TERMS",
                "product_code": "MED-TEST",
            },
        ),
    )

    result = restored_service.search("等待期")
    assert result["total"] == 1
    restored_document = repository.get_document(document_id)
    assert restored_document is not None
    assert restored_document.status == "PUBLISHED"