X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_kubectl.py;h=e67168e7157ec8b0f8275be41e8eec4a9acb29dd;hb=085fa8d4658a9b621354d5a08853086e2696abdc;hp=8fb0310ca41173d1848cb35097490971e6f70ffe;hpb=475a7221e3598ad1c75ce802c5ad74ef7ecf72f1;p=osm%2FN2VC.git diff --git a/n2vc/tests/unit/test_kubectl.py b/n2vc/tests/unit/test_kubectl.py index 8fb0310..e67168e 100644 --- a/n2vc/tests/unit/test_kubectl.py +++ b/n2vc/tests/unit/test_kubectl.py @@ -13,9 +13,15 @@ # limitations under the License. from unittest import TestCase, mock -from n2vc.kubectl import Kubectl +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": [ @@ -114,7 +152,9 @@ fake_list_services = Dict( class KubectlTestCase(TestCase): - def setUp(self,): + def setUp( + self, + ): pass @@ -123,7 +163,7 @@ class FakeCoreV1Api: return fake_list_services -class ProvisionerTest(TestCase): +class GetServices(TestCase): @mock.patch("n2vc.kubectl.config.load_kube_config") @mock.patch("n2vc.kubectl.client.CoreV1Api") def setUp(self, mock_core, mock_config): @@ -140,23 +180,34 @@ class ProvisionerTest(TestCase): keys = ["name", "cluster_ip", "type", "ports", "external_ip"] self.assertTrue(k in service for service in services for k in keys) - @mock.patch("n2vc.kubectl.client.CoreV1Api.list_service_for_all_namespaces") - def test_get_service_exception(self, list_services): - list_services.side_effect = ApiException() + def test_get_service_exception(self): + self.kubectl.clients[ + CORE_CLIENT + ].list_service_for_all_namespaces.side_effect = ApiException() with self.assertRaises(ApiException): self.kubectl.get_services() -@mock.patch("kubernetes.config.kube_config.Configuration") +@mock.patch("n2vc.kubectl.client") +@mock.patch("n2vc.kubectl.config.kube_config.Configuration.get_default_copy") @mock.patch("n2vc.kubectl.config.load_kube_config") class GetConfiguration(KubectlTestCase): def setUp(self): super(GetConfiguration, self).setUp() - def test_get_configuration(self, mock_load_kube_config, mock_configuration): + def test_get_configuration( + self, + mock_load_kube_config, + mock_configuration, + mock_client, + ): kubectl = Kubectl() - kubectl.get_configuration() + kubectl.configuration mock_configuration.assert_called_once() + mock_load_kube_config.assert_called_once() + mock_client.CoreV1Api.assert_called_once() + mock_client.RbacAuthorizationV1Api.assert_called_once() + mock_client.StorageV1Api.assert_called_once() @mock.patch("kubernetes.client.StorageV1Api.list_storage_class") @@ -235,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()