yardstick/tests/unit/common/test_kubernetes_utils.py

   1 # Copyright (c) 2018 Intel Corporation
   2 #
   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
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 import mock
  16 from kubernetes import client
  17 from kubernetes.client import rest
  18 from kubernetes import config
  19
  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
  24
  25
  26 class GetExtensionsV1betaApiTestCase(base.BaseUnitTestCase):
  27
  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()
  35
  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()
  41
  42
  43 class GetCustomObjectsApiTestCase(base.BaseUnitTestCase):
  44
  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()
  52
  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()
  58
  59
  60 class CreateCustomResourceDefinitionTestCase(base.BaseUnitTestCase):
  61
  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'}
  69
  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')
  76
  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'}
  86
  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')
  94
  95
  96 class DeleteCustomResourceDefinitionTestCase(base.BaseUnitTestCase):
  97
  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
 103
 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')
 109
 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
 117
 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')
 123
 124
 125 class GetCustomResourceDefinitionTestCase(base.BaseUnitTestCase):
 126
 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
 136         self.assertEqual(
 137             crd_obj,
 138             kubernetes_utils.get_custom_resource_definition('some_kind'))
 139
 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
 149         self.assertIsNone(
 150             kubernetes_utils.get_custom_resource_definition('other_kind'))
 151
 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')
 160
 161
 162 class GetNetworkTestCase(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
 167         group = 'group.com'
 168         version = mock.Mock()
 169         plural = 'networks'
 170         name = 'net_one'
 171
 172         kubernetes_utils.get_network(
 173             constants.SCOPE_CLUSTER, group, version, plural, name)
 174         mock_api.get_cluster_custom_object.assert_called_once_with(
 175             group, version, plural, name)
 176
 177         mock_api.reset_mock()
 178         kubernetes_utils.get_network(
 179             constants.SCOPE_NAMESPACED, group, version, plural, name)
 180         mock_api.get_namespaced_custom_object.assert_called_once_with(
 181             group, version, 'default', plural, name)
 182
 183     @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
 184     def test_execute_exception(self, mock_get_api):
 185         mock_api = mock.Mock()
 186         mock_api.get_cluster_custom_object.side_effect = rest.ApiException(404)
 187         mock_api.get_namespaced_custom_object.side_effect = rest.ApiException(404)
 188         mock_get_api.return_value = mock_api
 189         group = 'group.com'
 190         version = mock.Mock()
 191         plural = 'networks'
 192         name = 'net_one'
 193
 194         network_obj = kubernetes_utils.get_network(
 195             constants.SCOPE_CLUSTER, group, version, plural, name)
 196         self.assertIsNone(network_obj)
 197
 198         mock_api.reset_mock()
 199         network_obj = kubernetes_utils.get_network(
 200            constants.SCOPE_NAMESPACED, group, version, plural, name)
 201         self.assertIsNone(network_obj)
 202
 203
 204 class CreateNetworkTestCase(base.BaseUnitTestCase):
 205     @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
 206     @mock.patch.object(kubernetes_utils, 'get_network')
 207     def test_execute_correct(self, mock_get_net, mock_get_api):
 208         mock_get_net.return_value = None
 209         mock_api = mock.Mock()
 210         mock_get_api.return_value = mock_api
 211         group = 'group.com'
 212         version = mock.Mock()
 213         plural = 'networks'
 214         body = mock.Mock()
 215         name = 'net_one'
 216
 217         kubernetes_utils.create_network(
 218             constants.SCOPE_CLUSTER, group, version, plural, body, name)
 219         mock_api.create_cluster_custom_object.assert_called_once_with(
 220             group, version, plural, body)
 221
 222         mock_api.reset_mock()
 223         kubernetes_utils.create_network(
 224             constants.SCOPE_NAMESPACED, group, version, plural, body, name)
 225         mock_api.create_namespaced_custom_object.assert_called_once_with(
 226             group, version, 'default', plural, body)
 227
 228     @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
 229     @mock.patch.object(kubernetes_utils, 'get_network')
 230     def test_network_already_created(self, mock_get_net, mock_get_api):
 231         mock_get_net.return_value = mock.Mock
 232         mock_api = mock.Mock()
 233         mock_get_api.return_value = mock_api
 234         group = 'group.com'
 235         version = mock.Mock()
 236         plural = 'networks'
 237         body = mock.Mock()
 238         name = 'net_one'
 239
 240         mock_api.reset_mock()
 241         kubernetes_utils.create_network(
 242             constants.SCOPE_CLUSTER, group, version, plural, body, name)
 243         mock_api.create_cluster_custom_object.assert_not_called()
 244
 245         mock_api.reset_mock()
 246         kubernetes_utils.create_network(
 247             constants.SCOPE_NAMESPACED, group, version, plural, body, name)
 248         mock_api.create_namespaced_custom_object.assert_not_called()
 249
 250     @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
 251     @mock.patch.object(kubernetes_utils, 'get_network')
 252     def test_execute_exception(self, mock_get_net, mock_get_api):
 253         mock_get_net.return_value = None
 254         mock_api = mock.Mock()
 255         mock_api.create_cluster_custom_object.side_effect = rest.ApiException
 256         mock_get_api.return_value = mock_api
 257         with self.assertRaises(exceptions.KubernetesApiException):
 258             kubernetes_utils.create_network(
 259                 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
 260                 mock.ANY, mock.ANY)
 261
 262
 263 class DeleteNetworkTestCase(base.BaseUnitTestCase):
 264     @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
 265     def test_execute_correct(self, mock_get_api):
 266         mock_api = mock.Mock()
 267         mock_get_api.return_value = mock_api
 268         group = 'group.com'
 269         version = mock.Mock()
 270         plural = 'networks'
 271         name = 'network'
 272
 273         kubernetes_utils.delete_network(
 274             constants.SCOPE_CLUSTER, group, version, plural, name)
 275         mock_api.delete_cluster_custom_object.assert_called_once_with(
 276             group, version, plural, name, {})
 277
 278         mock_api.reset_mock()
 279         kubernetes_utils.delete_network(
 280             constants.SCOPE_NAMESPACED, group, version, plural, name)
 281         mock_api.delete_namespaced_custom_object.assert_called_once_with(
 282             group, version, 'default', plural, name, {})
 283
 284     @mock.patch.object(kubernetes_utils, 'get_custom_objects_api')
 285     def test_execute_exception(self, mock_get_api):
 286         mock_api = mock.Mock()
 287         mock_api.delete_cluster_custom_object.side_effect = rest.ApiException
 288         mock_get_api.return_value = mock_api
 289         with self.assertRaises(exceptions.KubernetesApiException):
 290             kubernetes_utils.delete_network(
 291                 constants.SCOPE_CLUSTER, mock.ANY, mock.ANY, mock.ANY,
 292                 mock.ANY)