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     V1Role,
  28     V1RoleBinding,
  29     V1RoleRef,
  30     V1Subject,
  31     V1PolicyRule,
  32     V1Namespace,
  33 )
  34
  35
  36 class FakeK8sResourceMetadata:
  37     def __init__(
  38         self,
  39         name: str = None,
  40         namespace: str = None,
  41         annotations: dict = {},
  42         labels: dict = {},
  43     ):
  44         self._annotations = annotations
  45         self._name = name or "name"
  46         self._namespace = namespace or "namespace"
  47         self._labels = labels or {"juju-app": "squid"}
  48
  49     @property
  50     def name(self):
  51         return self._name
  52
  53     @property
  54     def namespace(self):
  55         return self._namespace
  56
  57     @property
  58     def labels(self):
  59         return self._labels
  60
  61     @property
  62     def annotations(self):
  63         return self._annotations
  64
  65
  66 class FakeK8sStorageClass:
  67     def __init__(self, metadata=None):
  68         self._metadata = metadata or FakeK8sResourceMetadata()
  69
  70     @property
  71     def metadata(self):
  72         return self._metadata
  73
  74
  75 class FakeK8sStorageClassesList:
  76     def __init__(self, items=[]):
  77         self._items = items
  78
  79     @property
  80     def items(self):
  81         return self._items
  82
  83
  84 class FakeK8sServiceAccountsList:
  85     def __init__(self, items=[]):
  86         self._items = items
  87
  88     @property
  89     def items(self):
  90         return self._items
  91
  92
  93 class FakeK8sSecretList:
  94     def __init__(self, items=[]):
  95         self._items = items
  96
  97     @property
  98     def items(self):
  99         return self._items
 100
 101
 102 class FakeK8sRoleList:
 103     def __init__(self, items=[]):
 104         self._items = items
 105
 106     @property
 107     def items(self):
 108         return self._items
 109
 110
 111 class FakeK8sRoleBindingList:
 112     def __init__(self, items=[]):
 113         self._items = items
 114
 115     @property
 116     def items(self):
 117         return self._items
 118
 119
 120 class FakeK8sVersionApiCode:
 121     def __init__(self, major: str, minor: str):
 122         self._major = major
 123         self._minor = minor
 124
 125     @property
 126     def major(self):
 127         return self._major
 128
 129     @property
 130     def minor(self):
 131         return self._minor
 132
 133
 134 fake_list_services = Dict(
 135     {
 136         "items": [
 137             Dict(
 138                 {
 139                     "metadata": Dict(
 140                         {
 141                             "name": "squid",
 142                             "namespace": "test",
 143                             "labels": {"juju-app": "squid"},
 144                         }
 145                     ),
 146                     "spec": Dict(
 147                         {
 148                             "cluster_ip": "10.152.183.79",
 149                             "type": "LoadBalancer",
 150                             "ports": [
 151                                 Dict(
 152                                     {
 153                                         "name": None,
 154                                         "node_port": None,
 155                                         "port": 30666,
 156                                         "protocol": "TCP",
 157                                         "target_port": 30666,
 158                                     }
 159                                 )
 160                             ],
 161                         }
 162                     ),
 163                     "status": Dict(
 164                         {
 165                             "load_balancer": Dict(
 166                                 {
 167                                     "ingress": [
 168                                         Dict({"hostname": None, "ip": "192.168.0.201"})
 169                                     ]
 170                                 }
 171                             )
 172                         }
 173                     ),
 174                 }
 175             )
 176         ]
 177     }
 178 )
 179
 180
 181 class KubectlTestCase(TestCase):
 182     def setUp(
 183         self,
 184     ):
 185         pass
 186
 187
 188 class FakeCoreV1Api:
 189     def list_service_for_all_namespaces(self, **kwargs):
 190         return fake_list_services
 191
 192
 193 class GetServices(TestCase):
 194     @mock.patch("n2vc.kubectl.config.load_kube_config")
 195     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 196     def setUp(self, mock_core, mock_config):
 197         mock_core.return_value = mock.MagicMock()
 198         mock_config.return_value = mock.MagicMock()
 199         self.kubectl = Kubectl()
 200
 201     @mock.patch("n2vc.kubectl.client.CoreV1Api")
 202     def test_get_service(self, mock_corev1api):
 203         mock_corev1api.return_value = FakeCoreV1Api()
 204         services = self.kubectl.get_services(
 205             field_selector="metadata.namespace", label_selector="juju-operator=squid"
 206         )
 207         keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
 208         self.assertTrue(k in service for service in services for k in keys)
 209
 210     def test_get_service_exception(self):
 211         self.kubectl.clients[
 212             CORE_CLIENT
 213         ].list_service_for_all_namespaces.side_effect = ApiException()
 214         with self.assertRaises(ApiException):
 215             self.kubectl.get_services()
 216
 217
 218 @mock.patch("n2vc.kubectl.client")
 219 @mock.patch("n2vc.kubectl.config.kube_config.Configuration.get_default_copy")
 220 @mock.patch("n2vc.kubectl.config.load_kube_config")
 221 class GetConfiguration(KubectlTestCase):
 222     def setUp(self):
 223         super(GetConfiguration, self).setUp()
 224
 225     def test_get_configuration(
 226         self,
 227         mock_load_kube_config,
 228         mock_configuration,
 229         mock_client,
 230     ):
 231         kubectl = Kubectl()
 232         kubectl.configuration
 233         mock_configuration.assert_called_once()
 234         mock_load_kube_config.assert_called_once()
 235         mock_client.CoreV1Api.assert_called_once()
 236         mock_client.RbacAuthorizationV1Api.assert_called_once()
 237         mock_client.StorageV1Api.assert_called_once()
 238
 239
 240 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
 241 @mock.patch("kubernetes.config.load_kube_config")
 242 class GetDefaultStorageClass(KubectlTestCase):
 243     def setUp(self):
 244         super(GetDefaultStorageClass, self).setUp()
 245
 246         # Default Storage Class
 247         self.default_sc_name = "default-sc"
 248         default_sc_metadata = FakeK8sResourceMetadata(
 249             name=self.default_sc_name,
 250             annotations={"storageclass.kubernetes.io/is-default-class": "true"},
 251         )
 252         self.default_sc = FakeK8sStorageClass(metadata=default_sc_metadata)
 253
 254         # Default Storage Class with old annotation
 255         self.default_sc_old_name = "default-sc-old"
 256         default_sc_old_metadata = FakeK8sResourceMetadata(
 257             name=self.default_sc_old_name,
 258             annotations={"storageclass.beta.kubernetes.io/is-default-class": "true"},
 259         )
 260         self.default_sc_old = FakeK8sStorageClass(metadata=default_sc_old_metadata)
 261
 262         # Storage class - not default
 263         self.sc_name = "default-sc-old"
 264         self.sc = FakeK8sStorageClass(
 265             metadata=FakeK8sResourceMetadata(name=self.sc_name)
 266         )
 267
 268     def test_get_default_storage_class_exists_default(
 269         self, mock_load_kube_config, mock_list_storage_class
 270     ):
 271         kubectl = Kubectl()
 272         items = [self.default_sc]
 273         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 274         sc_name = kubectl.get_default_storage_class()
 275         self.assertEqual(sc_name, self.default_sc_name)
 276         mock_list_storage_class.assert_called_once()
 277
 278     def test_get_default_storage_class_exists_default_old(
 279         self, mock_load_kube_config, mock_list_storage_class
 280     ):
 281         kubectl = Kubectl()
 282         items = [self.default_sc_old]
 283         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 284         sc_name = kubectl.get_default_storage_class()
 285         self.assertEqual(sc_name, self.default_sc_old_name)
 286         mock_list_storage_class.assert_called_once()
 287
 288     def test_get_default_storage_class_none(
 289         self, mock_load_kube_config, mock_list_storage_class
 290     ):
 291         kubectl = Kubectl()
 292         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=[])
 293         sc_name = kubectl.get_default_storage_class()
 294         self.assertEqual(sc_name, None)
 295         mock_list_storage_class.assert_called_once()
 296
 297     def test_get_default_storage_class_exists_not_default(
 298         self, mock_load_kube_config, mock_list_storage_class
 299     ):
 300         kubectl = Kubectl()
 301         items = [self.sc]
 302         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 303         sc_name = kubectl.get_default_storage_class()
 304         self.assertEqual(sc_name, self.sc_name)
 305         mock_list_storage_class.assert_called_once()
 306
 307     def test_get_default_storage_class_choose(
 308         self, mock_load_kube_config, mock_list_storage_class
 309     ):
 310         kubectl = Kubectl()
 311         items = [self.sc, self.default_sc]
 312         mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
 313         sc_name = kubectl.get_default_storage_class()
 314         self.assertEqual(sc_name, self.default_sc_name)
 315         mock_list_storage_class.assert_called_once()
 316
 317
 318 @mock.patch("kubernetes.client.VersionApi.get_code")
 319 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_secret")
 320 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_secret")
 321 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_service_account")
 322 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_service_account")
 323 class CreateServiceAccountClass(KubectlTestCase):
 324     @mock.patch("kubernetes.config.load_kube_config")
 325     def setUp(self, mock_load_kube_config):
 326         super(CreateServiceAccountClass, self).setUp()
 327         self.service_account_name = "Service_account"
 328         self.labels = {"Key1": "Value1", "Key2": "Value2"}
 329         self.namespace = "kubernetes"
 330         self.token_id = "abc12345"
 331         self.kubectl = Kubectl()
 332
 333     def assert_create_secret(self, mock_create_secret, secret_name):
 334         annotations = {"kubernetes.io/service-account.name": self.service_account_name}
 335         secret_metadata = V1ObjectMeta(
 336             name=secret_name, namespace=self.namespace, annotations=annotations
 337         )
 338         secret_type = "kubernetes.io/service-account-token"
 339         secret = V1Secret(metadata=secret_metadata, type=secret_type)
 340         mock_create_secret.assert_called_once_with(self.namespace, secret)
 341
 342     def assert_create_service_account_v_1_24(
 343         self, mock_create_service_account, secret_name
 344     ):
 345         sevice_account_metadata = V1ObjectMeta(
 346             name=self.service_account_name, labels=self.labels, namespace=self.namespace
 347         )
 348         secrets = [V1SecretReference(name=secret_name, namespace=self.namespace)]
 349         service_account = V1ServiceAccount(
 350             metadata=sevice_account_metadata, secrets=secrets
 351         )
 352         mock_create_service_account.assert_called_once_with(
 353             self.namespace, service_account
 354         )
 355
 356     def assert_create_service_account_v_1_23(self, mock_create_service_account):
 357         metadata = V1ObjectMeta(
 358             name=self.service_account_name, labels=self.labels, namespace=self.namespace
 359         )
 360         service_account = V1ServiceAccount(metadata=metadata)
 361         mock_create_service_account.assert_called_once_with(
 362             self.namespace, service_account
 363         )
 364
 365     @mock.patch("n2vc.kubectl.uuid.uuid4")
 366     def test_secret_is_created_when_k8s_1_24(
 367         self,
 368         mock_uuid4,
 369         mock_list_service_account,
 370         mock_create_service_account,
 371         mock_create_secret,
 372         mock_list_secret,
 373         mock_version,
 374     ):
 375         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[])
 376         mock_list_secret.return_value = FakeK8sSecretList(items=[])
 377         mock_version.return_value = FakeK8sVersionApiCode("1", "24")
 378         mock_uuid4.return_value = self.token_id
 379         self.kubectl.create_service_account(
 380             self.service_account_name, self.labels, self.namespace
 381         )
 382         secret_name = "{}-token-{}".format(self.service_account_name, self.token_id[:5])
 383         self.assert_create_service_account_v_1_24(
 384             mock_create_service_account, secret_name
 385         )
 386         self.assert_create_secret(mock_create_secret, secret_name)
 387
 388     def test_secret_is_not_created_when_k8s_1_23(
 389         self,
 390         mock_list_service_account,
 391         mock_create_service_account,
 392         mock_create_secret,
 393         mock_list_secret,
 394         mock_version,
 395     ):
 396         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[])
 397         mock_version.return_value = FakeK8sVersionApiCode("1", "23+")
 398         self.kubectl.create_service_account(
 399             self.service_account_name, self.labels, self.namespace
 400         )
 401         self.assert_create_service_account_v_1_23(mock_create_service_account)
 402         mock_create_secret.assert_not_called()
 403         mock_list_secret.assert_not_called()
 404
 405     def test_raise_exception_if_service_account_already_exists(
 406         self,
 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=[1])
 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             "Service account with metadata.name={} already exists".format(
 420                 self.service_account_name
 421             )
 422             in str(context.exception)
 423         )
 424         mock_create_service_account.assert_not_called()
 425         mock_create_secret.assert_not_called()
 426
 427     @mock.patch("n2vc.kubectl.uuid.uuid4")
 428     def test_raise_exception_if_secret_already_exists(
 429         self,
 430         mock_uuid4,
 431         mock_list_service_account,
 432         mock_create_service_account,
 433         mock_create_secret,
 434         mock_list_secret,
 435         mock_version,
 436     ):
 437         mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[])
 438         mock_list_secret.return_value = FakeK8sSecretList(items=[1])
 439         mock_version.return_value = FakeK8sVersionApiCode("1", "24+")
 440         mock_uuid4.return_value = self.token_id
 441         with self.assertRaises(Exception) as context:
 442             self.kubectl.create_service_account(
 443                 self.service_account_name, self.labels, self.namespace
 444             )
 445         self.assertTrue(
 446             "Secret with metadata.name={}-token-{} already exists".format(
 447                 self.service_account_name, self.token_id[:5]
 448             )
 449             in str(context.exception)
 450         )
 451         mock_create_service_account.assert_called()
 452         mock_create_secret.assert_not_called()
 453
 454
 455 @mock.patch("kubernetes.client.CustomObjectsApi.create_namespaced_custom_object")
 456 class CreateCertificateClass(asynctest.TestCase):
 457     @mock.patch("kubernetes.config.load_kube_config")
 458     def setUp(self, mock_load_kube_config):
 459         super(CreateCertificateClass, self).setUp()
 460         self.namespace = "osm"
 461         self.name = "test-cert"
 462         self.dns_prefix = "*"
 463         self.secret_name = "test-cert-secret"
 464         self.usages = ["server auth"]
 465         self.issuer_name = "ca-issuer"
 466         self.kubectl = Kubectl()
 467
 468     @asynctest.fail_on(active_handles=True)
 469     async def test_certificate_is_created(
 470         self,
 471         mock_create_certificate,
 472     ):
 473         with open(
 474             os.path.join(
 475                 os.path.dirname(__file__), "testdata", "test_certificate.yaml"
 476             ),
 477             "r",
 478         ) as test_certificate:
 479             certificate_body = yaml.safe_load(test_certificate.read())
 480             print(certificate_body)
 481         await self.kubectl.create_certificate(
 482             namespace=self.namespace,
 483             name=self.name,
 484             dns_prefix=self.dns_prefix,
 485             secret_name=self.secret_name,
 486             usages=self.usages,
 487             issuer_name=self.issuer_name,
 488         )
 489         mock_create_certificate.assert_called_once_with(
 490             group="cert-manager.io",
 491             plural="certificates",
 492             version="v1",
 493             body=certificate_body,
 494             namespace=self.namespace,
 495         )
 496
 497     @asynctest.fail_on(active_handles=True)
 498     async def test_no_exception_if_alreadyexists(
 499         self,
 500         mock_create_certificate,
 501     ):
 502         api_exception = ApiException()
 503         api_exception.body = '{"reason": "AlreadyExists"}'
 504         self.kubectl.clients[
 505             CUSTOM_OBJECT_CLIENT
 506         ].create_namespaced_custom_object.side_effect = api_exception
 507         raised = False
 508         try:
 509             await self.kubectl.create_certificate(
 510                 namespace=self.namespace,
 511                 name=self.name,
 512                 dns_prefix=self.dns_prefix,
 513                 secret_name=self.secret_name,
 514                 usages=self.usages,
 515                 issuer_name=self.issuer_name,
 516             )
 517         except Exception:
 518             raised = True
 519         self.assertFalse(raised, "An exception was raised")
 520
 521     @asynctest.fail_on(active_handles=True)
 522     async def test_other_exceptions(
 523         self,
 524         mock_create_certificate,
 525     ):
 526         self.kubectl.clients[
 527             CUSTOM_OBJECT_CLIENT
 528         ].create_namespaced_custom_object.side_effect = Exception()
 529         with self.assertRaises(Exception):
 530             await self.kubectl.create_certificate(
 531                 namespace=self.namespace,
 532                 name=self.name,
 533                 dns_prefix=self.dns_prefix,
 534                 secret_name=self.secret_name,
 535                 usages=self.usages,
 536                 issuer_name=self.issuer_name,
 537             )
 538
 539
 540 @mock.patch("kubernetes.client.CustomObjectsApi.delete_namespaced_custom_object")
 541 class DeleteCertificateClass(asynctest.TestCase):
 542     @mock.patch("kubernetes.config.load_kube_config")
 543     def setUp(self, mock_load_kube_config):
 544         super(DeleteCertificateClass, self).setUp()
 545         self.namespace = "osm"
 546         self.object_name = "test-cert"
 547         self.kubectl = Kubectl()
 548
 549     @asynctest.fail_on(active_handles=True)
 550     async def test_no_exception_if_notfound(
 551         self,
 552         mock_create_certificate,
 553     ):
 554         api_exception = ApiException()
 555         api_exception.body = '{"reason": "NotFound"}'
 556         self.kubectl.clients[
 557             CUSTOM_OBJECT_CLIENT
 558         ].delete_namespaced_custom_object.side_effect = api_exception
 559         raised = False
 560         try:
 561             await self.kubectl.delete_certificate(
 562                 namespace=self.namespace,
 563                 object_name=self.object_name,
 564             )
 565         except Exception:
 566             raised = True
 567         self.assertFalse(raised, "An exception was raised")
 568
 569     @asynctest.fail_on(active_handles=True)
 570     async def test_other_exceptions(
 571         self,
 572         mock_create_certificate,
 573     ):
 574         self.kubectl.clients[
 575             CUSTOM_OBJECT_CLIENT
 576         ].delete_namespaced_custom_object.side_effect = Exception()
 577         with self.assertRaises(Exception):
 578             await self.kubectl.delete_certificate(
 579                 namespace=self.namespace,
 580                 object_name=self.object_name,
 581             )
 582
 583
 584 @mock.patch("kubernetes.client.RbacAuthorizationV1Api.create_namespaced_role")
 585 @mock.patch("kubernetes.client.RbacAuthorizationV1Api.list_namespaced_role")
 586 class CreateRoleClass(asynctest.TestCase):
 587     @mock.patch("kubernetes.config.load_kube_config")
 588     def setUp(self, mock_load_kube_config):
 589         super(CreateRoleClass, self).setUp()
 590         self.name = "role"
 591         self.namespace = "osm"
 592         self.resources = ["*"]
 593         self.api_groups = ["*"]
 594         self.verbs = ["*"]
 595         self.labels = {}
 596         self.kubectl = Kubectl()
 597
 598     @asynctest.fail_on(active_handles=True)
 599     async def assert_create_role(self, mock_create_role):
 600         metadata = V1ObjectMeta(
 601             name=self.name, labels=self.labels, namespace=self.namespace
 602         )
 603         role = V1Role(
 604             metadata=metadata,
 605             rules=[
 606                 V1PolicyRule(
 607                     api_groups=self.api_groups,
 608                     resources=self.resources,
 609                     verbs=self.verbs,
 610                 ),
 611             ],
 612         )
 613         await self.kubectl.create_role(
 614             namespace=self.namespace,
 615             api_groups=self.api_groups,
 616             name=self.name,
 617             resources=self.resources,
 618             verbs=self.verbs,
 619             labels=self.labels,
 620         )
 621         mock_create_role.assert_called_once_with(self.namespace, role)
 622
 623     @asynctest.fail_on(active_handles=True)
 624     async def test_raise_exception_if_role_already_exists(
 625         self,
 626         mock_list_role,
 627         mock_create_role,
 628     ):
 629         mock_list_role.return_value = FakeK8sRoleList(items=[1])
 630         with self.assertRaises(Exception) as context:
 631             await self.kubectl.create_role(
 632                 self.name,
 633                 self.labels,
 634                 self.api_groups,
 635                 self.resources,
 636                 self.verbs,
 637                 self.namespace,
 638             )
 639         self.assertTrue(
 640             "Role with metadata.name={} already exists".format(self.name)
 641             in str(context.exception)
 642         )
 643         mock_create_role.assert_not_called()
 644
 645
 646 @mock.patch("kubernetes.client.RbacAuthorizationV1Api.create_namespaced_role_binding")
 647 @mock.patch("kubernetes.client.RbacAuthorizationV1Api.list_namespaced_role_binding")
 648 class CreateRoleBindingClass(asynctest.TestCase):
 649     @mock.patch("kubernetes.config.load_kube_config")
 650     def setUp(self, mock_load_kube_config):
 651         super(CreateRoleBindingClass, self).setUp()
 652         self.name = "rolebinding"
 653         self.namespace = "osm"
 654         self.role_name = "role"
 655         self.sa_name = "Default"
 656         self.labels = {}
 657         self.kubectl = Kubectl()
 658
 659     @asynctest.fail_on(active_handles=True)
 660     async def assert_create_role_binding(self, mock_create_role_binding):
 661         role_binding = V1RoleBinding(
 662             metadata=V1ObjectMeta(name=self.name, labels=self.labels),
 663             role_ref=V1RoleRef(kind="Role", name=self.role_name, api_group=""),
 664             subjects=[
 665                 V1Subject(
 666                     kind="ServiceAccount",
 667                     name=self.sa_name,
 668                     namespace=self.namespace,
 669                 )
 670             ],
 671         )
 672         await self.kubectl.create_role_binding(
 673             namespace=self.namespace,
 674             role_name=self.role_name,
 675             name=self.name,
 676             sa_name=self.sa_name,
 677             labels=self.labels,
 678         )
 679         mock_create_role_binding.assert_called_once_with(self.namespace, role_binding)
 680
 681     @asynctest.fail_on(active_handles=True)
 682     async def test_raise_exception_if_role_binding_already_exists(
 683         self,
 684         mock_list_role_binding,
 685         mock_create_role_binding,
 686     ):
 687         mock_list_role_binding.return_value = FakeK8sRoleBindingList(items=[1])
 688         with self.assertRaises(Exception) as context:
 689             await self.kubectl.create_role_binding(
 690                 self.name,
 691                 self.role_name,
 692                 self.sa_name,
 693                 self.labels,
 694                 self.namespace,
 695             )
 696         self.assertTrue(
 697             "Role Binding with metadata.name={} already exists".format(self.name)
 698             in str(context.exception)
 699         )
 700         mock_create_role_binding.assert_not_called()
 701
 702
 703 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_secret")
 704 class CreateSecretClass(asynctest.TestCase):
 705     @mock.patch("kubernetes.config.load_kube_config")
 706     def setUp(self, mock_load_kube_config):
 707         super(CreateSecretClass, self).setUp()
 708         self.name = "secret"
 709         self.namespace = "osm"
 710         self.data = {"test": "1234"}
 711         self.secret_type = "Opaque"
 712         self.kubectl = Kubectl()
 713
 714     @asynctest.fail_on(active_handles=True)
 715     async def assert_create_secret(self, mock_create_secret):
 716         secret_metadata = V1ObjectMeta(name=self.name, namespace=self.namespace)
 717         secret = V1Secret(
 718             metadata=secret_metadata,
 719             data=self.data,
 720             type=self.secret_type,
 721         )
 722         await self.kubectl.create_secret(
 723             namespace=self.namespace,
 724             data=self.data,
 725             name=self.name,
 726             secret_type=self.secret_type,
 727         )
 728         mock_create_secret.assert_called_once_with(self.namespace, secret)
 729
 730
 731 @mock.patch("kubernetes.client.CoreV1Api.create_namespace")
 732 class CreateNamespaceClass(asynctest.TestCase):
 733     @mock.patch("kubernetes.config.load_kube_config")
 734     def setUp(self, mock_load_kube_config):
 735         super(CreateNamespaceClass, self).setUp()
 736         self.namespace = "osm"
 737         self.labels = {"key": "value"}
 738         self.kubectl = Kubectl()
 739
 740     @asynctest.fail_on(active_handles=True)
 741     async def test_namespace_is_created(
 742         self,
 743         mock_create_namespace,
 744     ):
 745         metadata = V1ObjectMeta(name=self.namespace, labels=self.labels)
 746         namespace = V1Namespace(
 747             metadata=metadata,
 748         )
 749         await self.kubectl.create_namespace(
 750             name=self.namespace,
 751             labels=self.labels,
 752         )
 753         mock_create_namespace.assert_called_once_with(namespace)
 754
 755     async def test_namespace_is_created_default_labels(
 756         self,
 757         mock_create_namespace,
 758     ):
 759         metadata = V1ObjectMeta(name=self.namespace, labels=None)
 760         namespace = V1Namespace(
 761             metadata=metadata,
 762         )
 763         await self.kubectl.create_namespace(
 764             name=self.namespace,
 765         )
 766         mock_create_namespace.assert_called_once_with(namespace)
 767
 768     @asynctest.fail_on(active_handles=True)
 769     async def test_no_exception_if_alreadyexists(
 770         self,
 771         mock_create_namespace,
 772     ):
 773         api_exception = ApiException()
 774         api_exception.body = '{"reason": "AlreadyExists"}'
 775         self.kubectl.clients[CORE_CLIENT].create_namespace.side_effect = api_exception
 776         raised = False
 777         try:
 778             await self.kubectl.create_namespace(
 779                 name=self.namespace,
 780             )
 781         except Exception:
 782             raised = True
 783         self.assertFalse(raised, "An exception was raised")
 784
 785     @asynctest.fail_on(active_handles=True)
 786     async def test_other_exceptions(
 787         self,
 788         mock_create_namespace,
 789     ):
 790         self.kubectl.clients[CORE_CLIENT].create_namespace.side_effect = Exception()
 791         with self.assertRaises(Exception):
 792             await self.kubectl.create_namespace(
 793                 name=self.namespace,
 794             )
 795
 796
 797 @mock.patch("kubernetes.client.CoreV1Api.delete_namespace")
 798 class DeleteNamespaceClass(asynctest.TestCase):
 799     @mock.patch("kubernetes.config.load_kube_config")
 800     def setUp(self, mock_load_kube_config):
 801         super(DeleteNamespaceClass, self).setUp()
 802         self.namespace = "osm"
 803         self.kubectl = Kubectl()
 804
 805     @asynctest.fail_on(active_handles=True)
 806     async def test_no_exception_if_notfound(
 807         self,
 808         mock_delete_namespace,
 809     ):
 810         api_exception = ApiException()
 811         api_exception.body = '{"reason": "NotFound"}'
 812         self.kubectl.clients[CORE_CLIENT].delete_namespace.side_effect = api_exception
 813         raised = False
 814         try:
 815             await self.kubectl.delete_namespace(
 816                 name=self.namespace,
 817             )
 818         except Exception:
 819             raised = True
 820         self.assertFalse(raised, "An exception was raised")
 821
 822     @asynctest.fail_on(active_handles=True)
 823     async def test_other_exceptions(
 824         self,
 825         mock_delete_namespace,
 826     ):
 827         self.kubectl.clients[CORE_CLIENT].delete_namespace.side_effect = Exception()
 828         with self.assertRaises(Exception):
 829             await self.kubectl.delete_namespace(
 830                 name=self.namespace,
 831             )
 832
 833
 834 @mock.patch("kubernetes.client.CoreV1Api.read_namespaced_secret")
 835 class GetSecretContentClass(asynctest.TestCase):
 836     @mock.patch("kubernetes.config.load_kube_config")
 837     def setUp(self, mock_load_kube_config):
 838         super(GetSecretContentClass, self).setUp()
 839         self.name = "my_secret"
 840         self.namespace = "osm"
 841         self.data = {"my_key": "my_value"}
 842         self.type = "Opaque"
 843         self.kubectl = Kubectl()
 844
 845     @asynctest.fail_on(active_handles=True)
 846     async def test_return_type_is_dict(
 847         self,
 848         mock_read_namespaced_secret,
 849     ):
 850         metadata = V1ObjectMeta(name=self.name, namespace=self.namespace)
 851         secret = V1Secret(metadata=metadata, data=self.data, type=self.type)
 852         mock_read_namespaced_secret.return_value = secret
 853         content = await self.kubectl.get_secret_content(self.name, self.namespace)
 854         assert type(content) is dict