X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_kubectl.py;fp=n2vc%2Ftests%2Funit%2Ftest_kubectl.py;h=e67168e7157ec8b0f8275be41e8eec4a9acb29dd;hp=eb9b01d45bd3ff88a56562b83fb7c454ea360cd0;hb=6343d434fa3cec28d8b9b470054d3a13ada8865a;hpb=015abee87f591b8e28f6b982ae4fb9c67e8791bb diff --git a/n2vc/tests/unit/test_kubectl.py b/n2vc/tests/unit/test_kubectl.py index eb9b01d..e67168e 100644 --- a/n2vc/tests/unit/test_kubectl.py +++ b/n2vc/tests/unit/test_kubectl.py @@ -16,6 +16,12 @@ from unittest import TestCase, mock from n2vc.kubectl import Kubectl, CORE_CLIENT from n2vc.utils import Dict from kubernetes.client.rest import ApiException +from kubernetes.client import ( + V1ObjectMeta, + V1Secret, + V1ServiceAccount, + V1SecretReference, +) class FakeK8sResourceMetadata: @@ -66,6 +72,38 @@ class FakeK8sStorageClassesList: return self._items +class FakeK8sServiceAccountsList: + def __init__(self, items=[]): + self._items = items + + @property + def items(self): + return self._items + + +class FakeK8sSecretList: + def __init__(self, items=[]): + self._items = items + + @property + def items(self): + return self._items + + +class FakeK8sVersionApiCode: + def __init__(self, major: str, minor: str): + self._major = major + self._minor = minor + + @property + def major(self): + return self._major + + @property + def minor(self): + return self._minor + + fake_list_services = Dict( { "items": [ @@ -248,3 +286,140 @@ class GetDefaultStorageClass(KubectlTestCase): sc_name = kubectl.get_default_storage_class() self.assertEqual(sc_name, self.default_sc_name) mock_list_storage_class.assert_called_once() + + +@mock.patch("kubernetes.client.VersionApi.get_code") +@mock.patch("kubernetes.client.CoreV1Api.list_namespaced_secret") +@mock.patch("kubernetes.client.CoreV1Api.create_namespaced_secret") +@mock.patch("kubernetes.client.CoreV1Api.create_namespaced_service_account") +@mock.patch("kubernetes.client.CoreV1Api.list_namespaced_service_account") +class CreateServiceAccountClass(KubectlTestCase): + @mock.patch("kubernetes.config.load_kube_config") + def setUp(self, mock_load_kube_config): + super(CreateServiceAccountClass, self).setUp() + self.service_account_name = "Service_account" + self.labels = {"Key1": "Value1", "Key2": "Value2"} + self.namespace = "kubernetes" + self.token_id = "abc12345" + self.kubectl = Kubectl() + + def assert_create_secret(self, mock_create_secret, secret_name): + annotations = {"kubernetes.io/service-account.name": self.service_account_name} + secret_metadata = V1ObjectMeta( + name=secret_name, namespace=self.namespace, annotations=annotations + ) + secret_type = "kubernetes.io/service-account-token" + secret = V1Secret(metadata=secret_metadata, type=secret_type) + mock_create_secret.assert_called_once_with(self.namespace, secret) + + def assert_create_service_account_v_1_24( + self, mock_create_service_account, secret_name + ): + sevice_account_metadata = V1ObjectMeta( + name=self.service_account_name, labels=self.labels, namespace=self.namespace + ) + secrets = [V1SecretReference(name=secret_name, namespace=self.namespace)] + service_account = V1ServiceAccount( + metadata=sevice_account_metadata, secrets=secrets + ) + mock_create_service_account.assert_called_once_with( + self.namespace, service_account + ) + + def assert_create_service_account_v_1_23(self, mock_create_service_account): + metadata = V1ObjectMeta( + name=self.service_account_name, labels=self.labels, namespace=self.namespace + ) + service_account = V1ServiceAccount(metadata=metadata) + mock_create_service_account.assert_called_once_with( + self.namespace, service_account + ) + + @mock.patch("n2vc.kubectl.uuid.uuid4") + def test_secret_is_created_when_k8s_1_24( + self, + mock_uuid4, + mock_list_service_account, + mock_create_service_account, + mock_create_secret, + mock_list_secret, + mock_version, + ): + mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[]) + mock_list_secret.return_value = FakeK8sSecretList(items=[]) + mock_version.return_value = FakeK8sVersionApiCode("1", "24") + mock_uuid4.return_value = self.token_id + self.kubectl.create_service_account( + self.service_account_name, self.labels, self.namespace + ) + secret_name = "{}-token-{}".format(self.service_account_name, self.token_id[:5]) + self.assert_create_service_account_v_1_24( + mock_create_service_account, secret_name + ) + self.assert_create_secret(mock_create_secret, secret_name) + + def test_secret_is_not_created_when_k8s_1_23( + self, + mock_list_service_account, + mock_create_service_account, + mock_create_secret, + mock_list_secret, + mock_version, + ): + mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[]) + mock_version.return_value = FakeK8sVersionApiCode("1", "23+") + self.kubectl.create_service_account( + self.service_account_name, self.labels, self.namespace + ) + self.assert_create_service_account_v_1_23(mock_create_service_account) + mock_create_secret.assert_not_called() + mock_list_secret.assert_not_called() + + def test_raise_exception_if_service_account_already_exists( + self, + mock_list_service_account, + mock_create_service_account, + mock_create_secret, + mock_list_secret, + mock_version, + ): + mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[1]) + with self.assertRaises(Exception) as context: + self.kubectl.create_service_account( + self.service_account_name, self.labels, self.namespace + ) + self.assertTrue( + "Service account with metadata.name={} already exists".format( + self.service_account_name + ) + in str(context.exception) + ) + mock_create_service_account.assert_not_called() + mock_create_secret.assert_not_called() + + @mock.patch("n2vc.kubectl.uuid.uuid4") + def test_raise_exception_if_secret_already_exists( + self, + mock_uuid4, + mock_list_service_account, + mock_create_service_account, + mock_create_secret, + mock_list_secret, + mock_version, + ): + mock_list_service_account.return_value = FakeK8sServiceAccountsList(items=[]) + mock_list_secret.return_value = FakeK8sSecretList(items=[1]) + mock_version.return_value = FakeK8sVersionApiCode("1", "24+") + mock_uuid4.return_value = self.token_id + with self.assertRaises(Exception) as context: + self.kubectl.create_service_account( + self.service_account_name, self.labels, self.namespace + ) + self.assertTrue( + "Secret with metadata.name={}-token-{} already exists".format( + self.service_account_name, self.token_id[:5] + ) + in str(context.exception) + ) + mock_create_service_account.assert_called() + mock_create_secret.assert_not_called()