-
Notifications
You must be signed in to change notification settings - Fork 52
/
service_test.py
238 lines (201 loc) · 11.3 KB
/
service_test.py
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
# -----------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# -----------------------------------------------------------------------------
"""Custom service command tests"""
import unittest
import sfctl.custom_service as sf_c
from knack.util import CLIError
#pylint: disable=invalid-name
class ServiceTests(unittest.TestCase): #pylint: disable=too-many-public-methods
"""Service tests"""
def parse_none_correlation_desc_test(self):
"""Parse None correlation description returns None"""
self.assertIs(sf_c.correlation_desc(None, None), None)
def parse_partial_correlation_desc_test(self):
"""Parse partial correlation description raises error"""
with self.assertRaises(CLIError):
sf_c.correlation_desc('test_svc', None)
def parse_complete_correlation_desc_test(self):
"""Parse a single correlation description"""
res = sf_c.correlation_desc('test', 'Affinity')
self.assertEqual(res.service_name, 'test')
self.assertEqual(res.scheme, 'Affinity')
def parse_empty_load_metrics_test(self):
"""Parse empty load metrics returns None"""
self.assertIsNone(sf_c.parse_load_metrics(''))
def parse_none_load_metrics_test(self):
"""Parse none load metrics returns None"""
self.assertIsNone(sf_c.parse_load_metrics(None))
def parse_incomplete_load_metrics_test(self):
"""Parse single incomplete load metrics definition"""
res = sf_c.parse_load_metrics([{'name': 'test_metric',
'default_load': 10}])
self.assertEqual(len(res), 1)
res = res[0]
self.assertEqual(res.name, 'test_metric')
self.assertIsNone(res.weight)
self.assertIsNone(res.primary_default_load)
self.assertIsNone(res.secondary_default_load)
self.assertEqual(res.default_load, 10)
def parse_invalid_placement_policy_type_test(self):
"""Parsing invalid placement policy type raises error"""
with self.assertRaises(CLIError):
sf_c.parse_placement_policies([{'type': 'test',
'domain_name': 'test'}])
def parse_missing_placement_policy_domain_name_test(self):
"""Parsing missing domain name in placement policy raises error"""
with self.assertRaises(CLIError):
sf_c.parse_placement_policies([{'type': 'PreferPrimaryDomain'}])
def parse_all_placement_policy_types_test(self):
"""Parse all placement policy types"""
from azure.servicefabric.models.service_placement_non_partially_place_service_policy_description import ServicePlacementNonPartiallyPlaceServicePolicyDescription # pylint: disable=line-too-long
from azure.servicefabric.models.service_placement_prefer_primary_domain_policy_description import ServicePlacementPreferPrimaryDomainPolicyDescription #pylint: disable=line-too-long
from azure.servicefabric.models.service_placement_required_domain_policy_description import ServicePlacementRequiredDomainPolicyDescription #pylint: disable=line-too-long
from azure.servicefabric.models.service_placement_require_domain_distribution_policy_description import ServicePlacementRequireDomainDistributionPolicyDescription #pylint: disable=line-too-long
res = sf_c.parse_placement_policies([{
'type': 'NonPartiallyPlaceService'
}, {
'type': 'PreferPrimaryDomain',
'domain_name': 'test_1'
}, {
'type': 'RequireDomain',
'domain_name': 'test-22'
}, {
'type': 'RequireDomainDistribution',
'domain_name': 'test_3'
}])
self.assertIsInstance(
res[0],
ServicePlacementNonPartiallyPlaceServicePolicyDescription
)
self.assertIsInstance(
res[1],
ServicePlacementPreferPrimaryDomainPolicyDescription
)
self.assertEqual(res[1].domain_name, 'test_1')
self.assertIsInstance(
res[2],
ServicePlacementRequiredDomainPolicyDescription
)
self.assertEqual(res[2].domain_name, 'test-22')
self.assertIsInstance(
res[3],
ServicePlacementRequireDomainDistributionPolicyDescription
)
self.assertEqual(res[3].domain_name, 'test_3')
def invalid_move_cost_test(self):
"""Invalid move cost raises error"""
with self.assertRaises(CLIError):
sf_c.validate_move_cost('test')
def empty_stateful_flags_test(self):
"""Empty stateful flags returns zero"""
self.assertEqual(sf_c.stateful_flags(), 0)
def all_stateful_flags_test(self):
"""All stateful flags sum up to correct value"""
self.assertEqual(sf_c.stateful_flags(10, 10, 10), 7)
def empty_service_update_flags_test(self):
"""Empty service update flags returns zero"""
self.assertEqual(sf_c.service_update_flags(), 0)
def all_service_update_flags_test(self):
"""All service update flags sum up to correct value"""
self.assertEqual(sf_c.service_update_flags(target_rep_size=1,
rep_restart_wait=10,
quorum_loss_wait=10,
standby_rep_keep=10,
min_rep_size=5,
placement_constraints='',
placement_policy='',
correlation='',
metrics='',
move_cost='high'), 1023)
def service_create_missing_service_state_test(self):
"""Service create must specify exactly stateful or stateless"""
with self.assertRaises(CLIError):
sf_c.validate_service_create_params(False, False, None, None,
None, None, None, None)
with self.assertRaises(CLIError):
sf_c.validate_service_create_params(True, True, None, None, None,
None, None, None)
def service_create_target_size_matches_state_test(self):
"""Service create target replica set and instance count match
stateful or stateless"""
with self.assertRaises(CLIError):
sf_c.validate_service_create_params(True, False, True, False,
False, 10, None, None)
with self.assertRaises(CLIError):
sf_c.validate_service_create_params(False, True, True, False,
False, None, 10, None)
def service_create_missing_stateful_replica_set_sizes_test(self):
"""Service create without target or min replica set sizes raises
error"""
with self.assertRaises(CLIError):
sf_c.validate_service_create_params(True, False, True, False,
False, None, 10, None)
def parse_incomplete_partition_policy_named_scheme_test(self):
"""Parsing named partition policy with unspecified names raises
error"""
with self.assertRaises(CLIError):
sf_c.parse_partition_policy(True, None, None, None, None, None,
None)
def parse_incomplete_partition_policy_int_test(self):
"""Parsing int partition policy with incomplete args raises error"""
with self.assertRaises(CLIError):
sf_c.parse_partition_policy(False, None, True, 0, 5, None, False)
def parse_multiple_partition_policy_test(self):
"""Parsing multiple different partition polices raises error"""
with self.assertRaises(CLIError):
sf_c.parse_partition_policy(True, ['test'], True, 0, 5, 3, True)
def parse_valid_partition_policy_test(self):
"""Parsing valid partition polices returns correct policies"""
from azure.servicefabric.models.named_partition_scheme_description import NamedPartitionSchemeDescription #pylint: disable=line-too-long
from azure.servicefabric.models.singleton_partition_scheme_description import SingletonPartitionSchemeDescription #pylint:disable=line-too-long
from azure.servicefabric.models.uniform_int64_range_partition_scheme_description import UniformInt64RangePartitionSchemeDescription #pylint:disable=line-too-long
res = sf_c.parse_partition_policy(True, ['test'], False, None, None,
None, False)
self.assertIsInstance(res, NamedPartitionSchemeDescription)
self.assertEqual(res.count, 1)
self.assertEqual(res.names, ['test'])
res = sf_c.parse_partition_policy(False, None, True, 1, 5, 3, False)
self.assertIsInstance(res, UniformInt64RangePartitionSchemeDescription)
self.assertEqual(res.high_key, 5)
self.assertEqual(res.low_key, 1)
self.assertEqual(res.count, 3)
res = sf_c.parse_partition_policy(False, None, False, None, None, None,
True)
self.assertIsInstance(res, SingletonPartitionSchemeDescription)
def activation_mode_invalid_test(self):
"""Invalid activation mode specified raises error"""
with self.assertRaises(CLIError):
sf_c.validate_activation_mode('test')
def activation_mode_none_test(self): #pylint: disable=no-self-use
"""None activation mode is considered valid"""
sf_c.validate_activation_mode(None)
def service_update_specify_state_test(self):
"""Service update incorrectly specifying service state raises error"""
with self.assertRaises(CLIError):
sf_c.validate_update_service_params(False, False, 10, 0, 10,
10, 10, False)
def service_update_stateful_invalid_params_test(self):
"""Stateful service update with invalid args raises error"""
with self.assertRaises(CLIError):
sf_c.validate_update_service_params(False, True, 5, 3, 10,
10, 10, 1)
def service_update_stateless_invalid_params_test(self):
"""Stateless service update with invalid args raises error"""
with self.assertRaises(CLIError):
sf_c.validate_update_service_params(True, False, 5, None, None,
None, None, 10)
with self.assertRaises(CLIError):
sf_c.validate_update_service_params(True, False, None, 1, None,
None, None, 10)
with self.assertRaises(CLIError):
sf_c.validate_update_service_params(True, False, None, None, 10,
None, None, 10)
with self.assertRaises(CLIError):
sf_c.validate_update_service_params(True, False, None, None, None,
10, None, 10)
with self.assertRaises(CLIError):
sf_c.validate_update_service_params(True, False, None, None, None,
None, 5, 10)