DataOps-platform/tests/knowledge/test_api.py

from __future__ import annotations


def _evidence():
    from app.core.knowledge.retrieval.contracts import KnowledgeEvidence

    return KnowledgeEvidence(
        chunk_id="chunk-1",
        content="客户同步用途",
        score=0.9,
        retriever="lexical+vector",
        object_uid="object-1",
        object_type="DataFlow",
        object_version=2,
        business_domain_uid="domain-a",
        point_keys=("DataFlow/object-1/purpose",),
        point_revisions=(3,),
        generation=1,
        source_updated_at="2026-07-23T00:00:00+00:00",
    )


def test_viewer_can_post_search_and_receives_stable_canonical_contract(monkeypatch):
    from app import create_app
    from app.core.knowledge.access import KnowledgeAccessContext
    from app.core.knowledge.retrieval.contracts import SearchResult

    class Pipeline:
        def search(self, *_args, **_kwargs):
            return SearchResult((_evidence(),), "semantic")

    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {"id": "user-1", "roles": ["viewer"]},
    )
    monkeypatch.setattr(
        "app.api.knowledge_base.routes.build_access_context",
        lambda *_args, **_kwargs: KnowledgeAccessContext(
            subject_id="user-1",
            roles=frozenset({"viewer"}),
            permissions=frozenset({"governance:read"}),
            business_domain_uids=frozenset({"domain-a"}),
            correlation_id="correlation-1",
        ),
    )
    app = create_app()
    app.extensions["knowledge_retrieval_pipeline"] = Pipeline()

    response = app.test_client().post(
        "/api/knowledge/search", json={"query": "用途", "mode": "semantic"}
    )

    assert response.status_code == 200
    payload = response.get_json()["data"]
    assert payload["evidence"][0]["object_uid"] == "object-1"
    assert payload["evidence"][0]["point_revisions"] == [3]
    assert payload["evidence"][0]["index_generation"] == 1


def test_search_rejects_client_domain_filter_shape_before_retrieval(monkeypatch):
    from app import create_app

    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {"id": "user-1", "roles": ["viewer"]},
    )
    app = create_app()
    response = app.test_client().post(
        "/api/knowledge/search",
        json={"query": "用途", "business_domain_uids": "domain-a"},
    )

    assert response.status_code == 400


def test_search_rejects_non_numeric_limit(monkeypatch):
    from app import create_app

    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {"id": "user-1", "roles": ["viewer"]},
    )
    app = create_app()
    response = app.test_client().post(
        "/api/knowledge/search",
        json={"query": "用途", "limit": "many"},
    )

    assert response.status_code == 400
    assert response.get_json()["message"] == "limit 必须为整数"