DataOps-platform/tests/knowledge/test_point_builder.py

from __future__ import annotations


def _point_map(snapshot):
    return {point.point_key: point for point in snapshot.points}


def test_dataflow_point_keys_are_stable_across_order_and_display_name_changes():
    from app.core.knowledge.point_builder import build_knowledge_snapshot

    source = {
        "uid": "01900000-0000-7000-8000-000000000101",
        "version": 7,
        "name_zh": "客户同步",
        "purpose": "同步客户主数据",
        "owner": "治理组",
        "business_domain_uid": "01900000-0000-7000-8000-000000000201",
        "aliases": ["customer_sync", "客户主数据同步"],
        "relations": [
            {
                "type": "READS_FROM",
                "target_uid": "01900000-0000-7000-8000-000000000301",
                "target_name": "客户源表",
            },
            {
                "type": "WRITES_TO",
                "target_uid": "01900000-0000-7000-8000-000000000302",
                "target_name": "客户主表",
            },
        ],
    }

    first = build_knowledge_snapshot("DataFlow", source)
    second = build_knowledge_snapshot(
        "DataFlow",
        {
            **source,
            "name_zh": "客户资料同步",
            "aliases": list(reversed(source["aliases"])),
            "relations": list(reversed(source["relations"])),
        },
    )

    first_points = _point_map(first)
    second_points = _point_map(second)
    assert set(first_points) == set(second_points)
    assert first.point_set_hash != second.point_set_hash
    assert (
        first_points[
            "DataFlow/01900000-0000-7000-8000-000000000101/purpose"
        ].content_hash
        == second_points[
            "DataFlow/01900000-0000-7000-8000-000000000101/purpose"
        ].content_hash
    )


def test_business_domain_builder_redacts_secrets_and_requires_stable_child_uid():
    import pytest

    from app.core.knowledge.point_builder import build_knowledge_snapshot

    snapshot = build_knowledge_snapshot(
        "BusinessDomain",
        {
            "uid": "01900000-0000-7000-8000-000000000401",
            "version": 2,
            "name_zh": "客户域",
            "definition": "统一客户信息",
            "password": "must-not-leak",
            "fields": [
                {
                    "uid": "01900000-0000-7000-8000-000000000402",
                    "name": "customer_id",
                    "definition": "客户唯一标识",
                }
            ],
        },
    )

    serialized = "\n".join(point.content for point in snapshot.points)
    assert "must-not-leak" not in serialized
    assert (
        "BusinessDomain/01900000-0000-7000-8000-000000000401/"
        "fields/01900000-0000-7000-8000-000000000402/definition"
    ) in _point_map(snapshot)

    with pytest.raises(ValueError, match="stable uid"):
        build_knowledge_snapshot(
            "BusinessDomain",
            {
                "uid": "01900000-0000-7000-8000-000000000401",
                "version": 3,
                "fields": [{"name": "customer_name", "definition": "姓名"}],
            },
        )


def test_same_semantic_source_produces_same_snapshot_hash():
    from app.core.knowledge.point_builder import build_knowledge_snapshot

    source = {
        "uid": "01900000-0000-7000-8000-000000000501",
        "version": 1,
        "name": "Customer",
        "definition": "Customer domain",
        "permission_scope": {"business_domains": ["b", "a"]},
    }
    first = build_knowledge_snapshot("BusinessDomain", source)
    second = build_knowledge_snapshot(
        "BusinessDomain",
        {
            "definition": "Customer domain",
            "name": "Customer",
            "version": 1,
            "uid": source["uid"],
            "permission_scope": {"business_domains": ["a", "b"]},
        },
    )
    assert first.source_snapshot_hash == second.source_snapshot_hash
    assert first.point_set_hash == second.point_set_hash


def test_semantic_assets_build_versioned_knowledge_points():
    from app.core.knowledge.point_builder import build_knowledge_snapshot

    term_uid = "01900000-0000-7000-8000-000000000601"
    metric_uid = "01900000-0000-7000-8000-000000000602"
    term = build_knowledge_snapshot(
        "BusinessTerm",
        {
            "uid": term_uid,
            "version": 3,
            "name": "物料",
            "definition": "统一采购、库存和维修语义。",
            "aliases": ["备件", "物资"],
            "business_domain_uid": "01900000-0000-7000-8000-000000000603",
        },
    )
    metric = build_knowledge_snapshot(
        "MetricDefinition",
        {
            "uid": metric_uid,
            "version": 2,
            "name": "物料完整率",
            "definition": "完整物料占全部物料的比例。",
            "formula": "完整物料数 / 全部物料数",
            "unit": "%",
            "dimensions": [
                {
                    "uid": "category",
                    "name": "物料分类",
                    "definition": "按物料分类观察。",
                }
            ],
        },
    )

    assert term.source_type == "BusinessTerm"
    assert f"BusinessTerm/{term_uid}/aliases/" in "\n".join(
        point.point_key for point in term.points
    )
    metric_points = _point_map(metric)
    assert metric_points[f"MetricDefinition/{metric_uid}/formula"].content == (
        "完整物料数 / 全部物料数"
    )
    assert (
        f"MetricDefinition/{metric_uid}/dimensions/category/definition"
        in metric_points
    )


def test_ontology_builds_stable_class_and_property_points():
    from app.core.knowledge.point_builder import build_knowledge_snapshot

    snapshot = build_knowledge_snapshot(
        "Ontology",
        {
            "uid": "01900000-0000-7000-8000-000000000901",
            "version": 2,
            "name": "客户本体",
            "classes": [
                {
                    "uid": "01900000-0000-7000-8000-000000000902",
                    "name": "客户",
                    "definition": "企业服务对象",
                }
            ],
            "properties": [
                {
                    "uid": "01900000-0000-7000-8000-000000000903",
                    "name": "客户等级",
                    "data_type": "string",
                }
            ],
        },
    )

    assert snapshot.source_type == "Ontology"
    point_keys = {point.point_key for point in snapshot.points}
    assert any("classes/" in key and key.endswith("/name") for key in point_keys)
    assert any(
        "properties/" in key and key.endswith("/data_type")
        for key in point_keys
    )