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,
244 class DeletePodTestCase(base.BaseUnitTestCase):
245 @mock.patch.object(kubernetes_utils, 'get_core_api')
246 def test_execute_correct(self, mock_get_api):
247 mock_api = mock.Mock()
248 mock_get_api.return_value = mock_api
250 kubernetes_utils.delete_pod("name", body=None)
251 mock_api.delete_namespaced_pod.assert_called_once_with(
252 "name", 'default', None)
254 @mock.patch.object(kubernetes_utils, 'get_core_api')
255 def test_execute_exception(self, mock_get_api):
256 mock_api = mock.Mock()
257 mock_api.delete_namespaced_pod.side_effect = rest.ApiException(status=200)
259 mock_get_api.return_value = mock_api
260 with self.assertRaises(exceptions.KubernetesApiException):
261 kubernetes_utils.delete_pod(mock.ANY, skip_codes=[404])
263 @mock.patch.object(kubernetes_utils, 'get_core_api')
264 def test_execute_skip_exception(self, mock_get_api):
265 mock_api = mock.Mock()
266 mock_api.delete_namespaced_pod.side_effect = rest.ApiException(status=404)
268 mock_get_api.return_value = mock_api
269 kubernetes_utils.delete_pod(mock.ANY, skip_codes=[404])
272 class DeleteServiceTestCase(base.BaseUnitTestCase):
273 @mock.patch.object(client, "V1DeleteOptions")
274 @mock.patch.object(kubernetes_utils, 'get_core_api')
275 def test_execute_correct(self, mock_get_api, mock_options):
276 mock_api = mock.Mock()
277 mock_get_api.return_value = mock_api
278 mock_options.return_value = None
279 kubernetes_utils.delete_service("name", "default", None)
280 mock_api.delete_namespaced_service.assert_called_once_with(
281 "name", 'default', None)
283 @mock.patch.object(kubernetes_utils, 'get_core_api')
284 def test_execute_exception(self, mock_get_api):
285 mock_api = mock.Mock()
286 mock_api.delete_namespaced_service.side_effect = rest.ApiException(status=200)
288 mock_get_api.return_value = mock_api
289 with self.assertRaises(exceptions.KubernetesApiException):
290 kubernetes_utils.delete_service(mock.ANY, skip_codes=[404])
292 @mock.patch.object(kubernetes_utils, 'get_core_api')
293 @mock.patch.object(kubernetes_utils, 'LOG')
294 def test_execute_skip_exception(self, mock_log, mock_get_api):
295 mock_api = mock.Mock()
296 mock_api.delete_namespaced_service.side_effect = rest.ApiException(status=404)
298 mock_get_api.return_value = mock_api
299 kubernetes_utils.delete_service(mock.ANY, skip_codes=[404])
300 mock_log.info.assert_called_once()