dbt-bigquery/tests/unit/test_dataset.py at aee5c2459d061d602407ad6f60fcf8757e7c9c45 · dbt-labs/dbt-bigquery · 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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from dbt.adapters.bigquery.dataset import add_access_entry_to_dataset, is_access_entry_in_dataset, delete_access_entry_from_dataset
from dbt.adapters.bigquery import BigQueryRelation

from google.cloud.bigquery import Dataset, AccessEntry, DatasetReference


def test_add_access_entry_to_dataset_updates_dataset():
    database = "someDb"
    dataset = "someDataset"
    entity = BigQueryRelation.from_dict(
        {
            "type": None,
            "path": {
                "database": "test-project",
                "schema": "test_schema",
                "identifier": "my_table",
            },
            "quote_policy": {"identifier": False},
        }
    ).to_dict()
    dataset_ref = DatasetReference(project=database, dataset_id=dataset)
    dataset = Dataset(dataset_ref)
    access_entry = AccessEntry(None, "table", entity)
    dataset = add_access_entry_to_dataset(dataset, access_entry)
    assert access_entry in dataset.access_entries


def test_add_access_entry_to_dataset_updates_with_pre_existing_entries():
    database = "someOtherDb"
    dataset = "someOtherDataset"
    entity_2 = BigQueryRelation.from_dict(
        {
            "type": None,
            "path": {
                "database": "test-project",
                "schema": "test_schema",
                "identifier": "some_other_view",
            },
            "quote_policy": {"identifier": False},
        }
    ).to_dict()
    dataset_ref = DatasetReference(project=database, dataset_id=dataset)
    dataset = Dataset(dataset_ref)
    initial_entry = AccessEntry(None, "view", entity_2)
    initial_entry._properties.pop("role")
    dataset.access_entries = [initial_entry]
    access_entry = AccessEntry(None, "view", entity_2)
    dataset = add_access_entry_to_dataset(dataset, access_entry)
    assert len(dataset.access_entries) == 2


def test_is_access_entry_in_dataset_returns_true_if_entry_in_dataset():
    database = "someDb"
    dataset = "someDataset"
    entity = BigQueryRelation.from_dict(
        {
            "type": None,
            "path": {
                "database": "test-project",
                "schema": "test_schema",
                "identifier": "my_table",
            },
            "quote_policy": {"identifier": False},
        }
    ).to_dict()
    dataset_ref = DatasetReference(project=database, dataset_id=dataset)
    dataset = Dataset(dataset_ref)
    access_entry = AccessEntry(None, "table", entity)
    dataset = add_access_entry_to_dataset(dataset, access_entry)
    assert is_access_entry_in_dataset(dataset, access_entry)


def test_is_access_entry_in_dataset_returns_false_if_entry_not_in_dataset():
    database = "someDb"
    dataset = "someDataset"
    entity = BigQueryRelation.from_dict(
        {
            "type": None,
            "path": {
                "database": "test-project",
                "schema": "test_schema",
                "identifier": "my_table",
            },
            "quote_policy": {"identifier": False},
        }
    ).to_dict()
    dataset_ref = DatasetReference(project=database, dataset_id=dataset)
    dataset = Dataset(dataset_ref)
    access_entry = AccessEntry(None, "table", entity)
    assert not is_access_entry_in_dataset(dataset, access_entry)


def test_delete_access_to_dataset_updates_dataset():
    """
    test the removal of views grants to dataset
    """
    database = "someDb"
    dataset = "someDataset"
    entity = BigQueryRelation.from_dict(
        {
            "type": None,
            "path": {
                "database": "test-project",
                "schema": "test_schema",
                "identifier": "my_table",
            },
            "quote_policy": {"identifier": False},
        }
    ).to_dict()
    dataset_ref = DatasetReference(project=database, dataset_id=dataset)
    dataset = Dataset(dataset_ref)
    access_entry = AccessEntry(None, "table", entity)
    dataset = add_access_entry_to_dataset(dataset, access_entry)
    assert is_access_entry_in_dataset(dataset, access_entry)
    dataset = delete_access_entry_from_dataset(dataset, access_entry)
    assert not is_access_entry_in_dataset(dataset, access_entry)