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
|
# -------------------------------------------------------------------------
# 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 re
import six
import os.path
import zlib
from azure_devtools.scenario_tests import (
ReplayableTest,
AzureTestError,
AbstractPreparer,
GeneralNameReplacer,
OAuthRequestResponsesFilter,
DeploymentNameReplacer,
RequestUrlNormalizer,
)
from azure_devtools.scenario_tests.utilities import is_text_payload
from .azure_testcase import AzureTestCase
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(AzureTestCase):
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,
**kwargs
):
self.region = "westus"
self.re_replacer = RENameReplacer()
super(AzureMgmtTestCase, self).__init__(
method_name,
config_file=config_file,
recording_dir=recording_dir,
recording_name=recording_name,
recording_processors=recording_processors,
replay_processors=replay_processors,
recording_patches=recording_patches,
replay_patches=replay_patches,
**kwargs
)
self.recording_processors.append(self.re_replacer)
def _setup_scrubber(self):
constants_to_scrub = ["SUBSCRIPTION_ID", "TENANT_ID"]
for key in constants_to_scrub:
key_value = self.get_settings_value(key)
if key_value and hasattr(self._fake_settings, key):
self.scrubber.register_name_pair(key_value, 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_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)
subscription_id = None
if self.is_live:
subscription_id = os.environ.get("AZURE_SUBSCRIPTION_ID", None)
if not subscription_id:
subscription_id = self.settings.SUBSCRIPTION_ID
return self.create_basic_client(client_class, subscription_id=subscription_id, **kwargs)
class AzureMgmtPreparer(AbstractPreparer):
def __init__(
self,
name_prefix,
random_name_length,
disable_recording=True,
playback_fake_resource=None,
client_kwargs=None,
random_name_enabled=False,
):
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 {}
self.random_name_enabled = random_name_enabled
@property
def is_live(self):
return self.test_class_instance.is_live
def create_random_name(self):
if self.random_name_enabled:
return super(AzureMgmtPreparer, self).create_random_name()
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, **kwargs):
return self.test_class_instance.create_mgmt_client(client_class, **kwargs)
class RENameReplacer(GeneralNameReplacer):
def __init__(self):
super(RENameReplacer, self).__init__()
self.patterns = []
def register_pattern_pair(self, expr, new):
self.patterns.append((expr, new))
def process_request(self, request):
request = super(RENameReplacer, self).process_request(request)
for expr, new in self.patterns:
if is_text_payload(request) and request.body:
if isinstance(request.body, dict):
continue
body = six.ensure_str(request.body)
for old in re.findall(expr, body):
request.body = body.replace(old, new)
return request
def process_response(self, response):
response = super(RENameReplacer, self).process_response(response)
for expr, new in self.patterns:
if is_text_payload(response) and response["body"]["string"]:
if isinstance(response["body"]["string"], bytes):
body = response["body"]["string"].decode("utf8", "backslashreplace")
else:
body = response["body"]["string"]
for old in re.findall(expr, body):
body = body.replace(old, new)
if isinstance(response["body"]["string"], bytes):
response["body"]["string"] = body.encode("utf8", "backslashreplace")
else:
response["body"]["string"] = body
return response
|
