This repository was archived by the owner on Mar 6, 2026. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 324
Expand file tree
/
Copy pathtest_delete_dataset.py
More file actions
79 lines (63 loc) · 2.73 KB
/
test_delete_dataset.py
File metadata and controls
79 lines (63 loc) · 2.73 KB
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
# Copyright 2021 Google LLC
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# https://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from .helpers import make_connection, make_client, dataset_polymorphic
import google.api_core.exceptions
from google.cloud.bigquery.retry import DEFAULT_TIMEOUT
import pytest
@dataset_polymorphic
def test_delete_dataset(make_dataset, get_reference, client, PROJECT, DS_ID):
dataset = make_dataset(PROJECT, DS_ID)
PATH = "projects/%s/datasets/%s" % (PROJECT, DS_ID)
conn = client._connection = make_connection({})
client.delete_dataset(dataset, timeout=7.5)
conn.api_request.assert_called_with(
method="DELETE", path="/%s" % PATH, query_params={}, timeout=7.5
)
@dataset_polymorphic
def test_delete_dataset_delete_contents(
make_dataset, get_reference, client, PROJECT, DS_ID
):
PATH = "projects/%s/datasets/%s" % (PROJECT, DS_ID)
conn = client._connection = make_connection({})
dataset = make_dataset(PROJECT, DS_ID)
client.delete_dataset(dataset, delete_contents=True)
conn.api_request.assert_called_with(
method="DELETE",
path="/%s" % PATH,
query_params={"deleteContents": "true"},
timeout=DEFAULT_TIMEOUT,
)
def test_delete_dataset_wrong_type(client):
with pytest.raises(TypeError):
client.delete_dataset(42)
def test_delete_dataset_w_not_found_ok_false(PROJECT, DS_ID):
path = "/projects/{}/datasets/{}".format(PROJECT, DS_ID)
http = object()
client = make_client(_http=http)
conn = client._connection = make_connection(
google.api_core.exceptions.NotFound("dataset not found")
)
with pytest.raises(google.api_core.exceptions.NotFound):
client.delete_dataset(DS_ID)
conn.api_request.assert_called_with(
method="DELETE", path=path, query_params={}, timeout=DEFAULT_TIMEOUT
)
def test_delete_dataset_w_not_found_ok_true(PROJECT, DS_ID):
path = "/projects/{}/datasets/{}".format(PROJECT, DS_ID)
http = object()
client = make_client(_http=http)
conn = client._connection = make_connection(
google.api_core.exceptions.NotFound("dataset not found")
)
client.delete_dataset(DS_ID, not_found_ok=True)
conn.api_request.assert_called_with(
method="DELETE", path=path, query_params={}, timeout=DEFAULT_TIMEOUT
)