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
|
import time
from azure.mgmt.resource import ResourceManagementClient
from devtools_testutils import AzureMgmtRecordedTestCase, recorded_by_proxy
import azure.mgmt.netapp.models
from azure.mgmt.netapp.models import NetAppAccount, NetAppAccountPatch
import setup
def create_account(client, rg, acc_name, location=setup.LOCATION, tags=None, active_directories=None):
account_body = NetAppAccount(location=location, tags=tags, active_directories=active_directories)
print("Creating NetApp Account {0}".format(acc_name))
account = client.accounts.begin_create_or_update(rg, acc_name, account_body).result()
wait_for_account(client, rg, acc_name)
print("\tDone creating NetApp Account {0}".format(acc_name))
return account
def wait_for_account(client, rg, acc_name):
# a work around for the async nature of certain ARM processes
retry = 0
while retry < 60:
account = client.accounts.get(rg, acc_name)
if account.provisioning_state == "Succeeded":
break
retry += 1
if setup.LIVE:
time.sleep(3)
if retry == 60:
raise Exception("Timeout when waiting for account")
def wait_for_no_account(client, rg, acc_name):
# a workaround for the async nature of certain ARM processes
retry = 0
while retry < 60:
try:
client.accounts.get(rg, acc_name)
except:
# not found is an exception case (status code 200 expected)
# and is actually what we are waiting for
break
retry += 1
if setup.LIVE:
time.sleep(3)
if retry == 60:
raise Exception("Timeout when waiting for no account")
def delete_account(client, rg, acc_name):
print("Deleting NetApp Account {0}".format(acc_name))
retry = 0
while retry < 5:
try:
client.accounts.begin_delete(rg, acc_name).wait()
break
except Exception as e:
print("failed to delete account. Retry number: {0}".format(retry))
print(e)
retry += 1
if setup.LIVE:
time.sleep(10)
if retry == 5:
raise Exception("Timeout when deleting account")
wait_for_no_account(client, rg, acc_name)
print("\tDone deleting account {0}".format(acc_name))
class TestNetAppAccount(AzureMgmtRecordedTestCase):
def setup_method(self, method):
self.client = self.create_mgmt_client(azure.mgmt.netapp.NetAppManagementClient)
print("Live status: {0}".format(self.is_live))
if self.is_live:
setup.LIVE = True
# Before tests are run live a resource group needs to be created
# Note that when tests are run in live mode it is best to run one test at a time.
@recorded_by_proxy
def test_create_delete_account(self):
print("Starting test_create_delete_account")
account_name = self.get_resource_name(setup.TEST_ACC_1 + "-")
account_list = self.client.accounts.list(setup.TEST_RG)
account_list_lenght = len(list(account_list))
print("create_delete_account {0}".format(account_name))
account = create_account(self.client, setup.TEST_RG, account_name)
assert account.name == account_name
account_list = self.client.accounts.list(setup.TEST_RG)
assert len(list(account_list)) == account_list_lenght + 1
delete_account(self.client, setup.TEST_RG, account_name)
account_list = self.client.accounts.list(setup.TEST_RG)
assert len(list(account_list)) == account_list_lenght
print("Finished with test_create_delete_account")
@recorded_by_proxy
def test_list_accounts(self):
print("Starting test_list_accounts")
account_name1 = self.get_resource_name(setup.TEST_ACC_1 + "-")
account_name2 = self.get_resource_name(setup.TEST_ACC_2 + "-")
account_list = self.client.accounts.list(setup.TEST_RG)
account_list_lenght = len(list(account_list))
create_account(self.client, setup.TEST_RG, account_name1)
create_account(self.client, setup.TEST_RG, account_name2)
accounts = [account_name1, account_name2]
account_list = self.client.accounts.list(setup.TEST_RG)
assert len(list(account_list)) == account_list_lenght + 2
idx = 0
for account in account_list:
assert account.name == accounts[idx]
idx += 1
delete_account(self.client, setup.TEST_RG, account_name1)
delete_account(self.client, setup.TEST_RG, account_name2)
print("Finished with test_list_accounts")
@recorded_by_proxy
def test_get_account_by_name(self):
print("Starting test_get_account_by_name")
account_name1 = self.get_resource_name(setup.TEST_ACC_1 + "-")
create_account(self.client, setup.TEST_RG, account_name1)
account = self.client.accounts.get(setup.TEST_RG, account_name1)
assert account.name == account_name1
delete_account(self.client, setup.TEST_RG, account_name1)
print("Finished with test_get_account_by_name")
@recorded_by_proxy
def test_patch_account(self):
print("Starting test_patch_account")
account_name1 = self.get_resource_name(setup.TEST_ACC_1 + "-")
create_account(self.client, setup.TEST_RG, account_name1)
tag = {"Tag1": "Value2"}
netapp_account_patch = NetAppAccountPatch(tags=tag)
account = self.client.accounts.begin_update(setup.TEST_RG, account_name1, netapp_account_patch).result()
assert account.tags["Tag1"] == "Value2"
delete_account(self.client, setup.TEST_RG, account_name1)
print("Finished with test_patch_account")
|
