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
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
|
# 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 unittest
import time
import pytest
from azure.data.tables._models import TableAnalyticsLogging, Metrics, RetentionPolicy, CorsRule
from msrest.exceptions import ValidationError # TODO This should be an azure-core error.
from devtools_testutils import ResourceGroupPreparer, StorageAccountPreparer
from azure.core.exceptions import HttpResponseError
from azure.data.tables import TableServiceClient
from _shared.testcase import TableTestCase, RERUNS_DELAY
from _shared.cosmos_testcase import CachedCosmosAccountPreparer
from devtools_testutils import CachedResourceGroupPreparer
# ------------------------------------------------------------------------------
class TableServicePropertiesTest(TableTestCase):
# --Helpers-----------------------------------------------------------------
def _assert_properties_default(self, prop):
assert prop is not None
self._assert_logging_equal(prop['analytics_logging'], TableAnalyticsLogging())
self._assert_metrics_equal(prop['hour_metrics'], Metrics())
self._assert_metrics_equal(prop['minute_metrics'], Metrics())
self._assert_cors_equal(prop['cors'], list())
def _assert_logging_equal(self, log1, log2):
if log1 is None or log2 is None:
assert log1 == log2
return
assert log1.version == log2.version
assert log1.read == log2.read
assert log1.write == log2.write
assert log1.delete == log2.delete
self._assert_retention_equal(log1.retention_policy, log2.retention_policy)
def _assert_delete_retention_policy_equal(self, policy1, policy2):
if policy1 is None or policy2 is None:
assert policy1 == policy2
return
assert policy1.enabled == policy2.enabled
assert policy1.days == policy2.days
def _assert_static_website_equal(self, prop1, prop2):
if prop1 is None or prop2 is None:
assert prop1 == prop2
return
assert prop1.enabled == prop2.enabled
assert prop1.index_document == prop2.index_document
assert prop1.error_document404_path == prop2.error_document404_path
def _assert_delete_retention_policy_not_equal(self, policy1, policy2):
if policy1 is None or policy2 is None:
assert policy1 != policy2
return
assert not (policy1.enabled == policy2.enabled and policy1.days == policy2.days)
def _assert_metrics_equal(self, metrics1, metrics2):
if metrics1 is None or metrics2 is None:
assert metrics1 == metrics2
return
assert metrics1.version == metrics2.version
assert metrics1.enabled == metrics2.enabled
assert metrics1.include_apis == metrics2.include_apis
self._assert_retention_equal(metrics1.retention_policy, metrics2.retention_policy)
def _assert_cors_equal(self, cors1, cors2):
if cors1 is None or cors2 is None:
assert cors1 == cors2
return
assert len(cors1) == len(cors2)
for i in range(0, len(cors1)):
rule1 = cors1[i]
rule2 = cors2[i]
assert len(rule1.allowed_origins) == len(rule2.allowed_origins)
assert len(rule1.allowed_methods) == len(rule2.allowed_methods)
assert rule1.max_age_in_seconds == rule2.max_age_in_seconds
assert len(rule1.exposed_headers) == len(rule2.exposed_headers)
assert len(rule1.allowed_headers) == len(rule2.allowed_headers)
def _assert_retention_equal(self, ret1, ret2):
assert ret1.enabled == ret2.enabled
assert ret1.days == ret2.days
# --Test cases per service ---------------------------------------
@pytest.mark.skip("Cosmos Tables does not yet support service properties")
@CachedResourceGroupPreparer(name_prefix="tablestest")
@CachedCosmosAccountPreparer(name_prefix="tablestest")
def test_table_service_properties(self, resource_group, location, storage_account, storage_account_key):
# Arrange
url = self.account_url(storage_account, "cosmos")
tsc = TableServiceClient(url, storage_account_key)
# Act
resp = tsc.set_service_properties(
analytics_logging=TableAnalyticsLogging(),
hour_metrics=Metrics(),
minute_metrics=Metrics(),
cors=list())
# Assert
assert resp is None
self._assert_properties_default(tsc.get_service_properties())
if self.is_live:
sleep(SLEEP_DELAY)
# --Test cases per feature ---------------------------------------
@pytest.mark.skip("Cosmos Tables does not yet support service properties")
@CachedResourceGroupPreparer(name_prefix="tablestest")
@CachedCosmosAccountPreparer(name_prefix="tablestest")
def test_set_logging(self, resource_group, location, storage_account, storage_account_key):
# Arrange
url = self.account_url(storage_account, "cosmos")
tsc = TableServiceClient(url, storage_account_key)
logging = TableAnalyticsLogging(read=True, write=True, delete=True, retention_policy=RetentionPolicy(enabled=True, days=5))
# Act
tsc.set_service_properties(analytics_logging=logging)
# Assert
received_props = tsc.get_service_properties()
self._assert_logging_equal(received_props['analytics_logging'], logging)
if self.is_live:
time.sleep(30)
@pytest.mark.skip("Cosmos Tables does not yet support service properties")
@CachedResourceGroupPreparer(name_prefix="tablestest")
@CachedCosmosAccountPreparer(name_prefix="tablestest")
def test_set_hour_metrics(self, resource_group, location, storage_account, storage_account_key):
# Arrange
url = self.account_url(storage_account, "cosmos")
tsc = TableServiceClient(url, storage_account_key)
hour_metrics = Metrics(enabled=True, include_apis=True, retention_policy=RetentionPolicy(enabled=True, days=5))
# Act
tsc.set_service_properties(hour_metrics=hour_metrics)
# Assert
received_props = tsc.get_service_properties()
self._assert_metrics_equal(received_props['hour_metrics'], hour_metrics)
if self.is_live:
sleep(SLEEP_DELAY)
@pytest.mark.skip("Cosmos Tables does not yet support service properties")
@CachedResourceGroupPreparer(name_prefix="tablestest")
@CachedCosmosAccountPreparer(name_prefix="tablestest")
def test_set_minute_metrics(self, resource_group, location, storage_account, storage_account_key):
# Arrange
url = self.account_url(storage_account, "cosmos")
tsc = TableServiceClient(url, storage_account_key)
minute_metrics = Metrics(enabled=True, include_apis=True,
retention_policy=RetentionPolicy(enabled=True, days=5))
# Act
tsc.set_service_properties(minute_metrics=minute_metrics)
# Assert
received_props = tsc.get_service_properties()
self._assert_metrics_equal(received_props['minute_metrics'], minute_metrics)
if self.is_live:
sleep(SLEEP_DELAY)
@pytest.mark.skip("Cosmos Tables does not yet support service properties")
@CachedResourceGroupPreparer(name_prefix="tablestest")
@CachedCosmosAccountPreparer(name_prefix="tablestest")
def test_set_cors(self, resource_group, location, storage_account, storage_account_key):
# Arrange
url = self.account_url(storage_account, "cosmos")
tsc = TableServiceClient(url, storage_account_key)
cors_rule1 = CorsRule(['www.xyz.com'], ['GET'])
allowed_origins = ['www.xyz.com', "www.ab.com", "www.bc.com"]
allowed_methods = ['GET', 'PUT']
max_age_in_seconds = 500
exposed_headers = ["x-ms-meta-data*", "x-ms-meta-source*", "x-ms-meta-abc", "x-ms-meta-bcd"]
allowed_headers = ["x-ms-meta-data*", "x-ms-meta-target*", "x-ms-meta-xyz", "x-ms-meta-foo"]
cors_rule2 = CorsRule(
allowed_origins,
allowed_methods,
max_age_in_seconds=max_age_in_seconds,
exposed_headers=exposed_headers,
allowed_headers=allowed_headers)
cors = [cors_rule1, cors_rule2]
# Act
tsc.set_service_properties(cors=cors)
# Assert
received_props = tsc.get_service_properties()
self._assert_cors_equal(received_props['cors'], cors)
if self.is_live:
sleep(SLEEP_DELAY)
# --Test cases for errors ---------------------------------------
@pytest.mark.skip("Cosmos Tables does not yet support service properties")
@CachedResourceGroupPreparer(name_prefix="tablestest")
@CachedCosmosAccountPreparer(name_prefix="tablestest")
def test_retention_no_days(self, resource_group, location, storage_account, storage_account_key):
# Assert
pytest.raises(ValueError,
RetentionPolicy,
True, None)
if self.is_live:
sleep(SLEEP_DELAY)
@pytest.mark.skip("Cosmos Tables does not yet support service properties")
@CachedResourceGroupPreparer(name_prefix="tablestest")
@CachedCosmosAccountPreparer(name_prefix="tablestest")
def test_too_many_cors_rules(self, resource_group, location, storage_account, storage_account_key):
# Arrange
tsc = TableServiceClient(self.account_url(storage_account, "cosmos"), storage_account_key)
cors = []
for i in range(0, 6):
cors.append(CorsRule(['www.xyz.com'], ['GET']))
# Assert
pytest.raises(HttpResponseError,
tsc.set_service_properties, None, None, None, cors)
if self.is_live:
sleep(SLEEP_DELAY)
@pytest.mark.skip("Cosmos Tables does not yet support service properties")
@CachedResourceGroupPreparer(name_prefix="tablestest")
@CachedCosmosAccountPreparer(name_prefix="tablestest")
def test_retention_too_long(self, resource_group, location, storage_account, storage_account_key):
# Arrange
tsc = TableServiceClient(self.account_url(storage_account, "cosmos"), storage_account_key)
minute_metrics = Metrics(enabled=True, include_apis=True,
retention_policy=RetentionPolicy(enabled=True, days=366))
# Assert
pytest.raises(HttpResponseError,
tsc.set_service_properties,
None, None, minute_metrics)
if self.is_live:
sleep(SLEEP_DELAY)
# ------------------------------------------------------------------------------
if __name__ == '__main__':
unittest.main()
|
