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 import asynctest
  16 import yaml
  17 import os
  18 from unittest import TestCase, mock
  19 from n2vc.kubectl import Kubectl, CORE_CLIENT, CUSTOM_OBJECT_CLIENT
  20 from n2vc.utils import Dict
  21 from kubernetes.client.rest import ApiException
  22 from kubernetes.client import (
  23     V1ObjectMeta,
  24     V1Secret,
  25     V1ServiceAccount,
  26     V1SecretReference,
  27 )
  28
  29
  30 class FakeK8sResourceMetadata:
  31     def __init__(
  32         self,
  33         name: str = None,
  34         namespace: str = None,
  35         annotations: dict = {},
  36         labels: dict = {},
  37     ):
  38         self._annotations = annotations
  39         self._name = name or "name"
  40         self._namespace = namespace or "namespace"
  41         self._labels = labels or {"juju-app": "squid"}
  42
  43     @property
  44     def name(self):
  45         return self._name
  46
  47     @property
  48     def namespace(self):
  49         return self._namespace
  50
  51     @property
  52     def labels(self):
  53         return self._labels
  54
  55     @property
  56     def annotations(self):
  57         return self._annotations
  58
  59
  60 class FakeK8sStorageClass:
  61     def __init__(self, metadata=None):
  62         self._metadata = metadata or FakeK8sResourceMetadata()
  63
  64     @property
  65     def metadata(self):
  66         return self._metadata
  67
  68
  69 class FakeK8sStorageClassesList:
  70     def __init__(self, items=[]):
  71         self._items = items
  72
  73     @property
  74     def items(self):
  75         return self._items
  76
  77
  78 class FakeK8sServiceAccountsList:
  79     def __init__(self, items=[]):
  80         self._items = items
  81
  82     @property
  83     def items(self):
  84         return self._items
  85
  86
  87 class FakeK8sSecretList:
  88     def __init__(self, items=[]):
  89         self._items = items
  90
  91     @property
  92     def items(self):
  93         return self._items
  94
  95
  96 class FakeK8sVersionApiCode:
  97     def __init__(self, major: str, minor: str):
  98         self._major = major
  99         self._minor = minor
 100
 101     @property
 102     def major(self):
 103         return self._major
 104
 105     @property
 106     def minor(self):
 107         return self._minor
 108
 109
 110 fake_list_services = Dict(
 111     {
 112         "items": [
 113             Dict(
 114                 {
 115                     "metadata": Dict(
 116                         {
 117                             "name": "squid",
 118                             "namespace": "test",
 119                             "labels": {"juju-app": "squid"},
 120                         }
 121                     ),
 122                     "spec": Dict(
 123                         {
 124                             "cluster_ip": "10.152.183.79",
 125                             "type": "LoadBalancer",
 126                             "ports": [
 127                                 Dict(
 128                                     {
 129                                         "name": None,
 130                                         "node_port": None,
 131                                         "port": 30666,
 132                                         "protocol": "TCP",
 133                                         "target_port": 30666,
 134                                     }
 135                                 )
 136                             ],
 137                         }
 138                     ),
 139                     "status": Dict(
 140                         {
 141                             "load_balancer": Dict(
 142                                 {
 143                                     "ingress": [
 144                                         Dict({"hostname": None, "ip": "192.168.0.201"})
 145                                     ]
 146                                 }
 147                             )
 148                         }
 149                     ),
 150                 }
 151             )
 152         ]
 153     }
 154 )
 155
 156
 157 class KubectlTestCase(TestCase):
 158     def setUp(
 159         self,
 160     ):
 161         pass
 162
 163
 164 class FakeCoreV1Api:
 165     def list_service_for_all_namespaces(self, **kwargs):
 166         return fake_list_services
 167
 168
 169 class GetServices(TestCase):
 170     @mock.patch("n2vc.kubectl.config.load_kube_config")
 171     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 172     def setUp(self, mock_core, mock_config):
 173         mock_core.return_value = mock.MagicMock()
 174         mock_config.return_value = mock.MagicMock()
 175         self.kubectl = Kubectl()
 176
 177     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 178     def test_get_service(self, mock_corev1api):
 179         mock_corev1api.return_value = FakeCoreV1Api()
 180         services = self.kubectl.get_services(
 181             field_selector="metadata.namespace", label_selector="juju-operator=squid"
 182         )
 183         keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
 184         self.assertTrue(k in service for service in services for k in keys)
 185
 186     def test_get_service_exception(self):
 187         self.kubectl.clients[
 188             CORE_CLIENT
 189         ].list_service_for_all_namespaces.side_effect = ApiException()
 190         with self.assertRaises(ApiException):
 191             self.kubectl.get_services()
 192
 193
 194 @mock.patch("n2vc.kubectl.client")
 195 @mock.patch("n2vc.kubectl.config.kube_config.Configuration.get_default_copy")
 196 @mock.patch("n2vc.kubectl.config.load_kube_config")
 197 class GetConfiguration(KubectlTestCase):
 198     def setUp(self):
 199         super(GetConfiguration, self).setUp()
 200
 201     def test_get_configuration(
 202         self,
 203         mock_load_kube_config,
 204         mock_configuration,
 205         mock_client,
 206     ):
 207         kubectl = Kubectl()
 208         kubectl.configuration
 209         mock_configuration.assert_called_once()
 210         mock_load_kube_config.assert_called_once()
 211         mock_client.CoreV1Api.assert_called_once()
 212         mock_client.RbacAuthorizationV1Api.assert_called_once()
 213         mock_client.StorageV1Api.assert_called_once()
 214
 215
 216 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
 217 @mock.patch("kubernetes.config.load_kube_config")
 218 class GetDefaultStorageClass(KubectlTestCase):
 219     def setUp(self):
 220         super(GetDefaultStorageClass, self).setUp()
 221
 222         # Default Storage Class
 223         self.default_sc_name = "default-sc"
 224         default_sc_metadata = FakeK8sResourceMetadata(
 225             name=self.default_sc_name,
 226             annotations={"storageclass.kubernetes.io/is-default-class": "true"},
 227         )
 228         self.default_sc = FakeK8sStorageClass(metadata=default_sc_metadata)
 229
 230         # Default Storage Class with old annotation
 231         self.default_sc_old_name = "default-sc-old"
 232         default_sc_old_metadata = FakeK8sResourceMetadata(
 233             name=self.default_sc_old_name,
 234             annotations={"storageclass.beta.kubernetes.io/is-default-class": "true"},
 235         )
 236         self.default_sc_old = FakeK8sStorageClass(metadata=default_sc_old_metadata)
 237
 238         # Storage class - not default
 239         self.sc_name = "default-sc-old"
 240         self.sc = FakeK8sStorageClass(
 241             metadata=FakeK8sResourceMetadata(name=self.sc_name)
 242         )
 243
 244     def test_get_default_storage_class_exists_default(
 245         self, mock_load_kube_config, mock_list_storage_class
 246     ):
 247         kubectl = Kubectl()
 248         items = [self.default_sc]
 249         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 250         sc_name = kubectl.get_default_storage_class()
 251         self.assertEqual(sc_name, self.default_sc_name)
 252         mock_list_storage_class.assert_called_once()
 253
 254     def test_get_default_storage_class_exists_default_old(
 255         self, mock_load_kube_config, mock_list_storage_class
 256     ):
 257         kubectl = Kubectl()
 258         items = [self.default_sc_old]
 259         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 260         sc_name = kubectl.get_default_storage_class()
 261         self.assertEqual(sc_name, self.default_sc_old_name)
 262         mock_list_storage_class.assert_called_once()
 263
 264     def test_get_default_storage_class_none(
 265         self, mock_load_kube_config, mock_list_storage_class
 266     ):
 267         kubectl = Kubectl()
 268         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=[])
 269         sc_name = kubectl.get_default_storage_class()
 270         self.assertEqual(sc_name, None)
 271         mock_list_storage_class.assert_called_once()
 272
 273     def test_get_default_storage_class_exists_not_default(
 274         self, mock_load_kube_config, mock_list_storage_class
 275     ):
 276         kubectl = Kubectl()
 277         items = [self.sc]
 278         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 279         sc_name = kubectl.get_default_storage_class()
 280         self.assertEqual(sc_name, self.sc_name)
 281         mock_list_storage_class.assert_called_once()
 282
 283     def test_get_default_storage_class_choose(
 284         self, mock_load_kube_config, mock_list_storage_class
 285     ):
 286         kubectl = Kubectl()
 287         items = [self.sc, self.default_sc]
 288         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 289         sc_name = kubectl.get_default_storage_class()
 290         self.assertEqual(sc_name, self.default_sc_name)
 291         mock_list_storage_class.assert_called_once()
 292
 293
 294 @mock.patch("kubernetes.client.VersionApi.get_code")
 295 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_secret")
 296 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_secret")
 297 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_service_account")
 298 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_service_account")
 299 class CreateServiceAccountClass(KubectlTestCase):
 300     @mock.patch("kubernetes.config.load_kube_config")
 301     def setUp(self, mock_load_kube_config):
 302         super(CreateServiceAccountClass, self).setUp()
 303         self.service_account_name = "Service_account"
 304         self.labels = {"Key1": "Value1", "Key2": "Value2"}
 305         self.namespace = "kubernetes"
 306         self.token_id = "abc12345"
 307         self.kubectl = Kubectl()
 308
 309     def assert_create_secret(self, mock_create_secret, secret_name):
 310         annotations = {"kubernetes.io/service-account.name": self.service_account_name}
 311         secret_metadata = V1ObjectMeta(
 312             name=secret_name, namespace=self.namespace, annotations=annotations
 313         )
 314         secret_type = "kubernetes.io/service-account-token"
 315         secret = V1Secret(metadata=secret_metadata, type=secret_type)
 316         mock_create_secret.assert_called_once_with(self.namespace, secret)
 317
 318     def assert_create_service_account_v_1_24(
 319         self, mock_create_service_account, secret_name
 320     ):
 321         sevice_account_metadata = V1ObjectMeta(
 322             name=self.service_account_name, labels=self.labels, namespace=self.namespace
 323         )
 324         secrets = [V1SecretReference(name=secret_name, namespace=self.namespace)]
 325         service_account = V1ServiceAccount(
 326             metadata=sevice_account_metadata, secrets=secrets
 327         )
 328         mock_create_service_account.assert_called_once_with(
 329             self.namespace, service_account
 330         )
 331
 332     def assert_create_service_account_v_1_23(self, mock_create_service_account):
 333         metadata = V1ObjectMeta(
 334             name=self.service_account_name, labels=self.labels, namespace=self.namespace
 335         )
 336         service_account = V1ServiceAccount(metadata=metadata)
 337         mock_create_service_account.assert_called_once_with(
 338             self.namespace, service_account
 339         )
 340
 341     @mock.patch("n2vc.kubectl.uuid.uuid4")
 342     def test_secret_is_created_when_k8s_1_24(
 343         self,
 344         mock_uuid4,
 345         mock_list_service_account,
 346         mock_create_service_account,
 347         mock_create_secret,
 348         mock_list_secret,
 349         mock_version,
 350     ):
 351         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[])
 352         mock_list_secret.return_value = FakeK8sSecretList(items=[])
 353         mock_version.return_value = FakeK8sVersionApiCode("1", "24")
 354         mock_uuid4.return_value = self.token_id
 355         self.kubectl.create_service_account(
 356             self.service_account_name, self.labels, self.namespace
 357         )
 358         secret_name = "{}-token-{}".format(self.service_account_name, self.token_id[:5])
 359         self.assert_create_service_account_v_1_24(
 360             mock_create_service_account, secret_name
 361         )
 362         self.assert_create_secret(mock_create_secret, secret_name)
 363
 364     def test_secret_is_not_created_when_k8s_1_23(
 365         self,
 366         mock_list_service_account,
 367         mock_create_service_account,
 368         mock_create_secret,
 369         mock_list_secret,
 370         mock_version,
 371     ):
 372         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[])
 373         mock_version.return_value = FakeK8sVersionApiCode("1", "23+")
 374         self.kubectl.create_service_account(
 375             self.service_account_name, self.labels, self.namespace
 376         )
 377         self.assert_create_service_account_v_1_23(mock_create_service_account)
 378         mock_create_secret.assert_not_called()
 379         mock_list_secret.assert_not_called()
 380
 381     def test_raise_exception_if_service_account_already_exists(
 382         self,
 383         mock_list_service_account,
 384         mock_create_service_account,
 385         mock_create_secret,
 386         mock_list_secret,
 387         mock_version,
 388     ):
 389         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[1])
 390         with self.assertRaises(Exception) as context:
 391             self.kubectl.create_service_account(
 392                 self.service_account_name, self.labels, self.namespace
 393             )
 394         self.assertTrue(
 395             "Service account with metadata.name={} already exists".format(
 396                 self.service_account_name
 397             )
 398             in str(context.exception)
 399         )
 400         mock_create_service_account.assert_not_called()
 401         mock_create_secret.assert_not_called()
 402
 403     @mock.patch("n2vc.kubectl.uuid.uuid4")
 404     def test_raise_exception_if_secret_already_exists(
 405         self,
 406         mock_uuid4,
 407         mock_list_service_account,
 408         mock_create_service_account,
 409         mock_create_secret,
 410         mock_list_secret,
 411         mock_version,
 412     ):
 413         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[])
 414         mock_list_secret.return_value = FakeK8sSecretList(items=[1])
 415         mock_version.return_value = FakeK8sVersionApiCode("1", "24+")
 416         mock_uuid4.return_value = self.token_id
 417         with self.assertRaises(Exception) as context:
 418             self.kubectl.create_service_account(
 419                 self.service_account_name, self.labels, self.namespace
 420             )
 421         self.assertTrue(
 422             "Secret with metadata.name={}-token-{} already exists".format(
 423                 self.service_account_name, self.token_id[:5]
 424             )
 425             in str(context.exception)
 426         )
 427         mock_create_service_account.assert_called()
 428         mock_create_secret.assert_not_called()
 429
 430
 431 @mock.patch("kubernetes.client.CustomObjectsApi.create_namespaced_custom_object")
 432 class CreateCertificateClass(asynctest.TestCase):
 433     @mock.patch("kubernetes.config.load_kube_config")
 434     def setUp(self, mock_load_kube_config):
 435         super(CreateCertificateClass, self).setUp()
 436         self.namespace = "osm"
 437         self.name = "test-cert"
 438         self.dns_prefix = "*"
 439         self.secret_name = "test-cert-secret"
 440         self.usages = ["server auth"]
 441         self.issuer_name = "ca-issuer"
 442         self.kubectl = Kubectl()
 443
 444     @asynctest.fail_on(active_handles=True)
 445     async def test_certificate_is_created(
 446         self,
 447         mock_create_certificate,
 448     ):
 449         with open(
 450             os.path.join(
 451                 os.path.dirname(__file__), "testdata", "test_certificate.yaml"
 452             ),
 453             "r",
 454         ) as test_certificate:
 455             certificate_body = yaml.safe_load(test_certificate.read())
 456             print(certificate_body)
 457         await self.kubectl.create_certificate(
 458             namespace=self.namespace,
 459             name=self.name,
 460             dns_prefix=self.dns_prefix,
 461             secret_name=self.secret_name,
 462             usages=self.usages,
 463             issuer_name=self.issuer_name,
 464         )
 465         mock_create_certificate.assert_called_once_with(
 466             group="cert-manager.io",
 467             plural="certificates",
 468             version="v1",
 469             body=certificate_body,
 470             namespace=self.namespace,
 471         )
 472
 473     @asynctest.fail_on(active_handles=True)
 474     async def test_no_exception_if_alreadyexists(
 475         self,
 476         mock_create_certificate,
 477     ):
 478         api_exception = ApiException()
 479         api_exception.body = '{"reason": "AlreadyExists"}'
 480         self.kubectl.clients[
 481             CUSTOM_OBJECT_CLIENT
 482         ].create_namespaced_custom_object.side_effect = api_exception
 483         raised = False
 484         try:
 485             await self.kubectl.create_certificate(
 486                 namespace=self.namespace,
 487                 name=self.name,
 488                 dns_prefix=self.dns_prefix,
 489                 secret_name=self.secret_name,
 490                 usages=self.usages,
 491                 issuer_name=self.issuer_name,
 492             )
 493         except Exception:
 494             raised = True
 495         self.assertFalse(raised, "An exception was raised")
 496
 497     @asynctest.fail_on(active_handles=True)
 498     async def test_other_exceptions(
 499         self,
 500         mock_create_certificate,
 501     ):
 502         self.kubectl.clients[
 503             CUSTOM_OBJECT_CLIENT
 504         ].create_namespaced_custom_object.side_effect = Exception()
 505         with self.assertRaises(Exception):
 506             await self.kubectl.create_certificate(
 507                 namespace=self.namespace,
 508                 name=self.name,
 509                 dns_prefix=self.dns_prefix,
 510                 secret_name=self.secret_name,
 511                 usages=self.usages,
 512                 issuer_name=self.issuer_name,
 513             )
 514
 515
 516 @mock.patch("kubernetes.client.CustomObjectsApi.delete_namespaced_custom_object")
 517 class DeleteCertificateClass(asynctest.TestCase):
 518     @mock.patch("kubernetes.config.load_kube_config")
 519     def setUp(self, mock_load_kube_config):
 520         super(DeleteCertificateClass, self).setUp()
 521         self.namespace = "osm"
 522         self.object_name = "test-cert"
 523         self.kubectl = Kubectl()
 524
 525     @asynctest.fail_on(active_handles=True)
 526     async def test_no_exception_if_notfound(
 527         self,
 528         mock_create_certificate,
 529     ):
 530         api_exception = ApiException()
 531         api_exception.body = '{"reason": "NotFound"}'
 532         self.kubectl.clients[
 533             CUSTOM_OBJECT_CLIENT
 534         ].delete_namespaced_custom_object.side_effect = api_exception
 535         raised = False
 536         try:
 537             await self.kubectl.delete_certificate(
 538                 namespace=self.namespace,
 539                 object_name=self.object_name,
 540             )
 541         except Exception:
 542             raised = True
 543         self.assertFalse(raised, "An exception was raised")
 544
 545     @asynctest.fail_on(active_handles=True)
 546     async def test_other_exceptions(
 547         self,
 548         mock_create_certificate,
 549     ):
 550         self.kubectl.clients[
 551             CUSTOM_OBJECT_CLIENT
 552         ].delete_namespaced_custom_object.side_effect = Exception()
 553         with self.assertRaises(Exception):
 554             await self.kubectl.delete_certificate(
 555                 namespace=self.namespace,
 556                 object_name=self.object_name,
 557             )