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')
125 class GetCustomResourceDefinitionTestCase(base.BaseUnitTestCase):
127 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
128 def test_execute_value(self, mock_get_api):
129 crd_obj = mock.Mock()
130 crd_obj.spec.names.kind = 'some_kind'
131 crd_list = mock.Mock()
132 crd_list.items = [crd_obj]
133 mock_api = mock.Mock()
134 mock_api.list_custom_resource_definition.return_value = crd_list
135 mock_get_api.return_value = mock_api
138 kubernetes_utils.get_custom_resource_definition('some_kind'))
140 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
141 def test_execute_none(self, mock_get_api):
142 crd_obj = mock.Mock()
143 crd_obj.spec.names.kind = 'some_kind'
144 crd_list = mock.Mock()
145 crd_list.items = [crd_obj]
146 mock_api = mock.Mock()
147 mock_api.list_custom_resource_definition.return_value = crd_list
148 mock_get_api.return_value = mock_api
150 kubernetes_utils.get_custom_resource_definition('other_kind'))
152 @mock.patch.object(kubernetes_utils, 'get_extensions_v1beta_api')
153 def test_execute_exception(self, mock_get_api):
154 mock_api = mock.Mock()
155 mock_api.list_custom_resource_definition.\
156 side_effect = rest.ApiException
157 mock_get_api.return_value = mock_api
158 with self.assertRaises(exceptions.KubernetesApiException):
159 kubernetes_utils.get_custom_resource_definition('kind')
162 class CreateNetworkTestCase(base.BaseUnitTestCase):
163 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
164 def test_execute_correct(self, mock_get_api):
165 mock_api = mock.Mock()
166 mock_get_api.return_value = mock_api
168 version = mock.Mock()
172 kubernetes_utils.create_network(
173 constants.SCOPE_CLUSTER, group, version, plural, body)
174 mock_api.create_cluster_custom_object.assert_called_once_with(
175 group, version, plural, body)
177 mock_api.reset_mock()
178 kubernetes_utils.create_network(
179 constants.SCOPE_NAMESPACED, group, version, plural, body)
180 mock_api.create_namespaced_custom_object.assert_called_once_with(
181 group, version, 'default', plural, body)
184 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
185 def test_execute_exception(self, mock_get_api):
186 mock_api = mock.Mock()
187 mock_api.create_cluster_custom_object.side_effect = rest.ApiException
188 mock_get_api.return_value = mock_api
189 with self.assertRaises(exceptions.KubernetesApiException):
190 kubernetes_utils.create_network(
191 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
195 class DeleteNetworkTestCase(base.BaseUnitTestCase):
196 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
197 def test_execute_correct(self, mock_get_api):
198 mock_api = mock.Mock()
199 mock_get_api.return_value = mock_api
201 version = mock.Mock()
205 kubernetes_utils.delete_network(
206 constants.SCOPE_CLUSTER, group, version, plural, name)
207 mock_api.delete_cluster_custom_object.assert_called_once_with(
208 group, version, plural, name, {})
210 mock_api.reset_mock()
211 kubernetes_utils.delete_network(
212 constants.SCOPE_NAMESPACED, group, version, plural, name)
213 mock_api.delete_namespaced_custom_object.assert_called_once_with(
214 group, version, 'default', plural, name, {})
216 @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
217 def test_execute_exception(self, mock_get_api):
218 mock_api = mock.Mock()
219 mock_api.delete_cluster_custom_object.side_effect = rest.ApiException
220 mock_get_api.return_value = mock_api
221 with self.assertRaises(exceptions.KubernetesApiException):
222 kubernetes_utils.delete_network(
223 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,