harness_zbt/backend/tests/test_knowledge_api.py

from datetime import UTC, datetime

from fastapi.testclient import TestClient

from zbt.core.config import Settings
from zbt.core.passwords import PasswordService
from zbt.domains.identity.models import AdminUser
from zbt.domains.identity.repository import InMemoryIdentityRepository
from zbt.domains.knowledge.index import InMemoryKnowledgeIndex
from zbt.domains.knowledge.repository import InMemoryKnowledgeRepository
from zbt.main import create_app


def test_admin_creates_draft_knowledge_document() -> None:
    settings = Settings(
        app_env="test",
        jwt_access_secret="a" * 32,
        jwt_refresh_secret="b" * 32,
        field_encryption_key="c" * 32,
    )
    identities = InMemoryIdentityRepository()
    identities.save_admin_user(
        AdminUser(
            id="01ADMIN0000000000000000001",
            username="admin",
            password_hash=PasswordService().hash("zaq1XSW@"),
            display_name="知识管理员",
            status="ACTIVE",
            roles=("SUPER_ADMIN",),
            permissions=("knowledge:read", "knowledge:write"),
            data_scope="ALL",
        )
    )
    knowledge = InMemoryKnowledgeRepository()
    application = create_app(
        settings=settings,
        identity_repository=identities,
        knowledge_repository=knowledge,
        clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC),
    )

    with TestClient(application) as client:
        login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        )
        token = login.json()["data"]["tokens"]["access_token"]
        response = client.post(
            "/api/v1/admin/knowledge/documents",
            headers={"Authorization": f"Bearer {token}"},
            json={
                "title": "安心医疗险保险条款",
                "document_type": "INSURANCE_TERMS",
                "source_name": "安心医疗险条款.pdf",
                "product_code": "MED-BASIC",
                "content": "本产品等待期为30天，意外伤害导致的医疗费用不受等待期限制。",
            },
        )

    assert response.status_code == 201
    document = response.json()["data"]
    assert document["title"] == "安心医疗险保险条款"
    assert document["document_type"] == "INSURANCE_TERMS"
    assert document["source_name"] == "安心医疗险条款.pdf"
    assert document["product_code"] == "MED-BASIC"
    assert document["status"] == "DRAFT"
    assert document["version_no"] == 1
    assert document["chunk_count"] == 0
    assert document["indexed_at"] is None
    assert document["published_at"] is None


def test_admin_cannot_publish_document_before_indexing() -> None:
    settings = Settings(
        app_env="test",
        jwt_access_secret="a" * 32,
        jwt_refresh_secret="b" * 32,
        field_encryption_key="c" * 32,
    )
    identities = InMemoryIdentityRepository()
    identities.save_admin_user(
        AdminUser(
            id="01ADMIN0000000000000000001",
            username="admin",
            password_hash=PasswordService().hash("zaq1XSW@"),
            display_name="知识管理员",
            status="ACTIVE",
            roles=("SUPER_ADMIN",),
            permissions=("knowledge:read", "knowledge:write", "knowledge:publish"),
            data_scope="ALL",
        )
    )
    application = create_app(
        settings=settings,
        identity_repository=identities,
        knowledge_repository=InMemoryKnowledgeRepository(),
        clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC),
    )

    with TestClient(application) as client:
        login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        )
        headers = {
            "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}"
        }
        created = client.post(
            "/api/v1/admin/knowledge/documents",
            headers=headers,
            json={
                "title": "安心医疗险保险条款",
                "document_type": "INSURANCE_TERMS",
                "source_name": "安心医疗险条款.pdf",
                "content": "本产品等待期为30天，意外伤害导致的医疗费用不受等待期限制。",
            },
        )
        document_id = created.json()["data"]["document_id"]
        response = client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/publish",
            headers=headers,
        )

    assert response.status_code == 409
    assert response.json()["error"]["code"] == "KNOWLEDGE_DOCUMENT_NOT_INDEXED"


