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
  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(self,):
 118         pass
 119
 120
 121 class FakeCoreV1Api:
 122     def list_service_for_all_namespaces(self, **kwargs):
 123         return fake_list_services
 124
 125
 126 class ProvisionerTest(TestCase):
 127     @mock.patch("n2vc.kubectl.config.load_kube_config")
 128     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 129     def setUp(self, mock_core, mock_config):
 130         mock_core.return_value = mock.MagicMock()
 131         mock_config.return_value = mock.MagicMock()
 132         self.kubectl = Kubectl()
 133
 134     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 135     def test_get_service(self, mock_corev1api):
 136         mock_corev1api.return_value = FakeCoreV1Api()
 137         services = self.kubectl.get_services(
 138             field_selector="metadata.namespace", label_selector="juju-operator=squid"
 139         )
 140         keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
 141         self.assertTrue(k in service for service in services for k in keys)
 142
 143     @mock.patch("n2vc.kubectl.client.CoreV1Api.list_service_for_all_namespaces")
 144     def test_get_service_exception(self, list_services):
 145         list_services.side_effect = ApiException()
 146         with self.assertRaises(ApiException):
 147             self.kubectl.get_services()
 148
 149
 150 @mock.patch("kubernetes.config.kube_config.Configuration")
 151 @mock.patch("n2vc.kubectl.config.load_kube_config")
 152 class GetConfiguration(KubectlTestCase):
 153     def setUp(self):
 154         super(GetConfiguration, self).setUp()
 155
 156     def test_get_configuration(self, mock_load_kube_config, mock_configuration):
 157         kubectl = Kubectl()
 158         kubectl.get_configuration()
 159         mock_configuration.assert_called_once()
 160
 161
 162 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
 163 @mock.patch("kubernetes.config.load_kube_config")
 164 class GetDefaultStorageClass(KubectlTestCase):
 165     def setUp(self):
 166         super(GetDefaultStorageClass, self).setUp()
 167
 168         # Default Storage Class
 169         self.default_sc_name = "default-sc"
 170         default_sc_metadata = FakeK8sResourceMetadata(
 171             name=self.default_sc_name,
 172             annotations={"storageclass.kubernetes.io/is-default-class": "true"},
 173         )
 174         self.default_sc = FakeK8sStorageClass(metadata=default_sc_metadata)
 175
 176         # Default Storage Class with old annotation
 177         self.default_sc_old_name = "default-sc-old"
 178         default_sc_old_metadata = FakeK8sResourceMetadata(
 179             name=self.default_sc_old_name,
 180             annotations={"storageclass.beta.kubernetes.io/is-default-class": "true"},
 181         )
 182         self.default_sc_old = FakeK8sStorageClass(metadata=default_sc_old_metadata)
 183
 184         # Storage class - not default
 185         self.sc_name = "default-sc-old"
 186         self.sc = FakeK8sStorageClass(
 187             metadata=FakeK8sResourceMetadata(name=self.sc_name)
 188         )
 189
 190     def test_get_default_storage_class_exists_default(
 191         self, mock_load_kube_config, mock_list_storage_class
 192     ):
 193         kubectl = Kubectl()
 194         items = [self.default_sc]
 195         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 196         sc_name = kubectl.get_default_storage_class()
 197         self.assertEqual(sc_name, self.default_sc_name)
 198         mock_list_storage_class.assert_called_once()
 199
 200     def test_get_default_storage_class_exists_default_old(
 201         self, mock_load_kube_config, mock_list_storage_class
 202     ):
 203         kubectl = Kubectl()
 204         items = [self.default_sc_old]
 205         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 206         sc_name = kubectl.get_default_storage_class()
 207         self.assertEqual(sc_name, self.default_sc_old_name)
 208         mock_list_storage_class.assert_called_once()
 209
 210     def test_get_default_storage_class_none(
 211         self, mock_load_kube_config, mock_list_storage_class
 212     ):
 213         kubectl = Kubectl()
 214         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=[])
 215         sc_name = kubectl.get_default_storage_class()
 216         self.assertEqual(sc_name, None)
 217         mock_list_storage_class.assert_called_once()
 218
 219     def test_get_default_storage_class_exists_not_default(
 220         self, mock_load_kube_config, mock_list_storage_class
 221     ):
 222         kubectl = Kubectl()
 223         items = [self.sc]
 224         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 225         sc_name = kubectl.get_default_storage_class()
 226         self.assertEqual(sc_name, self.sc_name)
 227         mock_list_storage_class.assert_called_once()
 228
 229     def test_get_default_storage_class_choose(
 230         self, mock_load_kube_config, mock_list_storage_class
 231     ):
 232         kubectl = Kubectl()
 233         items = [self.sc, self.default_sc]
 234         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 235         sc_name = kubectl.get_default_storage_class()
 236         self.assertEqual(sc_name, self.default_sc_name)
 237         mock_list_storage_class.assert_called_once()