n2vc/tests/unit/test_kubectl.py

# Copyright 2020 Canonical Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
#     Unless required by applicable law or agreed to in writing, software
#     distributed under the License is distributed on an "AS IS" BASIS,
#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#     See the License for the specific language governing permissions and
#     limitations under the License.

import asynctest
import yaml
import os
from unittest import TestCase, mock
from n2vc.kubectl import Kubectl, CORE_CLIENT, CUSTOM_OBJECT_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(
    {
        "items": [
            Dict(
                {
                    "metadata": Dict(
                        {
                            "name": "squid",
                            "namespace": "test",
                            "labels": {"juju-app": "squid"},
                        }
                    ),
                    "spec": Dict(
                        {
                            "cluster_ip": "10.152.183.79",
                            "type": "LoadBalancer",
                            "ports": [
                                Dict(
                                    {
                                        "name": None,
                                        "node_port": None,
                                        "port": 30666,
                                        "protocol": "TCP",
                                        "target_port": 30666,
                                    }
                                )
                            ],
                        }
                    ),
                    "status": Dict(
                        {
                            "load_balancer": Dict(
                                {
                                    "ingress": [
                                        Dict({"hostname": None, "ip": "192.168.0.201"})
                                    ]
                                }
                            )
                        }
                    ),
                }
            )
        ]
    }
)


class KubectlTestCase(TestCase):
    def setUp(
        self,
    ):
        pass


class FakeCoreV1Api:
    def list_service_for_all_namespaces(self, **kwargs):
        return fake_list_services


class GetServices(TestCase):
    @mock.patch("n2vc.kubectl.config.load_kube_config")
    @mock.patch("n2vc.kubectl.client.CoreV1Api")
    def setUp(self, mock_core, mock_config):
        mock_core.return_value = mock.MagicMock()
        mock_config.return_value = mock.MagicMock()
        self.kubectl = Kubectl()

    @mock.patch("n2vc.kubectl.client.CoreV1Api")
    def test_get_service(self, mock_corev1api):
        mock_corev1api.return_value = FakeCoreV1Api()
        services = self.kubectl.get_services(
            field_selector="metadata.namespace", label_selector="juju-operator=squid"
        )
        keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
        self.assertTrue(k in service for service in services for k in keys)

    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()


@mock.patch("kubernetes.client.CustomObjectsApi.create_namespaced_custom_object")
class CreateCertificateClass(asynctest.TestCase):
    @mock.patch("kubernetes.config.load_kube_config")
    def setUp(self, mock_load_kube_config):
        super(CreateCertificateClass, self).setUp()
        self.namespace = "osm"
        self.name = "test-cert"
        self.dns_prefix = "*"
        self.secret_name = "test-cert-secret"
        self.usages = ["server auth"]
        self.issuer_name = "ca-issuer"
        self.kubectl = Kubectl()

    @asynctest.fail_on(active_handles=True)
    async def test_certificate_is_created(
        self,
        mock_create_certificate,
    ):
        with open(
            os.path.join(
                os.path.dirname(__file__), "testdata", "test_certificate.yaml"
            ),
            "r",
        ) as test_certificate:
            certificate_body = yaml.safe_load(test_certificate.read())
            print(certificate_body)
        await self.kubectl.create_certificate(
            namespace=self.namespace,
            name=self.name,
            dns_prefix=self.dns_prefix,
            secret_name=self.secret_name,
            usages=self.usages,
            issuer_name=self.issuer_name,
        )
        mock_create_certificate.assert_called_once_with(
            group="cert-manager.io",
            plural="certificates",
            version="v1",
            body=certificate_body,
            namespace=self.namespace,
        )

    @asynctest.fail_on(active_handles=True)
    async def test_no_exception_if_alreadyexists(
        self,
        mock_create_certificate,
    ):
        api_exception = ApiException()
        api_exception.body = '{"reason": "AlreadyExists"}'
        self.kubectl.clients[
            CUSTOM_OBJECT_CLIENT
        ].create_namespaced_custom_object.side_effect = api_exception
        raised = False
        try:
            await self.kubectl.create_certificate(
                namespace=self.namespace,
                name=self.name,
                dns_prefix=self.dns_prefix,
                secret_name=self.secret_name,
                usages=self.usages,
                issuer_name=self.issuer_name,
            )
        except Exception:
            raised = True
        self.assertFalse(raised, "An exception was raised")

    @asynctest.fail_on(active_handles=True)
    async def test_other_exceptions(
        self,
        mock_create_certificate,
    ):
        self.kubectl.clients[
            CUSTOM_OBJECT_CLIENT
        ].create_namespaced_custom_object.side_effect = Exception()
        with self.assertRaises(Exception):
            await self.kubectl.create_certificate(
                namespace=self.namespace,
                name=self.name,
                dns_prefix=self.dns_prefix,
                secret_name=self.secret_name,
                usages=self.usages,
                issuer_name=self.issuer_name,
            )


@mock.patch("kubernetes.client.CustomObjectsApi.delete_namespaced_custom_object")
class DeleteCertificateClass(asynctest.TestCase):
    @mock.patch("kubernetes.config.load_kube_config")
    def setUp(self, mock_load_kube_config):
        super(DeleteCertificateClass, self).setUp()
        self.namespace = "osm"
        self.object_name = "test-cert"
        self.kubectl = Kubectl()

    @asynctest.fail_on(active_handles=True)
    async def test_no_exception_if_notfound(
        self,
        mock_create_certificate,
    ):
        api_exception = ApiException()
        api_exception.body = '{"reason": "NotFound"}'
        self.kubectl.clients[
            CUSTOM_OBJECT_CLIENT
        ].delete_namespaced_custom_object.side_effect = api_exception
        raised = False
        try:
            await self.kubectl.delete_certificate(
                namespace=self.namespace,
                object_name=self.object_name,
            )
        except Exception:
            raised = True
        self.assertFalse(raised, "An exception was raised")

    @asynctest.fail_on(active_handles=True)
    async def test_other_exceptions(
        self,
        mock_create_certificate,
    ):
        self.kubectl.clients[
            CUSTOM_OBJECT_CLIENT
        ].delete_namespaced_custom_object.side_effect = Exception()
        with self.assertRaises(Exception):
            await self.kubectl.delete_certificate(
                namespace=self.namespace,
                object_name=self.object_name,
            )