def test_admin_cannot_publish_indexed_document_without_effective_search_test() -> None:
    settings = Settings(
        app_env="test",
        jwt_access_secret="a" * 32,
        jwt_refresh_secret="b" * 32,
        field_encryption_key="c" * 32,
    )
    identities = InMemoryIdentityRepository()
    identities.save_admin_user(
        AdminUser(
            id="01ADMIN0000000000000000001",
            username="admin",
            password_hash=PasswordService().hash("zaq1XSW@"),
            display_name="知识管理员",
            status="ACTIVE",
            roles=("SUPER_ADMIN",),
            permissions=("knowledge:read", "knowledge:write", "knowledge:publish"),
            data_scope="ALL",
        )
    )
    application = create_app(
        settings=settings,
        identity_repository=identities,
        knowledge_repository=InMemoryKnowledgeRepository(),
        knowledge_index=InMemoryKnowledgeIndex(),
        clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC),
    )

    with TestClient(application) as client:
        login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        )
        headers = {
            "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}"
        }
        created = client.post(
            "/api/v1/admin/knowledge/documents",
            headers=headers,
            json={
                "title": "安心医疗险保险条款",
                "document_type": "INSURANCE_TERMS",
                "source_name": "安心医疗险条款.pdf",
                "content": "本产品等待期为30天，意外伤害不受等待期限制。",
            },
        )
        document_id = created.json()["data"]["document_id"]
        client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/index",
            headers=headers,
        )
        response = client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/publish",
            headers=headers,
        )

    assert response.status_code == 409
    assert response.json()["error"]["code"] == "KNOWLEDGE_SEARCH_TEST_REQUIRED"


def test_admin_indexes_then_publishes_document() -> None:
    settings = Settings(
        app_env="test",
        jwt_access_secret="a" * 32,
        jwt_refresh_secret="b" * 32,
        field_encryption_key="c" * 32,
    )
    identities = InMemoryIdentityRepository()
    identities.save_admin_user(
        AdminUser(
            id="01ADMIN0000000000000000001",
            username="admin",
            password_hash=PasswordService().hash("zaq1XSW@"),
            display_name="知识管理员",
            status="ACTIVE",
            roles=("SUPER_ADMIN",),
            permissions=("knowledge:read", "knowledge:write", "knowledge:publish"),
            data_scope="ALL",
        )
    )
    application = create_app(
        settings=settings,
        identity_repository=identities,
        knowledge_repository=InMemoryKnowledgeRepository(),
        knowledge_index=InMemoryKnowledgeIndex(),
        clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC),
    )

    with TestClient(application) as client:
        login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        )
        headers = {
            "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}"
        }
        created = client.post(
            "/api/v1/admin/knowledge/documents",
            headers=headers,
            json={
                "title": "安心医疗险保险条款",
                "document_type": "INSURANCE_TERMS",
                "source_name": "安心医疗险条款.pdf",
                "content": (
                    "等待期规则\n本产品疾病医疗等待期为30天，意外伤害不受等待期限制。\n\n"
                    "报销材料\n申请医疗费用报销协助时，需要提供发票、费用清单和诊断证明。"
                ),
            },
        )
        document_id = created.json()["data"]["document_id"]
        indexed = client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/index",
            headers=headers,
        )
        search_test = client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/search-test",
            headers=headers,
            json={"query": "等待期是多少天？", "limit": 3},
        )
        published = client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/publish",
            headers=headers,
        )
        search_tests = client.get(
            f"/api/v1/admin/knowledge/documents/{document_id}/search-tests",
            headers=headers,
        )

    assert indexed.status_code == 200
    assert indexed.json()["data"]["status"] == "INDEXED"
    assert indexed.json()["data"]["chunk_count"] == 2
    assert indexed.json()["data"]["indexed_at"] is not None
    assert search_test.status_code == 200
    assert search_test.json()["data"]["test_record"]["passed"] is True
    assert search_test.json()["data"]["test_record"]["hit_count"] >= 1
    assert published.status_code == 200
    assert published.json()["data"]["status"] == "PUBLISHED"
    assert published.json()["data"]["published_at"] is not None
    assert published.json()["data"]["search_test_count"] == 1
    assert published.json()["data"]["last_search_test_passed"] is True
    assert search_tests.status_code == 200
    assert search_tests.json()["data"]["total"] == 1
    assert search_tests.json()["data"]["items"][0]["query"] == "等待期是多少天？"


