DataOps-platform/tests/data_research/test_semantic_knowledge_sync.py

from __future__ import annotations


class RecordingKnowledgeSync:
    def __init__(self):
        self.snapshots = []

    def sync(self, snapshot):
        self.snapshots.append(snapshot)
        return {
            "status": "canonical_active",
            "source_uid": snapshot.source_uid,
            "source_revision": snapshot.source_revision,
        }


def test_published_semantic_asset_projects_to_canonical_knowledge_snapshot():
    from app.core.data_research.semantic_knowledge import (
        SemanticKnowledgeSyncHandler,
    )

    uid = "01900000-0000-7000-8000-000000009001"
    semantic = type(
        "SemanticService",
        (),
        {
            "get_published_version": lambda self, asset_uid, version: {
                "uid": asset_uid,
                "asset_kind": "business_term",
                "code": "TERM_MATERIAL",
                "name": "物料",
                "owner_uid": "01900000-0000-7000-8000-000000009002",
                "business_domain_uid": "01900000-0000-7000-8000-000000009003",
                "status": "published",
                "current_version": version,
                "definition": {
                    "definition": "统一采购、库存和维修语义。",
                    "aliases": ["备件", "物资"],
                },
            }
        },
    )()
    knowledge = RecordingKnowledgeSync()
    handler = SemanticKnowledgeSyncHandler(semantic, knowledge)

    result = handler.handle({"uid": uid, "asset_kind": "business_term", "version": 3})

    assert result["status"] == "canonical_active"
    snapshot = knowledge.snapshots[0]
    assert snapshot.source_type == "BusinessTerm"
    assert snapshot.source_revision == 3
    assert {
        point.content
        for point in snapshot.points
        if "/aliases/" in point.point_key
    } == {"备件", "物资"}


def test_semantic_knowledge_handler_rejects_kind_mismatch():
    import pytest

    from app.core.data_research.semantic_knowledge import (
        SemanticKnowledgeSyncHandler,
    )

    semantic = type(
        "SemanticService",
        (),
        {
            "get_published_version": lambda self, _uid, version: {
                "uid": "01900000-0000-7000-8000-000000009011",
                "asset_kind": "metric",
                "status": "published",
                "current_version": version,
                "definition": {
                    "definition": "完整率。",
                    "formula": "完整数 / 总数",
                    "unit": "%",
                    "dimensions": [{"code": "category", "name": "分类"}],
                },
            }
        },
    )()

    with pytest.raises(ValueError, match="kind"):
        SemanticKnowledgeSyncHandler(
            semantic,
            RecordingKnowledgeSync(),
        ).handle(
            {
                "uid": "01900000-0000-7000-8000-000000009011",
                "asset_kind": "business_term",
                "version": 1,
            }
        )


def test_standard_and_ontology_publications_use_canonical_versioned_knowledge():
    from app.core.data_research.semantic_knowledge import (
        PublishedGovernanceKnowledgeHandler,
    )

    knowledge = RecordingKnowledgeSync()
    handler = PublishedGovernanceKnowledgeHandler(
        knowledge_sync=knowledge,
        standard_loader=lambda uid: {
            "uid": uid,
            "standard_uid": "01900000-0000-7000-8000-000000000810",
            "version": 3,
            "status": "published",
            "name": "客户主数据标准",
            "definition": "统一客户等级",
            "owner": "data-team",
            "business_domain_uid": (
                "01900000-0000-7000-8000-000000000811"
            ),
            "rules": [
                {
                    "uid": "customer-level",
                    "name": "客户等级代码约束",
                    "definition": "必须引用已发布代码值",
                }
            ],
        },
        ontology_loader=lambda uid: {
            "uid": "01900000-0000-7000-8000-000000000820",
            "version_uid": uid,
            "version": 2,
            "status": "published",
            "name": "客户本体",
            "definition": "客户跨域语义关系",
            "owner": "ontology-team",
            "business_domain_uid": (
                "01900000-0000-7000-8000-000000000811"
            ),
            "classes": [
                {
                    "uid": "01900000-0000-7000-8000-000000000821",
                    "name": "客户",
                    "definition": "企业服务对象",
                }
            ],
        },
    )

    standard = handler.handle_standard(
        {"version_uid": "01900000-0000-7000-8000-000000000812"}
    )
    ontology = handler.handle_ontology(
        {"version_uid": "01900000-0000-7000-8000-000000000822"}
    )

    assert standard["status"] == "canonical_active"
    assert ontology["status"] == "canonical_active"
    assert knowledge.snapshots[0].source_type == "DataStandard"
    assert knowledge.snapshots[0].source_revision == 3
    assert knowledge.snapshots[1].source_type == "Ontology"
    assert knowledge.snapshots[1].source_revision == 2