DataOps-platform/tests/knowledge/test_query_audit.py

from __future__ import annotations

import hashlib


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

    return KnowledgeEvidence(
        chunk_id="device:asset-1:v2",
        content="设备名称：循环泵\n责任人：张工",
        score=0.91,
        retriever="device+lexical",
        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",
    )


def test_query_audit_normalizes_hash_and_persists_only_minimized_metadata():
    from app.core.knowledge.access import KnowledgeAccessContext
    from app.core.knowledge.query_audit import build_query_audit

    context = KnowledgeAccessContext(
        subject_id="11111111-1111-4111-8111-111111111111",
        roles=frozenset({"viewer"}),
        permissions=frozenset({"governance:read"}),
        business_domain_uids=frozenset({"domain-b", "domain-a"}),
        correlation_id="22222222-2222-4222-8222-222222222222",
    )
    record = build_query_audit(
        query="  循环泵   的责任人  ",
        context=context,
        mode="semantic",
        evidence=(_evidence(),),
        cited_points=(("DeviceAsset/asset-1/summary", 2),),
        degraded_components=("vector",),
        latency_ms=17,
    )

    assert record.query_hash == hashlib.sha256(
        "循环泵 的责任人".encode()
    ).hexdigest()
    assert record.roles == ("viewer",)
    assert record.business_domain_uids == ("domain-a", "domain-b")
    assert record.retriever_counts == {"device": 1, "lexical": 1}
    assert record.cited_points == (
        {
            "point_key": "DeviceAsset/asset-1/summary",
            "point_revision": 2,
        },
    )
    assert record.degraded_components == ("vector",)
    assert record.latency_ms == 17
    serialized = repr(record)
    for forbidden in (
        "循环泵",
        "张工",
        "设备名称",
        "password",
        "source_config",
    ):
        assert forbidden not in serialized


def test_query_audit_repository_records_and_lists_safe_fields_only():
    from app.core.knowledge.query_audit import (
        KnowledgeQueryAuditRecord,
        SqlKnowledgeQueryAuditRepository,
    )

    class Session:
        def __init__(self):
            self.statements = []
            self.committed = False
            self.rolled_back = False

        def execute(self, statement, parameters=None):
            self.statements.append((str(statement), parameters))

        def commit(self):
            self.committed = True

        def rollback(self):
            self.rolled_back = True

    session = Session()
    repository = SqlKnowledgeQueryAuditRepository(session)
    record = KnowledgeQueryAuditRecord(
        uid="33333333-3333-4333-8333-333333333333",
        query_hash="a" * 64,
        user_id="11111111-1111-4111-8111-111111111111",
        roles=("viewer",),
        business_domain_uids=("domain-a",),
        mode="semantic",
        retriever_counts={"device": 1},
        cited_points=(),
        degraded_components=("vector",),
        correlation_id="22222222-2222-4222-8222-222222222222",
        latency_ms=9,
    )

    repository.record(record)

    statement, parameters = session.statements[0]
    assert "knowledge_query_audits" in statement
    assert parameters["query_hash"] == "a" * 64
    assert "query" not in parameters
    assert "answer" not in parameters
    assert "content" not in parameters
    assert session.committed is True