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 from kubernetes.client import (
  20     V1ObjectMeta,
  21     V1Secret,
  22     V1ServiceAccount,
  23     V1SecretReference,
  24 )
  25
  26
  27 class FakeK8sResourceMetadata:
  28     def __init__(
  29         self,
  30         name: str = None,
  31         namespace: str = None,
  32         annotations: dict = {},
  33         labels: dict = {},
  34     ):
  35         self._annotations = annotations
  36         self._name = name or "name"
  37         self._namespace = namespace or "namespace"
  38         self._labels = labels or {"juju-app": "squid"}
  39
  40     @property
  41     def name(self):
  42         return self._name
  43
  44     @property
  45     def namespace(self):
  46         return self._namespace
  47
  48     @property
  49     def labels(self):
  50         return self._labels
  51
  52     @property
  53     def annotations(self):
  54         return self._annotations
  55
  56
  57 class FakeK8sStorageClass:
  58     def __init__(self, metadata=None):
  59         self._metadata = metadata or FakeK8sResourceMetadata()
  60
  61     @property
  62     def metadata(self):
  63         return self._metadata
  64
  65
  66 class FakeK8sStorageClassesList:
  67     def __init__(self, items=[]):
  68         self._items = items
  69
  70     @property
  71     def items(self):
  72         return self._items
  73
  74
  75 class FakeK8sServiceAccountsList:
  76     def __init__(self, items=[]):
  77         self._items = items
  78
  79     @property
  80     def items(self):
  81         return self._items
  82
  83
  84 class FakeK8sSecretList:
  85     def __init__(self, items=[]):
  86         self._items = items
  87
  88     @property
  89     def items(self):
  90         return self._items
  91
  92
  93 class FakeK8sVersionApiCode:
  94     def __init__(self, major: str, minor: str):
  95         self._major = major
  96         self._minor = minor
  97
  98     @property
  99     def major(self):
 100         return self._major
 101
 102     @property
 103     def minor(self):
 104         return self._minor
 105
 106
 107 fake_list_services = Dict(
 108     {
 109         "items": [
 110             Dict(
 111                 {
 112                     "metadata": Dict(
 113                         {
 114                             "name": "squid",
 115                             "namespace": "test",
 116                             "labels": {"juju-app": "squid"},
 117                         }
 118                     ),
 119                     "spec": Dict(
 120                         {
 121                             "cluster_ip": "10.152.183.79",
 122                             "type": "LoadBalancer",
 123                             "ports": [
 124                                 Dict(
 125                                     {
 126                                         "name": None,
 127                                         "node_port": None,
 128                                         "port": 30666,
 129                                         "protocol": "TCP",
 130                                         "target_port": 30666,
 131                                     }
 132                                 )
 133                             ],
 134                         }
 135                     ),
 136                     "status": Dict(
 137                         {
 138                             "load_balancer": Dict(
 139                                 {
 140                                     "ingress": [
 141                                         Dict({"hostname": None, "ip": "192.168.0.201"})
 142                                     ]
 143                                 }
 144                             )
 145                         }
 146                     ),
 147                 }
 148             )
 149         ]
 150     }
 151 )
 152
 153
 154 class KubectlTestCase(TestCase):
 155     def setUp(
 156         self,
 157     ):
 158         pass
 159
 160
 161 class FakeCoreV1Api:
 162     def list_service_for_all_namespaces(self, **kwargs):
 163         return fake_list_services
 164
 165
 166 class GetServices(TestCase):
 167     @mock.patch("n2vc.kubectl.config.load_kube_config")
 168     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 169     def setUp(self, mock_core, mock_config):
 170         mock_core.return_value = mock.MagicMock()
 171         mock_config.return_value = mock.MagicMock()
 172         self.kubectl = Kubectl()
 173
 174     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 175     def test_get_service(self, mock_corev1api):
 176         mock_corev1api.return_value = FakeCoreV1Api()
 177         services = self.kubectl.get_services(
 178             field_selector="metadata.namespace", label_selector="juju-operator=squid"
 179         )
 180         keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
 181         self.assertTrue(k in service for service in services for k in keys)
 182
 183     def test_get_service_exception(self):
 184         self.kubectl.clients[
 185             CORE_CLIENT
 186         ].list_service_for_all_namespaces.side_effect = ApiException()
 187         with self.assertRaises(ApiException):
 188             self.kubectl.get_services()
 189
 190
 191 @mock.patch("n2vc.kubectl.client")
 192 @mock.patch("n2vc.kubectl.config.kube_config.Configuration.get_default_copy")
 193 @mock.patch("n2vc.kubectl.config.load_kube_config")
 194 class GetConfiguration(KubectlTestCase):
 195     def setUp(self):
 196         super(GetConfiguration, self).setUp()
 197
 198     def test_get_configuration(
 199         self,
 200         mock_load_kube_config,
 201         mock_configuration,
 202         mock_client,
 203     ):
 204         kubectl = Kubectl()
 205         kubectl.configuration
 206         mock_configuration.assert_called_once()
 207         mock_load_kube_config.assert_called_once()
 208         mock_client.CoreV1Api.assert_called_once()
 209         mock_client.RbacAuthorizationV1Api.assert_called_once()
 210         mock_client.StorageV1Api.assert_called_once()
 211
 212
 213 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
 214 @mock.patch("kubernetes.config.load_kube_config")
 215 class GetDefaultStorageClass(KubectlTestCase):
 216     def setUp(self):
 217         super(GetDefaultStorageClass, self).setUp()
 218
 219         # Default Storage Class
 220         self.default_sc_name = "default-sc"
 221         default_sc_metadata = FakeK8sResourceMetadata(
 222             name=self.default_sc_name,
 223             annotations={"storageclass.kubernetes.io/is-default-class": "true"},
 224         )
 225         self.default_sc = FakeK8sStorageClass(metadata=default_sc_metadata)
 226
 227         # Default Storage Class with old annotation
 228         self.default_sc_old_name = "default-sc-old"
 229         default_sc_old_metadata = FakeK8sResourceMetadata(
 230             name=self.default_sc_old_name,
 231             annotations={"storageclass.beta.kubernetes.io/is-default-class": "true"},
 232         )
 233         self.default_sc_old = FakeK8sStorageClass(metadata=default_sc_old_metadata)
 234
 235         # Storage class - not default
 236         self.sc_name = "default-sc-old"
 237         self.sc = FakeK8sStorageClass(
 238             metadata=FakeK8sResourceMetadata(name=self.sc_name)
 239         )
 240
 241     def test_get_default_storage_class_exists_default(
 242         self, mock_load_kube_config, mock_list_storage_class
 243     ):
 244         kubectl = Kubectl()
 245         items = [self.default_sc]
 246         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 247         sc_name = kubectl.get_default_storage_class()
 248         self.assertEqual(sc_name, self.default_sc_name)
 249         mock_list_storage_class.assert_called_once()
 250
 251     def test_get_default_storage_class_exists_default_old(
 252         self, mock_load_kube_config, mock_list_storage_class
 253     ):
 254         kubectl = Kubectl()
 255         items = [self.default_sc_old]
 256         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 257         sc_name = kubectl.get_default_storage_class()
 258         self.assertEqual(sc_name, self.default_sc_old_name)
 259         mock_list_storage_class.assert_called_once()
 260
 261     def test_get_default_storage_class_none(
 262         self, mock_load_kube_config, mock_list_storage_class
 263     ):
 264         kubectl = Kubectl()
 265         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=[])
 266         sc_name = kubectl.get_default_storage_class()
 267         self.assertEqual(sc_name, None)
 268         mock_list_storage_class.assert_called_once()
 269
 270     def test_get_default_storage_class_exists_not_default(
 271         self, mock_load_kube_config, mock_list_storage_class
 272     ):
 273         kubectl = Kubectl()
 274         items = [self.sc]
 275         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 276         sc_name = kubectl.get_default_storage_class()
 277         self.assertEqual(sc_name, self.sc_name)
 278         mock_list_storage_class.assert_called_once()
 279
 280     def test_get_default_storage_class_choose(
 281         self, mock_load_kube_config, mock_list_storage_class
 282     ):
 283         kubectl = Kubectl()
 284         items = [self.sc, self.default_sc]
 285         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 286         sc_name = kubectl.get_default_storage_class()
 287         self.assertEqual(sc_name, self.default_sc_name)
 288         mock_list_storage_class.assert_called_once()
 289
 290
 291 @mock.patch("kubernetes.client.VersionApi.get_code")
 292 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_secret")
 293 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_secret")
 294 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_service_account")
 295 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_service_account")
 296 class CreateServiceAccountClass(KubectlTestCase):
 297     @mock.patch("kubernetes.config.load_kube_config")
 298     def setUp(self, mock_load_kube_config):
 299         super(CreateServiceAccountClass, self).setUp()
 300         self.service_account_name = "Service_account"
 301         self.labels = {"Key1": "Value1", "Key2": "Value2"}
 302         self.namespace = "kubernetes"
 303         self.token_id = "abc12345"
 304         self.kubectl = Kubectl()
 305
 306     def assert_create_secret(self, mock_create_secret, secret_name):
 307         annotations = {"kubernetes.io/service-account.name": self.service_account_name}
 308         secret_metadata = V1ObjectMeta(
 309             name=secret_name, namespace=self.namespace, annotations=annotations
 310         )
 311         secret_type = "kubernetes.io/service-account-token"
 312         secret = V1Secret(metadata=secret_metadata, type=secret_type)
 313         mock_create_secret.assert_called_once_with(self.namespace, secret)
 314
 315     def assert_create_service_account_v_1_24(
 316         self, mock_create_service_account, secret_name
 317     ):
 318         sevice_account_metadata = V1ObjectMeta(
 319             name=self.service_account_name, labels=self.labels, namespace=self.namespace
 320         )
 321         secrets = [V1SecretReference(name=secret_name, namespace=self.namespace)]
 322         service_account = V1ServiceAccount(
 323             metadata=sevice_account_metadata, secrets=secrets
 324         )
 325         mock_create_service_account.assert_called_once_with(
 326             self.namespace, service_account
 327         )
 328
 329     def assert_create_service_account_v_1_23(self, mock_create_service_account):
 330         metadata = V1ObjectMeta(
 331             name=self.service_account_name, labels=self.labels, namespace=self.namespace
 332         )
 333         service_account = V1ServiceAccount(metadata=metadata)
 334         mock_create_service_account.assert_called_once_with(
 335             self.namespace, service_account
 336         )
 337
 338     @mock.patch("n2vc.kubectl.uuid.uuid4")
 339     def test_secret_is_created_when_k8s_1_24(
 340         self,
 341         mock_uuid4,
 342         mock_list_service_account,
 343         mock_create_service_account,
 344         mock_create_secret,
 345         mock_list_secret,
 346         mock_version,
 347     ):
 348         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[])
 349         mock_list_secret.return_value = FakeK8sSecretList(items=[])
 350         mock_version.return_value = FakeK8sVersionApiCode("1", "24")
 351         mock_uuid4.return_value = self.token_id
 352         self.kubectl.create_service_account(
 353             self.service_account_name, self.labels, self.namespace
 354         )
 355         secret_name = "{}-token-{}".format(self.service_account_name, self.token_id[:5])
 356         self.assert_create_service_account_v_1_24(
 357             mock_create_service_account, secret_name
 358         )
 359         self.assert_create_secret(mock_create_secret, secret_name)
 360
 361     def test_secret_is_not_created_when_k8s_1_23(
 362         self,
 363         mock_list_service_account,
 364         mock_create_service_account,
 365         mock_create_secret,
 366         mock_list_secret,
 367         mock_version,
 368     ):
 369         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[])
 370         mock_version.return_value = FakeK8sVersionApiCode("1", "23+")
 371         self.kubectl.create_service_account(
 372             self.service_account_name, self.labels, self.namespace
 373         )
 374         self.assert_create_service_account_v_1_23(mock_create_service_account)
 375         mock_create_secret.assert_not_called()
 376         mock_list_secret.assert_not_called()
 377
 378     def test_raise_exception_if_service_account_already_exists(
 379         self,
 380         mock_list_service_account,
 381         mock_create_service_account,
 382         mock_create_secret,
 383         mock_list_secret,
 384         mock_version,
 385     ):
 386         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[1])
 387         with self.assertRaises(Exception) as context:
 388             self.kubectl.create_service_account(
 389                 self.service_account_name, self.labels, self.namespace
 390             )
 391         self.assertTrue(
 392             "Service account with metadata.name={} already exists".format(
 393                 self.service_account_name
 394             )
 395             in str(context.exception)
 396         )
 397         mock_create_service_account.assert_not_called()
 398         mock_create_secret.assert_not_called()
 399
 400     @mock.patch("n2vc.kubectl.uuid.uuid4")
 401     def test_raise_exception_if_secret_already_exists(
 402         self,
 403         mock_uuid4,
 404         mock_list_service_account,
 405         mock_create_service_account,
 406         mock_create_secret,
 407         mock_list_secret,
 408         mock_version,
 409     ):
 410         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[])
 411         mock_list_secret.return_value = FakeK8sSecretList(items=[1])
 412         mock_version.return_value = FakeK8sVersionApiCode("1", "24+")
 413         mock_uuid4.return_value = self.token_id
 414         with self.assertRaises(Exception) as context:
 415             self.kubectl.create_service_account(
 416                 self.service_account_name, self.labels, self.namespace
 417             )
 418         self.assertTrue(
 419             "Secret with metadata.name={}-token-{} already exists".format(
 420                 self.service_account_name, self.token_id[:5]
 421             )
 422             in str(context.exception)
 423         )
 424         mock_create_service_account.assert_called()
 425         mock_create_secret.assert_not_called()