DataOpsShowcase/tests/test_entity_publish_validation.py at master · svergio/DataOpsShowcase · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
from __future__ import annotations

import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "infra" / "metadata" / "atlas" / "scripts"))

from entity_publish import validate_entities  # noqa: E402


def test_validate_duplicate_qualified_names_rejected() -> None:
    entities = [
        {"typeName": "Process", "attributes": {"qualifiedName": "dup", "name": "a"}},
        {"typeName": "Process", "attributes": {"qualifiedName": "dup", "name": "b"}},
    ]
    errors = validate_entities(entities)
    assert errors


def test_validate_unique_ok() -> None:
    entities = [
        {"typeName": "Process", "attributes": {"qualifiedName": "a", "name": "a"}},
        {"typeName": "Process", "attributes": {"qualifiedName": "b", "name": "b"}},
    ]
    assert not validate_entities(entities)


def test_kafka_topic_requires_uri() -> None:
    entities = [
        {
            "typeName": "kafka_topic",
            "attributes": {"name": "x", "topic": "t", "qualifiedName": "kafka://dataops@dataops_net/t"},
        },
    ]
    assert validate_entities(entities)


def test_kafka_topic_with_uri_ok() -> None:
    entities = [
        {
            "typeName": "kafka_topic",
            "attributes": {
                "name": "x",
                "topic": "t",
                "qualifiedName": "kafka://dataops@dataops_net/t",
                "uri": "kafka://kafka:9092/t",
            },
        },
    ]
    assert not validate_entities(entities)


def test_hive_table_requires_db_reference() -> None:
    entities = [
        {
            "typeName": "hive_table",
            "attributes": {"name": "t", "qualifiedName": "db.public.t@cluster"},
        },
    ]
    assert validate_entities(entities)


def test_hive_table_unknown_db_reference() -> None:
    entities = [
        {
            "typeName": "hive_db",
            "attributes": {"name": "good", "qualifiedName": "good@cluster", "clusterName": "cluster"},
        },
        {
            "typeName": "hive_table",
            "relationshipAttributes": {
                "db": {"typeName": "hive_db", "uniqueAttributes": {"qualifiedName": "missing@cluster"}}
            },
            "attributes": {"name": "t", "qualifiedName": "good.public.t@cluster"},
        },
    ]
    assert validate_entities(entities)


def test_hive_db_table_ok() -> None:
    entities = [
        {
            "typeName": "hive_db",
            "attributes": {"name": "good", "qualifiedName": "good@cluster", "clusterName": "cluster"},
        },
        {
            "typeName": "hive_table",
            "relationshipAttributes": {
                "db": {"typeName": "hive_db", "uniqueAttributes": {"qualifiedName": "good@cluster"}}
            },
            "attributes": {"name": "t", "qualifiedName": "good.public.t@cluster"},
        },
    ]
    assert not validate_entities(entities)