forked from googleapis/python-spanner
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_instance_api.py
More file actions
142 lines (109 loc) · 4.41 KB
/
test_instance_api.py
File metadata and controls
142 lines (109 loc) · 4.41 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
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
# Copyright 2021 Google LLC All rights reserved.
#
# 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
#
# http://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.
import pytest
from test_utils import retry
from . import _helpers
@pytest.fixture(scope="function")
def instances_to_delete():
to_delete = []
yield to_delete
for instance in to_delete:
_helpers.scrub_instance_ignore_not_found(instance)
def test_list_instances(
no_create_instance,
spanner_client,
existing_instances,
shared_instance,
):
instances = list(spanner_client.list_instances())
for instance in instances:
assert instance in existing_instances or instance is shared_instance
def test_reload_instance(spanner_client, shared_instance_id, shared_instance):
# Use same arguments as shared_instance_id so we can use 'reload()'
# on a fresh instance.
instance = spanner_client.instance(shared_instance_id)
# Unset metadata before reloading.
instance.display_name = None
def _expected_display_name(instance):
return instance.display_name == shared_instance.display_name
retry_until = retry.RetryInstanceState(_expected_display_name)
retry_until(instance.reload)()
assert instance.display_name == shared_instance.display_name
def test_create_instance(
if_create_instance,
spanner_client,
instance_config,
instances_to_delete,
instance_operation_timeout,
):
alt_instance_id = _helpers.unique_id("new")
instance = spanner_client.instance(alt_instance_id, instance_config.name)
operation = instance.create()
# Make sure this instance gets deleted after the test case.
instances_to_delete.append(instance)
# We want to make sure the operation completes.
operation.result(instance_operation_timeout) # raises on failure / timeout.
# Create a new instance instance and make sure it is the same.
instance_alt = spanner_client.instance(alt_instance_id, instance_config.name)
instance_alt.reload()
assert instance == instance_alt
instance.display_name == instance_alt.display_name
def test_create_instance_with_processing_units(
not_emulator,
if_create_instance,
spanner_client,
instance_config,
instances_to_delete,
instance_operation_timeout,
):
alt_instance_id = _helpers.unique_id("wpn")
processing_units = 5000
instance = spanner_client.instance(
instance_id=alt_instance_id,
configuration_name=instance_config.name,
processing_units=processing_units,
)
operation = instance.create()
# Make sure this instance gets deleted after the test case.
instances_to_delete.append(instance)
# We want to make sure the operation completes.
operation.result(instance_operation_timeout) # raises on failure / timeout.
# Create a new instance instance and make sure it is the same.
instance_alt = spanner_client.instance(alt_instance_id, instance_config.name)
instance_alt.reload()
assert instance == instance_alt
assert instance.display_name == instance_alt.display_name
assert instance.processing_units == instance_alt.processing_units
def test_update_instance(
not_emulator,
spanner_client,
shared_instance,
shared_instance_id,
instance_operation_timeout,
):
old_display_name = shared_instance.display_name
new_display_name = "Foo Bar Baz"
shared_instance.display_name = new_display_name
operation = shared_instance.update()
# We want to make sure the operation completes.
operation.result(instance_operation_timeout) # raises on failure / timeout.
# Create a new instance instance and reload it.
instance_alt = spanner_client.instance(shared_instance_id, None)
assert instance_alt.display_name != new_display_name
instance_alt.reload()
assert instance_alt.display_name == new_display_name
# Make sure to put the instance back the way it was for the
# other test cases.
shared_instance.display_name = old_display_name
shared_instance.update()