1 # Copyright 2020 Canonical Ltd.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
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 (
36 class FakeK8sResourceMetadata
:
40 namespace
: str = None,
41 annotations
: dict = {},
44 self
._annotations
= annotations
45 self
._name
= name
or "name"
46 self
._namespace
= namespace
or "namespace"
47 self
._labels
= labels
or {"juju-app": "squid"}
55 return self
._namespace
62 def annotations(self
):
63 return self
._annotations
66 class FakeK8sStorageClass
:
67 def __init__(self
, metadata
=None):
68 self
._metadata
= metadata
or FakeK8sResourceMetadata()
75 class FakeK8sStorageClassesList
:
76 def __init__(self
, items
=[]):
84 class FakeK8sServiceAccountsList
:
85 def __init__(self
, items
=[]):
93 class FakeK8sSecretList
:
94 def __init__(self
, items
=[]):
102 class FakeK8sRoleList
:
103 def __init__(self
, items
=[]):
111 class FakeK8sRoleBindingList
:
112 def __init__(self
, items
=[]):
120 class FakeK8sVersionApiCode
:
121 def __init__(self
, major
: str, minor
: str):
134 fake_list_services
= Dict(
143 "labels": {"juju-app": "squid"},
148 "cluster_ip": "10.152.183.79",
149 "type": "LoadBalancer",
157 "target_port": 30666,
165 "load_balancer": Dict(
168 Dict({"hostname": None, "ip": "192.168.0.201"})
181 class KubectlTestCase(TestCase
):
189 def list_service_for_all_namespaces(self
, **kwargs
):
190 return fake_list_services
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()
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"
207 keys
= ["name", "cluster_ip", "type", "ports", "external_ip"]
208 self
.assertTrue(k
in service
for service
in services
for k
in keys
)
210 def test_get_service_exception(self
):
211 self
.kubectl
.clients
[
213 ].list_service_for_all_namespaces
.side_effect
= ApiException()
214 with self
.assertRaises(ApiException
):
215 self
.kubectl
.get_services()
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
):
223 super(GetConfiguration
, self
).setUp()
225 def test_get_configuration(
227 mock_load_kube_config
,
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()
240 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
241 @mock.patch("kubernetes.config.load_kube_config")
242 class GetDefaultStorageClass(KubectlTestCase
):
244 super(GetDefaultStorageClass
, self
).setUp()
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"},
252 self
.default_sc
= FakeK8sStorageClass(metadata
=default_sc_metadata
)
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"},
260 self
.default_sc_old
= FakeK8sStorageClass(metadata
=default_sc_old_metadata
)
262 # Storage class - not default
263 self
.sc_name
= "default-sc-old"
264 self
.sc
= FakeK8sStorageClass(
265 metadata
=FakeK8sResourceMetadata(name
=self
.sc_name
)
268 def test_get_default_storage_class_exists_default(
269 self
, mock_load_kube_config
, mock_list_storage_class
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()
278 def test_get_default_storage_class_exists_default_old(
279 self
, mock_load_kube_config
, mock_list_storage_class
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()
288 def test_get_default_storage_class_none(
289 self
, mock_load_kube_config
, mock_list_storage_class
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()
297 def test_get_default_storage_class_exists_not_default(
298 self
, mock_load_kube_config
, mock_list_storage_class
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()
307 def test_get_default_storage_class_choose(
308 self
, mock_load_kube_config
, mock_list_storage_class
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()
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()
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
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
)
342 def assert_create_service_account_v_1_24(
343 self
, mock_create_service_account
, secret_name
345 sevice_account_metadata
= V1ObjectMeta(
346 name
=self
.service_account_name
, labels
=self
.labels
, namespace
=self
.namespace
348 secrets
= [V1SecretReference(name
=secret_name
, namespace
=self
.namespace
)]
349 service_account
= V1ServiceAccount(
350 metadata
=sevice_account_metadata
, secrets
=secrets
352 mock_create_service_account
.assert_called_once_with(
353 self
.namespace
, service_account
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
360 service_account
= V1ServiceAccount(metadata
=metadata
)
361 mock_create_service_account
.assert_called_once_with(
362 self
.namespace
, service_account
365 @mock.patch("n2vc.kubectl.uuid.uuid4")
366 def test_secret_is_created_when_k8s_1_24(
369 mock_list_service_account
,
370 mock_create_service_account
,
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
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
386 self
.assert_create_secret(mock_create_secret
, secret_name
)
388 def test_secret_is_not_created_when_k8s_1_23(
390 mock_list_service_account
,
391 mock_create_service_account
,
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
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()
405 def test_raise_exception_if_service_account_already_exists(
407 mock_list_service_account
,
408 mock_create_service_account
,
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
419 "Service account with metadata.name={} already exists".format(
420 self
.service_account_name
422 in str(context
.exception
)
424 mock_create_service_account
.assert_not_called()
425 mock_create_secret
.assert_not_called()
427 @mock.patch("n2vc.kubectl.uuid.uuid4")
428 def test_raise_exception_if_secret_already_exists(
431 mock_list_service_account
,
432 mock_create_service_account
,
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
446 "Secret with metadata.name={}-token-{} already exists".format(
447 self
.service_account_name
, self
.token_id
[:5]
449 in str(context
.exception
)
451 mock_create_service_account
.assert_called()
452 mock_create_secret
.assert_not_called()
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()
468 @asynctest.fail_on(active_handles
=True)
469 async def test_certificate_is_created(
471 mock_create_certificate
,
475 os
.path
.dirname(__file__
), "testdata", "test_certificate.yaml"
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
,
484 dns_prefix
=self
.dns_prefix
,
485 secret_name
=self
.secret_name
,
487 issuer_name
=self
.issuer_name
,
489 mock_create_certificate
.assert_called_once_with(
490 group
="cert-manager.io",
491 plural
="certificates",
493 body
=certificate_body
,
494 namespace
=self
.namespace
,
497 @asynctest.fail_on(active_handles
=True)
498 async def test_no_exception_if_alreadyexists(
500 mock_create_certificate
,
502 api_exception
= ApiException()
503 api_exception
.body
= '{"reason": "AlreadyExists"}'
504 self
.kubectl
.clients
[
506 ].create_namespaced_custom_object
.side_effect
= api_exception
509 await self
.kubectl
.create_certificate(
510 namespace
=self
.namespace
,
512 dns_prefix
=self
.dns_prefix
,
513 secret_name
=self
.secret_name
,
515 issuer_name
=self
.issuer_name
,
519 self
.assertFalse(raised
, "An exception was raised")
521 @asynctest.fail_on(active_handles
=True)
522 async def test_other_exceptions(
524 mock_create_certificate
,
526 self
.kubectl
.clients
[
528 ].create_namespaced_custom_object
.side_effect
= Exception()
529 with self
.assertRaises(Exception):
530 await self
.kubectl
.create_certificate(
531 namespace
=self
.namespace
,
533 dns_prefix
=self
.dns_prefix
,
534 secret_name
=self
.secret_name
,
536 issuer_name
=self
.issuer_name
,
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()
549 @asynctest.fail_on(active_handles
=True)
550 async def test_no_exception_if_notfound(
552 mock_create_certificate
,
554 api_exception
= ApiException()
555 api_exception
.body
= '{"reason": "NotFound"}'
556 self
.kubectl
.clients
[
558 ].delete_namespaced_custom_object
.side_effect
= api_exception
561 await self
.kubectl
.delete_certificate(
562 namespace
=self
.namespace
,
563 object_name
=self
.object_name
,
567 self
.assertFalse(raised
, "An exception was raised")
569 @asynctest.fail_on(active_handles
=True)
570 async def test_other_exceptions(
572 mock_create_certificate
,
574 self
.kubectl
.clients
[
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
,
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()
591 self
.namespace
= "osm"
592 self
.resources
= ["*"]
593 self
.api_groups
= ["*"]
596 self
.kubectl
= Kubectl()
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
607 api_groups
=self
.api_groups
,
608 resources
=self
.resources
,
613 await self
.kubectl
.create_role(
614 namespace
=self
.namespace
,
615 api_groups
=self
.api_groups
,
617 resources
=self
.resources
,
621 mock_create_role
.assert_called_once_with(self
.namespace
, role
)
623 @asynctest.fail_on(active_handles
=True)
624 async def test_raise_exception_if_role_already_exists(
629 mock_list_role
.return_value
= FakeK8sRoleList(items
=[1])
630 with self
.assertRaises(Exception) as context
:
631 await self
.kubectl
.create_role(
640 "Role with metadata.name={} already exists".format(self
.name
)
641 in str(context
.exception
)
643 mock_create_role
.assert_not_called()
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"
657 self
.kubectl
= Kubectl()
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
=""),
666 kind
="ServiceAccount",
668 namespace
=self
.namespace
,
672 await self
.kubectl
.create_role_binding(
673 namespace
=self
.namespace
,
674 role_name
=self
.role_name
,
676 sa_name
=self
.sa_name
,
679 mock_create_role_binding
.assert_called_once_with(self
.namespace
, role_binding
)
681 @asynctest.fail_on(active_handles
=True)
682 async def test_raise_exception_if_role_binding_already_exists(
684 mock_list_role_binding
,
685 mock_create_role_binding
,
687 mock_list_role_binding
.return_value
= FakeK8sRoleBindingList(items
=[1])
688 with self
.assertRaises(Exception) as context
:
689 await self
.kubectl
.create_role_binding(
697 "Role Binding with metadata.name={} already exists".format(self
.name
)
698 in str(context
.exception
)
700 mock_create_role_binding
.assert_not_called()
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()
709 self
.namespace
= "osm"
710 self
.data
= {"test": "1234"}
711 self
.secret_type
= "Opaque"
712 self
.kubectl
= Kubectl()
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
)
718 metadata
=secret_metadata
,
720 type=self
.secret_type
,
722 await self
.kubectl
.create_secret(
723 namespace
=self
.namespace
,
726 secret_type
=self
.secret_type
,
728 mock_create_secret
.assert_called_once_with(self
.namespace
, secret
)
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()
740 @asynctest.fail_on(active_handles
=True)
741 async def test_namespace_is_created(
743 mock_create_namespace
,
745 metadata
= V1ObjectMeta(name
=self
.namespace
, labels
=self
.labels
)
746 namespace
= V1Namespace(
749 await self
.kubectl
.create_namespace(
753 mock_create_namespace
.assert_called_once_with(namespace
)
755 async def test_namespace_is_created_default_labels(
757 mock_create_namespace
,
759 metadata
= V1ObjectMeta(name
=self
.namespace
, labels
=None)
760 namespace
= V1Namespace(
763 await self
.kubectl
.create_namespace(
766 mock_create_namespace
.assert_called_once_with(namespace
)
768 @asynctest.fail_on(active_handles
=True)
769 async def test_no_exception_if_alreadyexists(
771 mock_create_namespace
,
773 api_exception
= ApiException()
774 api_exception
.body
= '{"reason": "AlreadyExists"}'
775 self
.kubectl
.clients
[CORE_CLIENT
].create_namespace
.side_effect
= api_exception
778 await self
.kubectl
.create_namespace(
783 self
.assertFalse(raised
, "An exception was raised")
785 @asynctest.fail_on(active_handles
=True)
786 async def test_other_exceptions(
788 mock_create_namespace
,
790 self
.kubectl
.clients
[CORE_CLIENT
].create_namespace
.side_effect
= Exception()
791 with self
.assertRaises(Exception):
792 await self
.kubectl
.create_namespace(
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()
805 @asynctest.fail_on(active_handles
=True)
806 async def test_no_exception_if_notfound(
808 mock_delete_namespace
,
810 api_exception
= ApiException()
811 api_exception
.body
= '{"reason": "NotFound"}'
812 self
.kubectl
.clients
[CORE_CLIENT
].delete_namespace
.side_effect
= api_exception
815 await self
.kubectl
.delete_namespace(
820 self
.assertFalse(raised
, "An exception was raised")
822 @asynctest.fail_on(active_handles
=True)
823 async def test_other_exceptions(
825 mock_delete_namespace
,
827 self
.kubectl
.clients
[CORE_CLIENT
].delete_namespace
.side_effect
= Exception()
828 with self
.assertRaises(Exception):
829 await self
.kubectl
.delete_namespace(
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"}
843 self
.kubectl
= Kubectl()
845 @asynctest.fail_on(active_handles
=True)
846 async def test_return_type_is_dict(
848 mock_read_namespaced_secret
,
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