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",
    )


class _AuditRepository:
    def __init__(self, records=()):
        self.records = list(records)
        self.recorded = []

    def record(self, record):
        self.recorded.append(record)

    def list(self, *, limit):
        return tuple(self.records[:limit])


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()
    audit = _AuditRepository()
    app.extensions["knowledge_query_audit_repository"] = audit

    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
    assert len(audit.recorded) == 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 必须为整数"


def test_device_search_and_grounded_answer_return_safe_evidence_and_citation_content(
    monkeypatch,
):
    from app import create_app
    from app.core.knowledge.access import KnowledgeAccessContext
    from app.core.knowledge.qa import AnswerResult, Citation
    from app.core.knowledge.retrieval.contracts import (
        KnowledgeEvidence,
        SearchResult,
    )

    evidence = KnowledgeEvidence(
        chunk_id="device:asset-1:v2",
        content="设备名称：循环泵\n源 ID：EQ-001\n责任人：张工",
        score=0.95,
        retriever="device",
        object_uid="asset-1",
        object_type="DeviceAsset",
        object_version=2,
        business_domain_uid="domain-a",
        point_keys=("DeviceAsset/asset-1/summary",),
        point_revisions=(2,),
        generation=2,
        source_updated_at="2026-07-29T08:00:00+00:00",
    )

    class Pipeline:
        def search(self, *_args, **_kwargs):
            return SearchResult((evidence,), "exact", ("vector",))

    class Synthesizer:
        def answer(self, _query, _evidence):
            return AnswerResult(
                "grounded",
                "循环泵的责任人是张工。",
                (
                    Citation(
                        object_uid="asset-1",
                        object_type="DeviceAsset",
                        object_version=2,
                        point_key="DeviceAsset/asset-1/summary",
                        point_revision=2,
                        chunk_id="device:asset-1:v2",
                        section_path=None,
                        source_updated_at="2026-07-29T08:00:00+00:00",
                        index_generation=2,
                        retrievers=("device",),
                        score=0.95,
                        freshness_status="fresh",
                    ),
                ),
                "fresh",
            )

    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {
            "id": "11111111-1111-4111-8111-111111111111",
            "roles": ["viewer"],
        },
    )
    monkeypatch.setattr(
        "app.api.knowledge_base.routes.build_access_context",
        lambda *_args, **_kwargs: KnowledgeAccessContext(
            subject_id="11111111-1111-4111-8111-111111111111",
            roles=frozenset({"viewer"}),
            permissions=frozenset({"governance:read"}),
            business_domain_uids=frozenset({"domain-a"}),
            correlation_id="22222222-2222-4222-8222-222222222222",
        ),
    )
    app = create_app()
    app.extensions["knowledge_retrieval_pipeline"] = Pipeline()
    app.extensions["knowledge_answer_synthesizer"] = Synthesizer()
    audit = _AuditRepository()
    app.extensions["knowledge_query_audit_repository"] = audit

    client = app.test_client()
    search_response = client.post(
        "/api/knowledge/search",
        json={"query": "EQ-001", "mode": "exact"},
    )
    ask_response = client.post(
        "/api/knowledge/ask",
        json={"query": "循环泵的责任人是谁？"},
    )

    assert search_response.status_code == 200
    assert (
        search_response.get_json()["data"]["evidence"][0]["object_type"]
        == "DeviceAsset"
    )
    assert ask_response.status_code == 200
    answer = ask_response.get_json()["data"]
    assert answer["evidence"][0]["content"].endswith("责任人：张工")
    assert answer["citations"][0]["content"].endswith("责任人：张工")
    assert answer["citations"][0]["object_uid"] == "asset-1"
    assert len(audit.recorded) == 2
    assert audit.recorded[1].cited_points == (
        {
            "point_key": "DeviceAsset/asset-1/summary",
            "point_revision": 2,
        },
    )


def test_model_unavailable_returns_no_answer_but_keeps_authorized_evidence(
    monkeypatch,
):
    from app import create_app
    from app.core.knowledge.access import KnowledgeAccessContext
    from app.core.knowledge.qa import AnswerResult
    from app.core.knowledge.retrieval.contracts import SearchResult

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

    class Synthesizer:
        def answer(self, _query, _evidence):
            return AnswerResult(
                "model_unavailable",
                None,
                (),
                "degraded",
            )

    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {
            "id": "11111111-1111-4111-8111-111111111111",
            "roles": ["viewer"],
        },
    )
    monkeypatch.setattr(
        "app.api.knowledge_base.routes.build_access_context",
        lambda *_args, **_kwargs: KnowledgeAccessContext(
            subject_id="11111111-1111-4111-8111-111111111111",
            roles=frozenset({"viewer"}),
            permissions=frozenset({"governance:read"}),
            business_domain_uids=frozenset({"domain-a"}),
            correlation_id="22222222-2222-4222-8222-222222222222",
        ),
    )
    app = create_app()
    app.extensions["knowledge_retrieval_pipeline"] = Pipeline()
    app.extensions["knowledge_answer_synthesizer"] = Synthesizer()
    app.extensions["knowledge_query_audit_repository"] = _AuditRepository()

    response = app.test_client().post(
        "/api/knowledge/ask",
        json={"query": "用途是什么？"},
    )

    assert response.status_code == 200
    data = response.get_json()["data"]
    assert data["answer"] is None
    assert data["answer_status"] == "model_unavailable"
    assert data["citations"] == []
    assert data["evidence"][0]["content"] == "客户同步用途"


