1 # Copyright (c) 2018 Intel Corporation
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
16 from kubernetes import client
17 from kubernetes.client import rest
18 from kubernetes import config
20 from yardstick.common import constants
21 from yardstick.common import exceptions
22 from yardstick.common import kubernetes_utils
23 from yardstick.tests.unit import base
26 class GetExtensionsV1betaApiTestCase(base.BaseUnitTestCase):
28 @mock.patch.object(client, 'ApiextensionsV1beta1Api', return_value='api')
29 @mock.patch.object(config, 'load_kube_config')
30 def test_execute_correct(self, mock_load_kube_config, mock_api):
31 self.assertEqual('api', kubernetes_utils.get_extensions_v1beta_api())
32 mock_load_kube_config.assert_called_once_with(
33 config_file=constants.K8S_CONF_FILE)
34 mock_api.assert_called_once()
36 @mock.patch.object(config, 'load_kube_config')
37 def test_execute_exception(self, mock_load_kube_config):
38 mock_load_kube_config.side_effect = IOError
39 with self.assertRaises(exceptions.KubernetesConfigFileNotFound):
40 kubernetes_utils.get_extensions_v1beta_api()
43 class GetCustomObjectsApiTestCase(base.BaseUnitTestCase):
45 @mock.patch.object(client, 'CustomObjectsApi', return_value='api')
46 @mock.patch.object(config, 'load_kube_config')
47 def test_execute_correct(self, mock_load_kube_config, mock_api):
48 self.assertEqual('api', kubernetes_utils.get_custom_objects_api())
49 mock_load_kube_config.assert_called_once_with(
50 config_file=constants.K8S_CONF_FILE)
51 mock_api.assert_called_once()
53 @mock.patch.object(config, 'load_kube_config')
54 def test_execute_exception(self, mock_load_kube_config):
55 mock_load_kube_config.side_effect = IOError
56 with self.assertRaises(exceptions.KubernetesConfigFileNotFound):
57 kubernetes_utils.get_custom_objects_api()
60 class CreateCustomResourceDefinitionTestCase(base.BaseUnitTestCase):
62 @mock.patch.object(client, 'V1beta1CustomResourceDefinition',
63 return_value='crd_obj')
64 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
65 def test_execute_correct(self, mock_get_api, mock_crd):
66 mock_create_crd = mock.Mock()
67 mock_get_api.return_value = mock_create_crd
68 body = {'spec': 'fake_spec', 'metadata': 'fake_metadata'}
70 kubernetes_utils.create_custom_resource_definition(body)
71 mock_get_api.assert_called_once()
72 mock_crd.assert_called_once_with(spec='fake_spec',
73 metadata='fake_metadata')
74 mock_create_crd.create_custom_resource_definition.\
75 assert_called_once_with('crd_obj')
77 @mock.patch.object(client, 'V1beta1CustomResourceDefinition',
78 return_value='crd_obj')
79 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
80 def test_execute_exception(self, mock_get_api, mock_crd):
81 mock_create_crd = mock.Mock()
82 mock_create_crd.create_custom_resource_definition.\
83 side_effect = rest.ApiException
84 mock_get_api.return_value = mock_create_crd
85 body = {'spec': 'fake_spec', 'metadata': 'fake_metadata'}
87 with self.assertRaises(exceptions.KubernetesApiException):
88 kubernetes_utils.create_custom_resource_definition(body)
89 mock_get_api.assert_called_once()
90 mock_crd.assert_called_once_with(spec='fake_spec',
91 metadata='fake_metadata')
92 mock_create_crd.create_custom_resource_definition.\
93 assert_called_once_with('crd_obj')
96 class DeleteCustomResourceDefinitionTestCase(base.BaseUnitTestCase):
98 @mock.patch.object(client, 'V1DeleteOptions', return_value='del_obj')
99 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
100 def test_execute_correct(self, mock_get_api, mock_delobj):
101 mock_delete_crd = mock.Mock()
102 mock_get_api.return_value = mock_delete_crd
104 kubernetes_utils.delete_custom_resource_definition('name')
105 mock_get_api.assert_called_once()
106 mock_delobj.assert_called_once()
107 mock_delete_crd.delete_custom_resource_definition.\
108 assert_called_once_with('name', 'del_obj')
110 @mock.patch.object(client, 'V1DeleteOptions', return_value='del_obj')
111 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
112 def test_execute_exception(self, mock_get_api, mock_delobj):
113 mock_delete_crd = mock.Mock()
114 mock_delete_crd.delete_custom_resource_definition.\
115 side_effect = rest.ApiException
116 mock_get_api.return_value = mock_delete_crd
118 with self.assertRaises(exceptions.KubernetesApiException):
119 kubernetes_utils.delete_custom_resource_definition('name')
120 mock_delobj.assert_called_once()
121 mock_delete_crd.delete_custom_resource_definition.\
122 assert_called_once_with('name', 'del_obj')
124 @mock.patch.object(client, 'V1DeleteOptions', return_value='del_obj')
125 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
126 @mock.patch.object(kubernetes_utils, 'LOG')
127 def test_execute_skip_exception(self, mock_log, mock_get_api, mock_delobj):
128 mock_delete_crd = mock.Mock()
129 mock_delete_crd.delete_custom_resource_definition.side_effect = rest.ApiException(
132 mock_get_api.return_value = mock_delete_crd
133 kubernetes_utils.delete_custom_resource_definition('name', skip_codes=[404])
135 mock_delobj.assert_called_once()
136 mock_delete_crd.delete_custom_resource_definition.assert_called_once_with(
139 mock_log.info.assert_called_once()
142 class GetCustomResourceDefinitionTestCase(base.BaseUnitTestCase):
144 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
145 def test_execute_value(self, mock_get_api):
146 crd_obj = mock.Mock()
147 crd_obj.spec.names.kind = 'some_kind'
148 crd_list = mock.Mock()
149 crd_list.items = [crd_obj]
150 mock_api = mock.Mock()
151 mock_api.list_custom_resource_definition.return_value = crd_list
152 mock_get_api.return_value = mock_api
155 kubernetes_utils.get_custom_resource_definition('some_kind'))
157 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
158 def test_execute_none(self, mock_get_api):
159 crd_obj = mock.Mock()
160 crd_obj.spec.names.kind = 'some_kind'
161 crd_list = mock.Mock()
162 crd_list.items = [crd_obj]
163 mock_api = mock.Mock()
164 mock_api.list_custom_resource_definition.return_value = crd_list
165 mock_get_api.return_value = mock_api
167 kubernetes_utils.get_custom_resource_definition('other_kind'))
169 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
170 def test_execute_exception(self, mock_get_api):
171 mock_api = mock.Mock()
172 mock_api.list_custom_resource_definition.\
173 side_effect = rest.ApiException
174 mock_get_api.return_value = mock_api
175 with self.assertRaises(exceptions.KubernetesApiException):
176 kubernetes_utils.get_custom_resource_definition('kind')
179 class GetNetworkTestCase(base.BaseUnitTestCase):
180 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
181 def test_execute_correct(self, mock_get_api):
182 mock_api = mock.Mock()
183 mock_get_api.return_value = mock_api
185 version = mock.Mock()
189 kubernetes_utils.get_network(
190 constants.SCOPE_CLUSTER, group, version, plural, name)
191 mock_api.get_cluster_custom_object.assert_called_once_with(
192 group, version, plural, name)
194 mock_api.reset_mock()
195 kubernetes_utils.get_network(
196 constants.SCOPE_NAMESPACED, group, version, plural, name)
197 mock_api.get_namespaced_custom_object.assert_called_once_with(
198 group, version, 'default', plural, name)
200 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
201 def test_execute_exception(self, mock_get_api):
202 mock_api = mock.Mock()
203 mock_api.get_cluster_custom_object.side_effect = rest.ApiException(404)
204 mock_api.get_namespaced_custom_object.side_effect = rest.ApiException(404)
205 mock_get_api.return_value = mock_api
207 version = mock.Mock()
211 network_obj = kubernetes_utils.get_network(
212 constants.SCOPE_CLUSTER, group, version, plural, name)
213 self.assertIsNone(network_obj)
215 mock_api.reset_mock()
216 network_obj = kubernetes_utils.get_network(
217 constants.SCOPE_NAMESPACED, group, version, plural, name)
218 self.assertIsNone(network_obj)
221 class CreateNetworkTestCase(base.BaseUnitTestCase):
222 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
223 @mock.patch.object(kubernetes_utils, 'get_network')
224 def test_execute_correct(self, mock_get_net, mock_get_api):
225 mock_get_net.return_value = None
226 mock_api = mock.Mock()
227 mock_get_api.return_value = mock_api
229 version = mock.Mock()
234 kubernetes_utils.create_network(
235 constants.SCOPE_CLUSTER, group, version, plural, body, name)
236 mock_api.create_cluster_custom_object.assert_called_once_with(
237 group, version, plural, body)
239 mock_api.reset_mock()
240 kubernetes_utils.create_network(
241 constants.SCOPE_NAMESPACED, group, version, plural, body, name)
242 mock_api.create_namespaced_custom_object.assert_called_once_with(
243 group, version, 'default', plural, body)
245 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
246 @mock.patch.object(kubernetes_utils, 'get_network')
247 def test_network_already_created(self, mock_get_net, mock_get_api):
248 mock_get_net.return_value = mock.Mock
249 mock_api = mock.Mock()
250 mock_get_api.return_value = mock_api
252 version = mock.Mock()
257 mock_api.reset_mock()
258 kubernetes_utils.create_network(
259 constants.SCOPE_CLUSTER, group, version, plural, body, name)
260 mock_api.create_cluster_custom_object.assert_not_called()
262 mock_api.reset_mock()
263 kubernetes_utils.create_network(
264 constants.SCOPE_NAMESPACED, group, version, plural, body, name)
265 mock_api.create_namespaced_custom_object.assert_not_called()
267 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
268 @mock.patch.object(kubernetes_utils, 'get_network')
269 def test_execute_exception(self, mock_get_net, mock_get_api):
270 mock_get_net.return_value = None
271 mock_api = mock.Mock()
272 mock_api.create_cluster_custom_object.side_effect = rest.ApiException
273 mock_get_api.return_value = mock_api
274 with self.assertRaises(exceptions.KubernetesApiException):
275 kubernetes_utils.create_network(
276 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
280 class DeleteNetworkTestCase(base.BaseUnitTestCase):
281 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
282 def test_execute_correct(self, mock_get_api):
283 mock_api = mock.Mock()
284 mock_get_api.return_value = mock_api
286 version = mock.Mock()
290 kubernetes_utils.delete_network(
291 constants.SCOPE_CLUSTER, group, version, plural, name)
292 mock_api.delete_cluster_custom_object.assert_called_once_with(
293 group, version, plural, name, {})
295 mock_api.reset_mock()
296 kubernetes_utils.delete_network(
297 constants.SCOPE_NAMESPACED, group, version, plural, name)
298 mock_api.delete_namespaced_custom_object.assert_called_once_with(
299 group, version, 'default', plural, name, {})
301 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
302 def test_execute_exception(self, mock_get_api):
303 mock_api = mock.Mock()
304 mock_api.delete_cluster_custom_object.side_effect = rest.ApiException
305 mock_get_api.return_value = mock_api
306 with self.assertRaises(exceptions.KubernetesApiException):
307 kubernetes_utils.delete_network(
308 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
311 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
312 @mock.patch.object(kubernetes_utils, 'LOG')
313 def test_execute_skip_exception(self, mock_log, mock_get_api):
314 mock_api = mock.Mock()
315 mock_api.delete_cluster_custom_object.side_effect = rest.ApiException(status=404)
317 mock_get_api.return_value = mock_api
318 kubernetes_utils.delete_network(
319 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
320 mock.ANY, skip_codes=[404])
322 mock_log.info.assert_called_once()
325 class DeletePodTestCase(base.BaseUnitTestCase):
326 @mock.patch.object(kubernetes_utils, 'get_core_api')
327 def test_execute_correct(self, mock_get_api):
328 mock_api = mock.Mock()
329 mock_get_api.return_value = mock_api
331 kubernetes_utils.delete_pod("name", body=None)
332 mock_api.delete_namespaced_pod.assert_called_once_with(
333 "name", 'default', None)
335 @mock.patch.object(kubernetes_utils, 'get_core_api')
336 def test_execute_exception(self, mock_get_api):
337 mock_api = mock.Mock()
338 mock_api.delete_namespaced_pod.side_effect = rest.ApiException(status=200)
340 mock_get_api.return_value = mock_api
341 with self.assertRaises(exceptions.KubernetesApiException):
342 kubernetes_utils.delete_pod(mock.ANY, skip_codes=[404])
344 @mock.patch.object(kubernetes_utils, 'LOG')
345 @mock.patch.object(kubernetes_utils, 'get_core_api')
346 def test_execute_skip_exception(self, mock_get_api, *args):
347 mock_api = mock.Mock()
348 mock_api.delete_namespaced_pod.side_effect = rest.ApiException(status=404)
350 mock_get_api.return_value = mock_api
351 kubernetes_utils.delete_pod(mock.ANY, skip_codes=[404])
354 class DeleteServiceTestCase(base.BaseUnitTestCase):
355 @mock.patch.object(client, "V1DeleteOptions")
356 @mock.patch.object(kubernetes_utils, 'get_core_api')
357 def test_execute_correct(self, mock_get_api, mock_options):
358 mock_api = mock.Mock()
359 mock_get_api.return_value = mock_api
360 mock_options.return_value = None
361 kubernetes_utils.delete_service("name", "default", None)
362 mock_api.delete_namespaced_service.assert_called_once_with(
363 "name", 'default', None)
365 @mock.patch.object(kubernetes_utils, 'get_core_api')
366 def test_execute_exception(self, mock_get_api):
367 mock_api = mock.Mock()
368 mock_api.delete_namespaced_service.side_effect = rest.ApiException(status=200)
370 mock_get_api.return_value = mock_api
371 with self.assertRaises(exceptions.KubernetesApiException):
372 kubernetes_utils.delete_service(mock.ANY, skip_codes=[404])
374 @mock.patch.object(kubernetes_utils, 'LOG')
375 @mock.patch.object(kubernetes_utils, 'get_core_api')
376 def test_execute_skip_exception(self, mock_get_api, *args):
377 mock_api = mock.Mock()
378 mock_api.delete_namespaced_service.side_effect = rest.ApiException(status=404)
380 mock_get_api.return_value = mock_api
381 kubernetes_utils.delete_service(mock.ANY, skip_codes=[404])
384 class DeleteReplicationControllerTestCase(base.BaseUnitTestCase):
385 @mock.patch.object(kubernetes_utils, 'get_core_api')
386 def test_execute_correct(self, mock_get_api):
387 mock_api = mock.Mock()
388 mock_get_api.return_value = mock_api
389 kubernetes_utils.delete_replication_controller(
390 "name", "default", body=None)
392 mock_api.delete_namespaced_replication_controller.assert_called_once_with(
393 "name", "default", None)
395 @mock.patch.object(kubernetes_utils, 'get_core_api')
396 def test_execute_exception(self, mock_get_api):
397 mock_api = mock.Mock()
398 mock_api.delete_namespaced_replication_controller.side_effect = (
399 rest.ApiException(status=200)
402 mock_get_api.return_value = mock_api
403 with self.assertRaises(exceptions.KubernetesApiException):
404 kubernetes_utils.delete_replication_controller(mock.ANY, skip_codes=[404])
406 @mock.patch.object(kubernetes_utils, 'get_core_api')
407 @mock.patch.object(kubernetes_utils, 'LOG')
408 def test_execute_skip_exception(self, mock_log, mock_get_api):
409 mock_api = mock.Mock()
410 mock_api.delete_namespaced_replication_controller.side_effect = (
411 rest.ApiException(status=404)
414 mock_get_api.return_value = mock_api
415 kubernetes_utils.delete_replication_controller(mock.ANY, skip_codes=[404])
417 mock_log.info.assert_called_once()
420 class DeleteConfigMapTestCase(base.BaseUnitTestCase):
421 @mock.patch.object(kubernetes_utils, 'get_core_api')
422 def test_execute_correct(self, mock_get_api):
423 mock_api = mock.Mock()
424 mock_get_api.return_value = mock_api
425 kubernetes_utils.delete_config_map("name", body=None)
426 mock_api.delete_namespaced_config_map.assert_called_once_with(
427 "name", "default", None
430 @mock.patch.object(kubernetes_utils, 'get_core_api')
431 def test_execute_exception(self, mock_get_api):
432 mock_api = mock.Mock()
433 mock_api.delete_namespaced_config_map.side_effect = rest.ApiException(status=200)
435 mock_get_api.return_value = mock_api
436 with self.assertRaises(exceptions.KubernetesApiException):
437 kubernetes_utils.delete_config_map(mock.ANY, skip_codes=[404])
439 @mock.patch.object(kubernetes_utils, 'get_core_api')
440 @mock.patch.object(kubernetes_utils, 'LOG')
441 def test_execute_skip_exception(self, mock_log, mock_get_api):
442 mock_api = mock.Mock()
443 mock_api.delete_namespaced_config_map.side_effect = rest.ApiException(status=404)
445 mock_get_api.return_value = mock_api
446 kubernetes_utils.delete_config_map(mock.ANY, skip_codes=[404])
447 mock_log.info.assert_called_once()