X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_kubectl.py;h=e67168e7157ec8b0f8275be41e8eec4a9acb29dd;hb=2f0692e61e18466f6e7df3e580140f27cad396de;hp=dfb954a00648280b8f80a0908613afb4ec90fabf;hpb=37004983e8e484d5504ae4253bdb75204ff389d9;p=osm%2FN2VC.git diff --git a/n2vc/tests/unit/test_kubectl.py b/n2vc/tests/unit/test_kubectl.py index dfb954a..e67168e 100644 --- a/n2vc/tests/unit/test_kubectl.py +++ b/n2vc/tests/unit/test_kubectl.py @@ -13,9 +13,96 @@ # 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: + def __init__( + self, + name: str = None, + namespace: str = None, + annotations: dict = {}, + labels: dict = {}, + ): + self._annotations = annotations + self._name = name or "name" + self._namespace = namespace or "namespace" + self._labels = labels or {"juju-app": "squid"} + + @property + def name(self): + return self._name + + @property + def namespace(self): + return self._namespace + + @property + def labels(self): + return self._labels + + @property + def annotations(self): + return self._annotations + + +class FakeK8sStorageClass: + def __init__(self, metadata=None): + self._metadata = metadata or FakeK8sResourceMetadata() + + @property + def metadata(self): + return self._metadata + + +class FakeK8sStorageClassesList: + def __init__(self, items=[]): + self._items = items + + @property + def items(self): + 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( { @@ -64,12 +151,19 @@ fake_list_services = Dict( ) +class KubectlTestCase(TestCase): + def setUp( + self, + ): + pass + + class FakeCoreV1Api: def list_service_for_all_namespaces(self, **kwargs): 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): @@ -86,8 +180,246 @@ 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("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, + mock_client, + ): + kubectl = Kubectl() + 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") +@mock.patch("kubernetes.config.load_kube_config") +class GetDefaultStorageClass(KubectlTestCase): + def setUp(self): + super(GetDefaultStorageClass, self).setUp() + + # Default Storage Class + self.default_sc_name = "default-sc" + default_sc_metadata = FakeK8sResourceMetadata( + name=self.default_sc_name, + annotations={"storageclass.kubernetes.io/is-default-class": "true"}, + ) + self.default_sc = FakeK8sStorageClass(metadata=default_sc_metadata) + + # Default Storage Class with old annotation + self.default_sc_old_name = "default-sc-old" + default_sc_old_metadata = FakeK8sResourceMetadata( + name=self.default_sc_old_name, + annotations={"storageclass.beta.kubernetes.io/is-default-class": "true"}, + ) + self.default_sc_old = FakeK8sStorageClass(metadata=default_sc_old_metadata) + + # Storage class - not default + self.sc_name = "default-sc-old" + self.sc = FakeK8sStorageClass( + metadata=FakeK8sResourceMetadata(name=self.sc_name) + ) + + def test_get_default_storage_class_exists_default( + self, mock_load_kube_config, mock_list_storage_class + ): + kubectl = Kubectl() + items = [self.default_sc] + mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items) + sc_name = kubectl.get_default_storage_class() + self.assertEqual(sc_name, self.default_sc_name) + mock_list_storage_class.assert_called_once() + + def test_get_default_storage_class_exists_default_old( + self, mock_load_kube_config, mock_list_storage_class + ): + kubectl = Kubectl() + items = [self.default_sc_old] + mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items) + sc_name = kubectl.get_default_storage_class() + self.assertEqual(sc_name, self.default_sc_old_name) + mock_list_storage_class.assert_called_once() + + def test_get_default_storage_class_none( + self, mock_load_kube_config, mock_list_storage_class + ): + kubectl = Kubectl() + mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=[]) + sc_name = kubectl.get_default_storage_class() + self.assertEqual(sc_name, None) + mock_list_storage_class.assert_called_once() + + def test_get_default_storage_class_exists_not_default( + self, mock_load_kube_config, mock_list_storage_class + ): + kubectl = Kubectl() + items = [self.sc] + mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items) + sc_name = kubectl.get_default_storage_class() + self.assertEqual(sc_name, self.sc_name) + mock_list_storage_class.assert_called_once() + + def test_get_default_storage_class_choose( + self, mock_load_kube_config, mock_list_storage_class + ): + kubectl = Kubectl() + items = [self.sc, self.default_sc] + mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items) + 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()