Fix flake8 minor issues

[osm/N2VC.git] / n2vc / tests / unit / test_kubectl.py

diff --git a/n2vc/tests/unit/test_kubectl.py b/n2vc/tests/unit/test_kubectl.py
index dfb954a..28d66c5 100644 (file)
--- a/n2vc/tests/unit/test_kubectl.py
+++ b/n2vc/tests/unit/test_kubectl.py
@@ -13,10 +13,59 @@
 #     limitations under the License.
 
 from unittest import TestCase, mock
 #     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 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": [
 fake_list_services = Dict(
     {
         "items": [


@@ -64,12 +113,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 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):
     @mock.patch("n2vc.kubectl.config.load_kube_config")
     @mock.patch("n2vc.kubectl.client.CoreV1Api")
     def setUp(self, mock_core, mock_config):


@@ -86,8 +142,99 @@ class ProvisionerTest(TestCase):
         keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
         self.assertTrue(k in service for service in services for k in keys)
 
         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()
         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.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()