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