agent-platform/tests/test_knowledge_service_api.py

from pathlib import Path

from tests.conftest import (
    build_fastapi_test_client,
    build_postgres_database_url,
    build_postgres_engine,
    prepare_known_service_import,
)


def test_knowledge_service_post_contract_matches_frontend(
    tmp_path: Path,
    monkeypatch,
) -> None:
    prepare_known_service_import("knowledge-service")

    from app.bootstrap.app import create_app
    from app.db.models import Base
    from core_db import create_session_factory

    database_url = build_postgres_database_url(tmp_path, "knowledge-api")
    monkeypatch.setenv("AGENT_PLATFORM_DATABASE_URL", database_url)
    monkeypatch.setenv("AGENT_PLATFORM_OBJECT_STORAGE_BACKEND", "memory")
    engine = build_postgres_engine(database_url)
    Base.metadata.create_all(engine)

    app = create_app()
    app.state.session_factory = create_session_factory(engine)
    client = build_fastapi_test_client(app)

    storage_health_response = client.post("/knowledge/storage/health", json={})
    assert storage_health_response.status_code == 200
    assert storage_health_response.json()["data"]["backend"] == "memory"
    assert storage_health_response.json()["data"]["available"] is True

    base_response = client.post(
        "/knowledge/bases/create",
        json={
            "name": "Product Knowledge",
            "description": "Customer-facing docs",
            "metadata": {"owner": "support"},
        },
    )
    assert base_response.status_code == 200
    base_payload = base_response.json()["data"]
    assert base_payload["name"] == "Product Knowledge"
    assert "code" not in base_payload
    assert base_payload["createdTime"]

    list_bases_response = client.post(
        "/knowledge/bases/list",
        json={"page": 1, "pageSize": 20, "keyword": "Product"},
    )
    assert list_bases_response.status_code == 200
    assert list_bases_response.json()["data"]["total"] == 1

    archive_base_response = client.post(
        "/knowledge/bases/status",
        json={"knowledgeBaseId": base_payload["id"], "status": "archived"},
    )
    assert archive_base_response.status_code == 200
    assert archive_base_response.json()["data"]["status"] == "archived"

    restore_base_response = client.post(
        "/knowledge/bases/status",
        json={"knowledgeBaseId": base_payload["id"], "status": "active"},
    )
    assert restore_base_response.status_code == 200
    assert restore_base_response.json()["data"]["status"] == "active"

    document_response = client.post(
        "/knowledge/documents/create",
        json={
            "knowledgeBaseId": base_payload["id"],
            "title": "Invoice FAQ",
            "sourceType": "markdown",
            "contentText": "# Invoice FAQ\nCustomers can download invoices from Billing.",
            "chunkSize": 40,
            "chunkOverlap": 5,
            "asyncMode": False,
            "metadata": {"category": "billing"},
        },
    )
    assert document_response.status_code == 200
    document_payload = document_response.json()["data"]["document"]
    chunk_payloads = document_response.json()["data"]["chunks"]
    assert document_payload["knowledgeBaseId"] == base_payload["id"]
    assert document_payload["status"] == "indexed"
    assert document_payload["objectStorage"]["backend"] == "memory"
    assert chunk_payloads

    storage_status_response = client.post(
        "/knowledge/documents/storage/status",
        json={"documentId": document_payload["id"]},
    )
    assert storage_status_response.status_code == 200
    storage_status_payload = storage_status_response.json()["data"]
    assert storage_status_payload["exists"] is True
    assert storage_status_payload["sizeBytes"] > 0
    assert storage_status_payload["checkedTime"]

    content_response = client.post(
        "/knowledge/documents/content",
        json={
            "documentId": document_payload["id"],
            "includeText": True,
            "includeBase64": True,
        },
    )
    assert content_response.status_code == 200
    content_payload = content_response.json()["data"]
    assert content_payload["contentText"].startswith("# Invoice FAQ")
    assert content_payload["contentBase64"]

    reindex_response = client.post(
        "/knowledge/documents/reindex",
        json={
            "documentId": document_payload["id"],
            "chunkSize": 40,
            "chunkOverlap": 5,
            "asyncMode": False,
        },
    )
    assert reindex_response.status_code == 200
    assert reindex_response.json()["data"]["chunks"]

    chunks_response = client.post(
        "/knowledge/chunks/list",
        json={"page": 1, "pageSize": 20, "documentId": document_payload["id"]},
    )
    assert chunks_response.status_code == 200
    assert chunks_response.json()["data"]["total"] == len(chunk_payloads)

    search_response = client.post(
        "/knowledge/search/query",
        json={
            "knowledgeBaseId": base_payload["id"],
            "query": "download invoices",
            "topK": 3,
            "filters": {"status": "indexed"},
        },
    )
    assert search_response.status_code == 200
    search_payload = search_response.json()["data"]
    assert search_payload
    assert search_payload[0]["document"]["id"] == document_payload["id"]
    assert "scoreDetails" in search_payload[0]

    settings_response = client.post(
        "/knowledge/settings/update",
        json={
            "knowledgeBaseId": base_payload["id"],
            "retrievalMode": "hybrid",
            "embeddingModelId": "auto",
            "rerankModelId": "auto",
            "chunkSize": 600,
            "chunkOverlap": 80,
            "topK": 6,
            "minScore": 0.1,
            "maxCandidates": 40,
            "keywordWeight": 0.4,
            "vectorWeight": 0.45,
            "rerankWeight": 0.15,
            "queryRewrite": True,
            "requireCitations": True,
        },
    )
    assert settings_response.status_code == 200
    assert settings_response.json()["data"]["knowledgeBaseId"] == base_payload["id"]
    assert settings_response.json()["data"]["chunkSize"] == 600

    from app.infrastructure.object_storage import InMemoryObjectStorage

    InMemoryObjectStorage(bucket="agent-platform-knowledge").delete_object(
        object_key=document_payload["objectStorage"]["objectKey"])
    missing_storage_status_response = client.post(
        "/knowledge/documents/storage/status",
        json={"documentId": document_payload["id"]},
    )
    assert missing_storage_status_response.status_code == 200
    assert missing_storage_status_response.json()["data"]["exists"] is False
    missing_reindex_response = client.post(
        "/knowledge/documents/reindex",
        json={
            "documentId": document_payload["id"],
            "chunkSize": 40,
            "chunkOverlap": 5,
            "asyncMode": False,
        },
    )
    assert missing_reindex_response.status_code == 503
    document_detail_response = client.post(
        "/knowledge/documents/detail",
        json={"documentId": document_payload["id"]},
    )
    assert document_detail_response.status_code == 200
    assert document_detail_response.json()["data"]["status"] == "failed"

    delete_document_response = client.post(
        "/knowledge/documents/delete",
        json={"documentId": document_payload["id"]},
    )
    assert delete_document_response.status_code == 200
    assert delete_document_response.json()["data"]["deleted"] is True
    assert delete_document_response.json()["data"]["objectDeleted"] is False

    engine.dispose()
    Base.metadata.clear()