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
|
#-------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
#--------------------------------------------------------------------------
import json
import os.path
import time
import azure.mgmt.resource
from azure.common.exceptions import (
CloudError
)
from testutils.common_recordingtestcase import (
RecordingTestCase,
TestMode,
)
import tests.mgmt_settings_fake as fake_settings
should_log = os.getenv('SDK_TESTS_LOG', '0')
if should_log.lower() == 'true' or should_log == '1':
import logging
logger = logging.getLogger('msrest')
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())
class HttpStatusCode(object):
OK = 200
Created = 201
Accepted = 202
NoContent = 204
NotFound = 404
class AzureMgmtTestCase(RecordingTestCase):
def setUp(self):
self.working_folder = os.path.dirname(__file__)
super(AzureMgmtTestCase, self).setUp()
self.fake_settings = fake_settings
if TestMode.is_playback(self.test_mode):
self.settings = self.fake_settings
else:
import tests.mgmt_settings_real as real_settings
self.settings = real_settings
self.resource_client = self.create_mgmt_client(
azure.mgmt.resource.ResourceManagementClient
)
# 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.group_name = self.get_resource_name(
self.qualified_test_name.replace('.', '_')
)
self.region = 'westus'
if not self.is_playback():
self.delete_resource_group(wait_timeout=600)
def tearDown(self):
if not self.is_playback():
self.delete_resource_group(wait_timeout=None)
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
)
# Whatever the client, if accept_language is not str, fail
with self.assertRaises(TypeError):
client = client_class(
credentials=self.settings.get_credentials(),
accept_language=42,
**kwargs
)
# Real client creation
client = client_class(
credentials=self.settings.get_credentials(),
**kwargs
)
if self.is_playback():
client.config.long_running_operation_timeout = 0
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
)
# Whatever the client, if subscription_id is not a string, fail
with self.assertRaises(TypeError):
self.create_basic_client(
client_class,
subscription_id=42,
**kwargs
)
return self.create_basic_client(
client_class,
subscription_id=self.settings.SUBSCRIPTION_ID,
**kwargs
)
def _scrub(self, val):
val = super(AzureMgmtTestCase, self)._scrub(val)
real_to_fake_dict = {
self.settings.SUBSCRIPTION_ID: self.fake_settings.SUBSCRIPTION_ID,
self.settings.AD_DOMAIN: self.fake_settings.AD_DOMAIN
}
val = self._scrub_using_dict(val, real_to_fake_dict)
return val
def create_resource_group(self):
result = self.resource_client.resource_groups.create_or_update(
self.group_name,
{
'location': self.region
}
)
def delete_resource_group(self, wait_timeout):
azure_poller = self.resource_client.resource_groups.delete(self.group_name)
if wait_timeout:
try:
azure_poller.wait(wait_timeout)
if azure_poller.done():
return
self.assertTrue(False, 'Timed out waiting for resource group to be deleted.')
except CloudError:
pass
|
