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.

from unittest import TestCase, mock
from n2vc.kubectl import Kubectl
from n2vc.utils import Dict
from kubernetes.client.rest import ApiException


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


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

    @mock.patch("n2vc.kubectl.client.CoreV1Api.list_service_for_all_namespaces")
    def test_get_service_exception(self, list_services):
        list_services.side_effect = ApiException()
        with self.assertRaises(ApiException):
            self.kubectl.get_services()


@mock.patch("kubernetes.config.kube_config.Configuration")
@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):
        kubectl = Kubectl()
        kubectl.get_configuration()
        mock_configuration.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()