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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
#-------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
#--------------------------------------------------------------------------
from collections import namedtuple
import inspect
import os.path
import zlib
from azure_devtools.scenario_tests import (
ReplayableTest, AzureTestError,
AbstractPreparer, GeneralNameReplacer,
OAuthRequestResponsesFilter, DeploymentNameReplacer,
RequestUrlNormalizer
)
from .config import TEST_SETTING_FILENAME
from . import mgmt_settings_fake as fake_settings
class HttpStatusCode(object):
OK = 200
Created = 201
Accepted = 202
NoContent = 204
NotFound = 404
def get_resource_name(name_prefix, identifier):
# Append a suffix to the name, based on the fully qualified test name
# We use a checksum of the test name so that each test gets different
# resource names, but each test will get the same name on repeat runs,
# which is needed for playback.
# Most resource names have a length limit, so we use a crc32
checksum = zlib.adler32(identifier) & 0xffffffff
name = '{}{}'.format(name_prefix, hex(checksum)[2:]).rstrip('L')
if name.endswith('L'):
name = name[:-1]
return name
def get_qualified_method_name(obj, method_name):
# example of qualified test name:
# test_mgmt_network.test_public_ip_addresses
_, filename = os.path.split(inspect.getsourcefile(type(obj)))
module_name, _ = os.path.splitext(filename)
return '{0}.{1}'.format(module_name, method_name)
class AzureMgmtTestCase(ReplayableTest):
def __init__(self, method_name, config_file=None,
recording_dir=None, recording_name=None,
recording_processors=None, replay_processors=None,
recording_patches=None, replay_patches=None):
self.working_folder = os.path.dirname(__file__)
self.qualified_test_name = get_qualified_method_name(self, method_name)
self._fake_settings, self._real_settings = self._load_settings()
self.region = 'westus'
self.scrubber = GeneralNameReplacer()
config_file = config_file or os.path.join(self.working_folder, TEST_SETTING_FILENAME)
if not os.path.exists(config_file):
config_file = None
super(AzureMgmtTestCase, self).__init__(
method_name,
config_file=config_file,
recording_dir=recording_dir,
recording_name=recording_name or self.qualified_test_name,
recording_processors=recording_processors or self._get_recording_processors(),
replay_processors=replay_processors or self._get_replay_processors(),
recording_patches=recording_patches,
replay_patches=replay_patches,
)
@property
def settings(self):
if self.is_live:
if self._real_settings:
return self._real_settings
else:
raise AzureTestError('Need a mgmt_settings_real.py file to run tests live.')
else:
return self._fake_settings
def _load_settings(self):
try:
from . import mgmt_settings_real as real_settings
return fake_settings, real_settings
except ImportError:
return fake_settings, None
def _get_recording_processors(self):
return [
self.scrubber,
OAuthRequestResponsesFilter(),
# DeploymentNameReplacer(), Not use this one, give me full control on deployment name
RequestUrlNormalizer()
]
def _get_replay_processors(self):
return [
RequestUrlNormalizer()
]
def is_playback(self):
return not self.is_live
def _setup_scrubber(self):
constants_to_scrub = ['SUBSCRIPTION_ID', 'AD_DOMAIN', 'TENANT_ID', 'CLIENT_OID', 'ADLA_JOB_ID']
for key in constants_to_scrub:
if hasattr(self.settings, key) and hasattr(self._fake_settings, key):
self.scrubber.register_name_pair(getattr(self.settings, key),
getattr(self._fake_settings, key))
def setUp(self):
# Every test uses a different resource group name calculated from its
# qualified test name.
#
# When running all tests serially, this allows us to delete
# the resource group in teardown without waiting for the delete to
# complete. The next test in line will use a different resource group,
# so it won't have any trouble creating its resource group even if the
# previous test resource group hasn't finished deleting.
#
# When running tests individually, if you try to run the same test
# multiple times in a row, it's possible that the delete in the previous
# teardown hasn't completed yet (because we don't wait), and that
# would make resource group creation fail.
# To avoid that, we also delete the resource group in the
# setup, and we wait for that delete to complete.
self._setup_scrubber()
super(AzureMgmtTestCase, self).setUp()
def tearDown(self):
return super(AzureMgmtTestCase, self).tearDown()
def create_basic_client(self, client_class, **kwargs):
# Whatever the client, if credentials is None, fail
with self.assertRaises(ValueError):
client = client_class(
credentials=None,
**kwargs
)
# Real client creation
client = client_class(
credentials=self.settings.get_credentials(),
**kwargs
)
if self.is_playback():
client.config.long_running_operation_timeout = 0
client.config.enable_http_logger = True
return client
def create_mgmt_client(self, client_class, **kwargs):
# Whatever the client, if subscription_id is None, fail
with self.assertRaises(ValueError):
self.create_basic_client(
client_class,
subscription_id=None,
**kwargs
)
return self.create_basic_client(
client_class,
subscription_id=self.settings.SUBSCRIPTION_ID,
**kwargs
)
def create_random_name(self, name):
return get_resource_name(name, self.qualified_test_name.encode())
def get_resource_name(self, name):
"""Alias to create_random_name for back compatibility."""
return self.create_random_name(name)
def get_preparer_resource_name(self, prefix):
"""Random name generation for use by preparers.
If prefix is a blank string, use the fully qualified test name instead.
This is what legacy tests do for resource groups."""
return self.get_resource_name(prefix or self.qualified_test_name.replace('.', '_'))
class AzureMgmtPreparer(AbstractPreparer):
def __init__(self, name_prefix, random_name_length,
disable_recording=True,
playback_fake_resource=None,
client_kwargs=None):
super(AzureMgmtPreparer, self).__init__(name_prefix, random_name_length,
disable_recording=disable_recording)
self.client = None
self.resource = playback_fake_resource
self.client_kwargs = client_kwargs or {}
@property
def is_live(self):
return self.test_class_instance.is_live
def create_random_name(self):
return self.test_class_instance.get_preparer_resource_name(self.name_prefix)
@property
def moniker(self):
"""Override moniker generation for backwards compatibility.
azure-devtools preparers, by default, generate "monikers" which replace
resource names in request URIs by tacking on a resource count to
name_prefix. By contrast, SDK tests used the fully qualified (module + method)
test name and the hashing process in get_resource_name.
This property override implements the SDK test name generation so that
the URIs don't change and tests don't need to be re-recorded.
"""
if not self.resource_moniker:
self.resource_moniker = self.random_name
return self.resource_moniker
def create_mgmt_client(self, client_class):
return client_class(
credentials=self.test_class_instance.settings.get_credentials(),
subscription_id=self.test_class_instance.settings.SUBSCRIPTION_ID,
**self.client_kwargs
)
|
