n2vc/tests/unit/test_kubectl.py

   1 # Copyright 2020 Canonical Ltd.
   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 from unittest import TestCase, mock
  16 from n2vc.kubectl import Kubectl, CORE_CLIENT
  17 from n2vc.utils import Dict
  18 from kubernetes.client.rest import ApiException
  19
  20
  21 class FakeK8sResourceMetadata:
  22     def __init__(
  23         self,
  24         name: str = None,
  25         namespace: str = None,
  26         annotations: dict = {},
  27         labels: dict = {},
  28     ):
  29         self._annotations = annotations
  30         self._name = name or "name"
  31         self._namespace = namespace or "namespace"
  32         self._labels = labels or {"juju-app": "squid"}
  33
  34     @property
  35     def name(self):
  36         return self._name
  37
  38     @property
  39     def namespace(self):
  40         return self._namespace
  41
  42     @property
  43     def labels(self):
  44         return self._labels
  45
  46     @property
  47     def annotations(self):
  48         return self._annotations
  49
  50
  51 class FakeK8sStorageClass:
  52     def __init__(self, metadata=None):
  53         self._metadata = metadata or FakeK8sResourceMetadata()
  54
  55     @property
  56     def metadata(self):
  57         return self._metadata
  58
  59
  60 class FakeK8sStorageClassesList:
  61     def __init__(self, items=[]):
  62         self._items = items
  63
  64     @property
  65     def items(self):
  66         return self._items
  67
  68
  69 fake_list_services = Dict(
  70     {
  71         "items": [
  72             Dict(
  73                 {
  74                     "metadata": Dict(
  75                         {
  76                             "name": "squid",
  77                             "namespace": "test",
  78                             "labels": {"juju-app": "squid"},
  79                         }
  80                     ),
  81                     "spec": Dict(
  82                         {
  83                             "cluster_ip": "10.152.183.79",
  84                             "type": "LoadBalancer",
  85                             "ports": [
  86                                 Dict(
  87                                     {
  88                                         "name": None,
  89                                         "node_port": None,
  90                                         "port": 30666,
  91                                         "protocol": "TCP",
  92                                         "target_port": 30666,
  93                                     }
  94                                 )
  95                             ],
  96                         }
  97                     ),
  98                     "status": Dict(
  99                         {
 100                             "load_balancer": Dict(
 101                                 {
 102                                     "ingress": [
 103                                         Dict({"hostname": None, "ip": "192.168.0.201"})
 104                                     ]
 105                                 }
 106                             )
 107                         }
 108                     ),
 109                 }
 110             )
 111         ]
 112     }
 113 )
 114
 115
 116 class KubectlTestCase(TestCase):
 117     def setUp(
 118         self,
 119     ):
 120         pass
 121
 122
 123 class FakeCoreV1Api:
 124     def list_service_for_all_namespaces(self, **kwargs):
 125         return fake_list_services
 126
 127
 128 class GetServices(TestCase):
 129     @mock.patch("n2vc.kubectl.config.load_kube_config")
 130     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 131     def setUp(self, mock_core, mock_config):
 132         mock_core.return_value = mock.MagicMock()
 133         mock_config.return_value = mock.MagicMock()
 134         self.kubectl = Kubectl()
 135
 136     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 137     def test_get_service(self, mock_corev1api):
 138         mock_corev1api.return_value = FakeCoreV1Api()
 139         services = self.kubectl.get_services(
 140             field_selector="metadata.namespace", label_selector="juju-operator=squid"
 141         )
 142         keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
 143         self.assertTrue(k in service for service in services for k in keys)
 144
 145     def test_get_service_exception(self):
 146         self.kubectl.clients[
 147             CORE_CLIENT
 148         ].list_service_for_all_namespaces.side_effect = ApiException()
 149         with self.assertRaises(ApiException):
 150             self.kubectl.get_services()
 151
 152
 153 @mock.patch("kubernetes.config.kube_config.Configuration")
 154 @mock.patch("n2vc.kubectl.config.load_kube_config")
 155 class GetConfiguration(KubectlTestCase):
 156     def setUp(self):
 157         super(GetConfiguration, self).setUp()
 158
 159     def test_get_configuration(self, mock_load_kube_config, mock_configuration):
 160         kubectl = Kubectl()
 161         kubectl.configuration
 162         mock_configuration.assert_called_once()
 163
 164
 165 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
 166 @mock.patch("kubernetes.config.load_kube_config")
 167 class GetDefaultStorageClass(KubectlTestCase):
 168     def setUp(self):
 169         super(GetDefaultStorageClass, self).setUp()
 170
 171         # Default Storage Class
 172         self.default_sc_name = "default-sc"
 173         default_sc_metadata = FakeK8sResourceMetadata(
 174             name=self.default_sc_name,
 175             annotations={"storageclass.kubernetes.io/is-default-class": "true"},
 176         )
 177         self.default_sc = FakeK8sStorageClass(metadata=default_sc_metadata)
 178
 179         # Default Storage Class with old annotation
 180         self.default_sc_old_name = "default-sc-old"
 181         default_sc_old_metadata = FakeK8sResourceMetadata(
 182             name=self.default_sc_old_name,
 183             annotations={"storageclass.beta.kubernetes.io/is-default-class": "true"},
 184         )
 185         self.default_sc_old = FakeK8sStorageClass(metadata=default_sc_old_metadata)
 186
 187         # Storage class - not default
 188         self.sc_name = "default-sc-old"
 189         self.sc = FakeK8sStorageClass(
 190             metadata=FakeK8sResourceMetadata(name=self.sc_name)
 191         )
 192
 193     def test_get_default_storage_class_exists_default(
 194         self, mock_load_kube_config, mock_list_storage_class
 195     ):
 196         kubectl = Kubectl()
 197         items = [self.default_sc]
 198         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 199         sc_name = kubectl.get_default_storage_class()
 200         self.assertEqual(sc_name, self.default_sc_name)
 201         mock_list_storage_class.assert_called_once()
 202
 203     def test_get_default_storage_class_exists_default_old(
 204         self, mock_load_kube_config, mock_list_storage_class
 205     ):
 206         kubectl = Kubectl()
 207         items = [self.default_sc_old]
 208         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 209         sc_name = kubectl.get_default_storage_class()
 210         self.assertEqual(sc_name, self.default_sc_old_name)
 211         mock_list_storage_class.assert_called_once()
 212
 213     def test_get_default_storage_class_none(
 214         self, mock_load_kube_config, mock_list_storage_class
 215     ):
 216         kubectl = Kubectl()
 217         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=[])
 218         sc_name = kubectl.get_default_storage_class()
 219         self.assertEqual(sc_name, None)
 220         mock_list_storage_class.assert_called_once()
 221
 222     def test_get_default_storage_class_exists_not_default(
 223         self, mock_load_kube_config, mock_list_storage_class
 224     ):
 225         kubectl = Kubectl()
 226         items = [self.sc]
 227         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 228         sc_name = kubectl.get_default_storage_class()
 229         self.assertEqual(sc_name, self.sc_name)
 230         mock_list_storage_class.assert_called_once()
 231
 232     def test_get_default_storage_class_choose(
 233         self, mock_load_kube_config, mock_list_storage_class
 234     ):
 235         kubectl = Kubectl()
 236         items = [self.sc, self.default_sc]
 237         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 238         sc_name = kubectl.get_default_storage_class()
 239         self.assertEqual(sc_name, self.default_sc_name)
 240         mock_list_storage_class.assert_called_once()