DataOps-platform/tests/knowledge/test_diff.py

from __future__ import annotations

from dataclasses import replace


def test_diff_classifies_added_modified_deleted_and_unchanged():
    from app.core.knowledge.diff import diff_snapshots
    from app.core.knowledge.point_builder import build_knowledge_snapshot

    old = build_knowledge_snapshot(
        "DataFlow",
        {
            "uid": "01900000-0000-7000-8000-000000000601",
            "version": 1,
            "name": "sync",
            "purpose": "old purpose",
            "owner": "team-a",
        },
    )
    new = build_knowledge_snapshot(
        "DataFlow",
        {
            "uid": "01900000-0000-7000-8000-000000000601",
            "version": 2,
            "name": "sync",
            "purpose": "new purpose",
            "aliases": ["sync_job"],
        },
    )

    result = diff_snapshots(old, new)

    assert len(result.modified) == 1
    assert result.modified[0].point_key.endswith("/purpose")
    assert len(result.added) == 1
    assert "/aliases/" in result.added[0].point_key
    assert len(result.deleted) == 1
    assert result.deleted[0].point_key.endswith("/owner")
    assert any(change.point_key.endswith("/name") for change in result.unchanged)


def test_permission_change_is_modified_even_when_content_is_same():
    from app.core.knowledge.diff import diff_snapshots
    from app.core.knowledge.point_builder import build_knowledge_snapshot

    old = build_knowledge_snapshot(
        "BusinessDomain",
        {
            "uid": "01900000-0000-7000-8000-000000000701",
            "version": 1,
            "definition": "customer",
            "permission_scope": {"business_domains": ["a", "b"]},
        },
    )
    new = replace(
        build_knowledge_snapshot(
            "BusinessDomain",
            {
                "uid": "01900000-0000-7000-8000-000000000701",
                "version": 2,
                "definition": "customer",
                "permission_scope": {"business_domains": ["a"]},
            },
        ),
    )

    result = diff_snapshots(old, new)
    assert not result.unchanged
    assert result.modified
    assert all(change.permission_changed for change in result.modified)


def test_diff_rejects_different_source_objects():
    import pytest

    from app.core.knowledge.diff import diff_snapshots
    from app.core.knowledge.point_builder import build_knowledge_snapshot

    first = build_knowledge_snapshot(
        "Label", {"uid": "01900000-0000-7000-8000-000000000801", "name": "PII"}
    )
    second = build_knowledge_snapshot(
        "Label", {"uid": "01900000-0000-7000-8000-000000000802", "name": "PII"}
    )
    with pytest.raises(ValueError, match="same source"):
        diff_snapshots(first, second)