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 CreateNetworkTestCase(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.create_network(
190 constants.SCOPE_CLUSTER, group, version, plural, body)
191 mock_api.create_cluster_custom_object.assert_called_once_with(
192 group, version, plural, body)
194 mock_api.reset_mock()
195 kubernetes_utils.create_network(
196 constants.SCOPE_NAMESPACED, group, version, plural, body)
197 mock_api.create_namespaced_custom_object.assert_called_once_with(
198 group, version, 'default', plural, body)
201 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
202 def test_execute_exception(self, mock_get_api):
203 mock_api = mock.Mock()
204 mock_api.create_cluster_custom_object.side_effect = rest.ApiException
205 mock_get_api.return_value = mock_api
206 with self.assertRaises(exceptions.KubernetesApiException):
207 kubernetes_utils.create_network(
208 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
212 class DeleteNetworkTestCase(base.BaseUnitTestCase):
213 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
214 def test_execute_correct(self, mock_get_api):
215 mock_api = mock.Mock()
216 mock_get_api.return_value = mock_api
218 version = mock.Mock()
222 kubernetes_utils.delete_network(
223 constants.SCOPE_CLUSTER, group, version, plural, name)
224 mock_api.delete_cluster_custom_object.assert_called_once_with(
225 group, version, plural, name, {})
227 mock_api.reset_mock()
228 kubernetes_utils.delete_network(
229 constants.SCOPE_NAMESPACED, group, version, plural, name)
230 mock_api.delete_namespaced_custom_object.assert_called_once_with(
231 group, version, 'default', plural, name, {})
233 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
234 def test_execute_exception(self, mock_get_api):
235 mock_api = mock.Mock()
236 mock_api.delete_cluster_custom_object.side_effect = rest.ApiException
237 mock_get_api.return_value = mock_api
238 with self.assertRaises(exceptions.KubernetesApiException):
239 kubernetes_utils.delete_network(
240 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
243 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
244 @mock.patch.object(kubernetes_utils, 'LOG')
245 def test_execute_skip_exception(self, mock_log, mock_get_api):
246 mock_api = mock.Mock()
247 mock_api.delete_cluster_custom_object.side_effect = rest.ApiException(status=404)
249 mock_get_api.return_value = mock_api
250 kubernetes_utils.delete_network(
251 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
252 mock.ANY, skip_codes=[404])
254 mock_log.info.assert_called_once()
257 class DeletePodTestCase(base.BaseUnitTestCase):
258 @mock.patch.object(kubernetes_utils, 'get_core_api')
259 def test_execute_correct(self, mock_get_api):
260 mock_api = mock.Mock()
261 mock_get_api.return_value = mock_api
263 kubernetes_utils.delete_pod("name", body=None)
264 mock_api.delete_namespaced_pod.assert_called_once_with(
265 "name", 'default', None)
267 @mock.patch.object(kubernetes_utils, 'get_core_api')
268 def test_execute_exception(self, mock_get_api):
269 mock_api = mock.Mock()
270 mock_api.delete_namespaced_pod.side_effect = rest.ApiException(status=200)
272 mock_get_api.return_value = mock_api
273 with self.assertRaises(exceptions.KubernetesApiException):
274 kubernetes_utils.delete_pod(mock.ANY, skip_codes=[404])
276 @mock.patch.object(kubernetes_utils, 'LOG')
277 @mock.patch.object(kubernetes_utils, 'get_core_api')
278 def test_execute_skip_exception(self, mock_get_api, *args):
279 mock_api = mock.Mock()
280 mock_api.delete_namespaced_pod.side_effect = rest.ApiException(status=404)
282 mock_get_api.return_value = mock_api
283 kubernetes_utils.delete_pod(mock.ANY, skip_codes=[404])
286 class DeleteServiceTestCase(base.BaseUnitTestCase):
287 @mock.patch.object(client, "V1DeleteOptions")
288 @mock.patch.object(kubernetes_utils, 'get_core_api')
289 def test_execute_correct(self, mock_get_api, mock_options):
290 mock_api = mock.Mock()
291 mock_get_api.return_value = mock_api
292 mock_options.return_value = None
293 kubernetes_utils.delete_service("name", "default", None)
294 mock_api.delete_namespaced_service.assert_called_once_with(
295 "name", 'default', None)
297 @mock.patch.object(kubernetes_utils, 'get_core_api')
298 def test_execute_exception(self, mock_get_api):
299 mock_api = mock.Mock()
300 mock_api.delete_namespaced_service.side_effect = rest.ApiException(status=200)
302 mock_get_api.return_value = mock_api
303 with self.assertRaises(exceptions.KubernetesApiException):
304 kubernetes_utils.delete_service(mock.ANY, skip_codes=[404])
306 @mock.patch.object(kubernetes_utils, 'LOG')
307 @mock.patch.object(kubernetes_utils, 'get_core_api')
308 def test_execute_skip_exception(self, mock_get_api, *args):
309 mock_api = mock.Mock()
310 mock_api.delete_namespaced_service.side_effect = rest.ApiException(status=404)
312 mock_get_api.return_value = mock_api
313 kubernetes_utils.delete_service(mock.ANY, skip_codes=[404])