def test_admin_lists_and_disables_published_document() -> None:
    settings = Settings(
        app_env="test",
        jwt_access_secret="a" * 32,
        jwt_refresh_secret="b" * 32,
        field_encryption_key="c" * 32,
    )
    identities = InMemoryIdentityRepository()
    identities.save_admin_user(
        AdminUser(
            id="01ADMIN0000000000000000001",
            username="admin",
            password_hash=PasswordService().hash("zaq1XSW@"),
            display_name="知识管理员",
            status="ACTIVE",
            roles=("SUPER_ADMIN",),
            permissions=("knowledge:read", "knowledge:write", "knowledge:publish"),
            data_scope="ALL",
        )
    )
    application = create_app(
        settings=settings,
        identity_repository=identities,
        knowledge_repository=InMemoryKnowledgeRepository(),
        knowledge_index=InMemoryKnowledgeIndex(),
        clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC),
    )

    with TestClient(application) as client:
        login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        )
        headers = {
            "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}"
        }
        created = client.post(
            "/api/v1/admin/knowledge/documents",
            headers=headers,
            json={
                "title": "安心医疗险保险条款",
                "document_type": "INSURANCE_TERMS",
                "source_name": "安心医疗险条款.pdf",
                "content": "本产品疾病医疗等待期为30天，意外伤害不受等待期限制。",
            },
        )
        document_id = created.json()["data"]["document_id"]
        client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/index",
            headers=headers,
        )
        client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/search-test",
            headers=headers,
            json={"query": "等待期是多少天？"},
        )
        client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/publish",
            headers=headers,
        )
        disabled = client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/disable",
            headers=headers,
        )
        listing = client.get("/api/v1/admin/knowledge/documents", headers=headers)

    assert disabled.status_code == 200
    assert disabled.json()["data"]["status"] == "INACTIVE"
    assert listing.status_code == 200
    assert listing.json()["data"]["total"] == 1
    assert listing.json()["data"]["items"][0]["status"] == "INACTIVE"


def test_admin_uploads_utf8_text_as_draft_document() -> None:
    settings = Settings(
        app_env="test",
        jwt_access_secret="a" * 32,
        jwt_refresh_secret="b" * 32,
        field_encryption_key="c" * 32,
    )
    identities = InMemoryIdentityRepository()
    identities.save_admin_user(
        AdminUser(
            id="01ADMIN0000000000000000001",
            username="admin",
            password_hash=PasswordService().hash("zaq1XSW@"),
            display_name="知识管理员",
            status="ACTIVE",
            roles=("SUPER_ADMIN",),
            permissions=("knowledge:read", "knowledge:write"),
            data_scope="ALL",
        )
    )
    application = create_app(
        settings=settings,
        identity_repository=identities,
        knowledge_repository=InMemoryKnowledgeRepository(),
        knowledge_index=InMemoryKnowledgeIndex(),
        clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC),
    )

    with TestClient(application) as client:
        login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        )
        headers = {
            "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}"
        }
        response = client.post(
            "/api/v1/admin/knowledge/documents/upload",
            headers=headers,
            data={
                "title": "医疗费用报销材料说明",
                "document_type": "SERVICE_RULES",
                "product_code": "MED-BASIC",
            },
            files={
                "file": (
                    "医疗费用报销材料说明.txt",
                    "申请时需要提供发票、费用清单和诊断证明。".encode(),
                    "text/plain",
                )
            },
        )

    assert response.status_code == 201
    document = response.json()["data"]
    assert document["title"] == "医疗费用报销材料说明"
    assert document["source_name"] == "医疗费用报销材料说明.txt"
    assert document["document_type"] == "SERVICE_RULES"
    assert document["status"] == "DRAFT"


def test_admin_tests_indexed_document_before_publishing() -> None:
    settings = Settings(
        app_env="test",
        jwt_access_secret="a" * 32,
        jwt_refresh_secret="b" * 32,
        field_encryption_key="c" * 32,
    )
    identities = InMemoryIdentityRepository()
    identities.save_admin_user(
        AdminUser(
            id="01ADMIN0000000000000000001",
            username="admin",
            password_hash=PasswordService().hash("zaq1XSW@"),
            display_name="知识管理员",
            status="ACTIVE",
            roles=("SUPER_ADMIN",),
            permissions=("knowledge:read", "knowledge:write"),
            data_scope="ALL",
        )
    )
    application = create_app(
        settings=settings,
        identity_repository=identities,
        knowledge_repository=InMemoryKnowledgeRepository(),
        knowledge_index=InMemoryKnowledgeIndex(),
        clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC),
    )

    with TestClient(application) as client:
        login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        )
        headers = {
            "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}"
        }
        created = client.post(
            "/api/v1/admin/knowledge/documents",
            headers=headers,
            json={
                "title": "安心医疗险保险条款",
                "document_type": "INSURANCE_TERMS",
                "source_name": "安心医疗险条款.md",
                "content": "疾病医疗等待期为30天，意外伤害不受等待期限制。",
            },
        )
        document_id = created.json()["data"]["document_id"]
        client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/index",
            headers=headers,
        )
        response = client.post(
            f"/api/v1/admin/knowledge/documents/{document_id}/search-test",
            headers=headers,
            json={"query": "等待期是多少天？", "limit": 3},
        )

    assert response.status_code == 200
    result = response.json()["data"]
    assert result["document_id"] == document_id
    assert result["total"] == 1
    assert "30天" in result["items"][0]["content"]