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
|
# coding: utf-8
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
import pytest
import unittest
import os
from azure.mgmt.loadtesting import LoadTestMgmtClient
from azure.mgmt.loadtesting.models import (
LoadTestResource,
LoadTestResourcePatchRequestBody,
ManagedServiceIdentity,
UserAssignedIdentity,
EncryptionProperties,
EncryptionPropertiesIdentity,
)
from azure.mgmt.msi import ManagedServiceIdentityClient
from azure.mgmt.keyvault import KeyVaultManagementClient
from azure.keyvault.keys import KeyClient
from devtools_testutils import (
AzureMgmtRecordedTestCase,
recorded_by_proxy,
ResourceGroupPreparer,
)
from mgmt_test_helper import create_key, create_key_vault, create_managed_identity
DEFAULT_SANITIZER = "00000000-0000-0000-0000-000000000000"
@pytest.mark.live_test_only
class TestResourceOperations(AzureMgmtRecordedTestCase):
AZURE_LOCATION = "westus2"
def setup_method(self, method):
self.loadtestservice_client = self.create_mgmt_client(LoadTestMgmtClient)
self.msi_client = self.create_mgmt_client(ManagedServiceIdentityClient)
self.akv_client = self.create_mgmt_client(KeyVaultManagementClient)
self.credential = self.get_credential(KeyClient)
self.tenant_id = (
os.environ.get("AZURE_TENANT_ID", None)
if self.is_live
else DEFAULT_SANITIZER
)
self.object_id = (
os.environ.get("CLIENT_OID", None) if self.is_live else DEFAULT_SANITIZER
)
@ResourceGroupPreparer(name_prefix="altpysdk")
@recorded_by_proxy
def test_load_test_resource_operations(self, resource_group):
self.msi1 = create_managed_identity(
self.msi_client,
resource_group.name + "mi1",
resource_group.name,
self.AZURE_LOCATION,
)
self.msi2 = create_managed_identity(
self.msi_client,
resource_group.name + "mi2",
resource_group.name,
self.AZURE_LOCATION,
)
self.akv = create_key_vault(
self.akv_client,
resource_group.name + "akv",
resource_group.name,
self.AZURE_LOCATION,
self.msi1,
self.tenant_id,
self.object_id,
)
self.key = create_key(self.akv, self.credential, resource_group.name + "key")
# Create a load test resource
loadtestresource_create_payload = LoadTestResource(
location=self.AZURE_LOCATION,
identity=ManagedServiceIdentity(
type="SystemAssigned,UserAssigned",
user_assigned_identities={
self.msi1.id: UserAssignedIdentity(),
self.msi2.id: UserAssignedIdentity(),
},
),
encryption=EncryptionProperties(
identity=EncryptionPropertiesIdentity(
type="UserAssigned", resource_id=self.msi1.id
),
key_url=self.key.id,
),
)
# Create a load test resource create begin - returns a poller
loadtest_resource_poller = (
self.loadtestservice_client.load_tests.begin_create_or_update(
resource_group.name,
resource_group.name + "resource",
loadtestresource_create_payload,
)
)
# Get the result of the poller
loadtest_resource = loadtest_resource_poller.result()
# Assert the result of the poller
assert loadtest_resource
assert loadtest_resource.id
assert loadtest_resource.name == resource_group.name + "resource"
assert loadtest_resource.location == self.AZURE_LOCATION
assert loadtest_resource.identity
assert loadtest_resource.identity.type == "SystemAssigned, UserAssigned"
assert len(loadtest_resource.identity.user_assigned_identities) == 2
assert loadtest_resource.encryption
assert loadtest_resource.encryption.key_url == self.key.id
assert loadtest_resource.encryption.identity
assert loadtest_resource.encryption.identity.type == "UserAssigned"
assert loadtest_resource.encryption.identity.resource_id == self.msi1.id
# # Get the load test resource
loadtest_resource_get = self.loadtestservice_client.load_tests.get(
resource_group.name, resource_group.name + "resource"
)
# # Assert the result of the get operation
assert loadtest_resource_get
assert loadtest_resource_get.id
assert loadtest_resource_get.name == resource_group.name + "resource"
assert loadtest_resource_get.location == self.AZURE_LOCATION
assert loadtest_resource_get.identity
assert loadtest_resource_get.identity.type == "SystemAssigned, UserAssigned"
assert len(loadtest_resource_get.identity.user_assigned_identities) == 2
assert loadtest_resource_get.encryption
assert loadtest_resource_get.encryption.key_url == self.key.id
assert loadtest_resource_get.encryption.identity
assert loadtest_resource_get.encryption.identity.type == "UserAssigned"
assert loadtest_resource_get.encryption.identity.resource_id == self.msi1.id
# Update the load test resource
loadtestresourcePatchdata = LoadTestResourcePatchRequestBody(
identity=ManagedServiceIdentity(
type="SystemAssigned,UserAssigned",
user_assigned_identities={self.msi2.id: None},
),
)
# load test resource update begin - returns a poller
loadtest_resource_patch_poller = (
self.loadtestservice_client.load_tests.begin_update(
resource_group.name,
resource_group.name + "resource",
loadtestresourcePatchdata,
)
)
# Get the result of the poller
loadtest_resource_patch_response = loadtest_resource_patch_poller.result()
# Assert the result of the poller
assert loadtest_resource_patch_response
assert loadtest_resource_patch_response.id
assert loadtest_resource_patch_response.name == resource_group.name + "resource"
assert loadtest_resource_patch_response.location == self.AZURE_LOCATION
assert loadtest_resource_patch_response.identity
assert (
loadtest_resource_patch_response.identity.type
== "SystemAssigned, UserAssigned"
)
assert (
len(loadtest_resource_patch_response.identity.user_assigned_identities) == 1
)
assert loadtest_resource_patch_response.encryption
assert loadtest_resource_patch_response.encryption.key_url == self.key.id
assert loadtest_resource_patch_response.encryption.identity
assert (
loadtest_resource_patch_response.encryption.identity.type == "UserAssigned"
)
assert (
loadtest_resource_patch_response.encryption.identity.resource_id
== self.msi1.id
)
# Delete the load test resource - returns a poller
loadtest_resource_delete_poller = (
self.loadtestservice_client.load_tests.begin_delete(
resource_group.name, resource_group.name + "resource"
)
)
# ------------------------------------------------------------------------------
if __name__ == "__main__":
unittest.main()
|