def test_device_source_detail_is_authorized_and_contains_no_private_payload(
    monkeypatch,
):
    from datetime import UTC, datetime

    from app import create_app
    from app.core.knowledge.access import KnowledgeAccessContext
    from app.core.knowledge.retrieval.device import DeviceSearchRow

    class Repository:
        def get_detail(self, asset_uid, **_kwargs):
            if asset_uid != "11111111-1111-4111-8111-111111111111":
                return None
            return DeviceSearchRow(
                asset_uid=asset_uid,
                asset_type="device",
                name="循环泵",
                current_version=2,
                location="动力车间",
                organization="设备动力部",
                responsible_person="张工",
                source_codes=("EQ-001",),
                related_events=(("fault", "轴承故障", "FT-001"),),
                business_domain_uid="domain-a",
                updated_at=datetime(2026, 7, 29, tzinfo=UTC),
                rank=1.0,
            )

    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {
            "id": "11111111-1111-4111-8111-111111111111",
            "roles": ["viewer"],
        },
    )
    monkeypatch.setattr(
        "app.api.knowledge_base.routes._source_document",
        lambda *_args, **_kwargs: None,
    )
    monkeypatch.setattr(
        "app.api.knowledge_base.routes._request_context",
        lambda _payload: (
            KnowledgeAccessContext(
                subject_id="11111111-1111-4111-8111-111111111111",
                roles=frozenset({"viewer"}),
                permissions=frozenset({"governance:read"}),
                business_domain_uids=frozenset({"domain-a"}),
                correlation_id="22222222-2222-4222-8222-222222222222",
            ),
            "22222222-2222-4222-8222-222222222222",
        ),
    )
    app = create_app()
    app.extensions["device_knowledge_repository"] = Repository()

    response = app.test_client().get(
        "/api/knowledge/sources/11111111-1111-4111-8111-111111111111"
    )

    assert response.status_code == 200
    data = response.get_json()["data"]
    assert data["object_type"] == "DeviceAsset"
    assert data["source_codes"] == ["EQ-001"]
    assert data["related_events"][0]["title"] == "轴承故障"
    assert "config" not in repr(data)
    assert "permission_scope" not in repr(data)


def test_admin_can_manage_device_source_scope_and_list_minimized_query_audits(
    monkeypatch,
):
    from app import create_app
    from app.core.data_research.sources import IngestionSourceRecord

    domain_uid = "33333333-3333-4333-8333-333333333333"
    source_uid = "44444444-4444-4444-8444-444444444444"

    class ScopeService:
        def list(self):
            return (
                {
                    "uid": source_uid,
                    "name": "设备源",
                    "business_domains": (),
                    "admin_only": True,
                },
            )

        def update(self, uid, payload, *, actor_is_admin):
            assert uid == source_uid
            assert payload == {"business_domains": [domain_uid]}
            assert actor_is_admin is True
            return IngestionSourceRecord(
                uid=uid,
                source_type="database",
                name="设备源",
                config={"password": "private"},
                permission_scope={"business_domains": [domain_uid]},
                status="active",
                created_by="admin-1",
            )

    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {
            "id": "11111111-1111-4111-8111-111111111111",
            "roles": ["admin"],
        },
    )
    audit_row = {
        "id": "audit-1",
        "query_hash": "a" * 64,
        "mode": "exact",
        "retriever_counts": {"device": 1},
        "cited_points": [],
        "degraded_components": ["vector"],
        "correlation_id": "correlation-1",
        "created_at": "2026-07-29T08:00:00+00:00",
    }
    app = create_app()
    app.extensions["device_source_scope_service"] = ScopeService()
    app.extensions["knowledge_query_audit_repository"] = _AuditRepository(
        (audit_row,)
    )
    client = app.test_client()

    listed = client.get("/api/knowledge/admin/device-sources")
    updated = client.put(
        f"/api/knowledge/admin/device-sources/{source_uid}/scope",
        json={"business_domains": [domain_uid]},
    )
    audits = client.get("/api/knowledge/admin/query-audits")

    assert listed.status_code == 200
    assert listed.get_json()["data"][0]["admin_only"] is True
    assert updated.status_code == 200
    updated_data = updated.get_json()["data"]
    assert updated_data["business_domains"] == [domain_uid]
    assert "config" not in repr(updated_data)
    assert audits.status_code == 200
    assert audits.get_json()["data"][0]["query_hash"] == "a" * 64
    assert "query" not in audits.get_json()["data"][0]


def test_executed_search_fails_when_mandatory_audit_persistence_fails(
    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")

    class Audit:
        def record(self, _record):
            raise RuntimeError("audit unavailable")

    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {
            "id": "11111111-1111-4111-8111-111111111111",
            "roles": ["viewer"],
        },
    )
    monkeypatch.setattr(
        "app.api.knowledge_base.routes.build_access_context",
        lambda *_args, **_kwargs: KnowledgeAccessContext(
            subject_id="11111111-1111-4111-8111-111111111111",
            roles=frozenset({"viewer"}),
            permissions=frozenset({"governance:read"}),
            business_domain_uids=frozenset({"domain-a"}),
            correlation_id="22222222-2222-4222-8222-222222222222",
        ),
    )
    app = create_app()
    app.extensions["knowledge_retrieval_pipeline"] = Pipeline()
    app.extensions["knowledge_query_audit_repository"] = Audit()

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

    assert response.status_code == 503
    assert response.get_json()["message"] == "知识查询审计写入失败"