Add new kubectl.py functions, modify some libjuju.py functions, add unit tests
- Kubectl.py: two new functions added (get_configuration and get_default_storage_class)
- get_configuration(): Returns a kubernetes Configuration object.
It can be used to properly parse the kubeconfig.
- get_default_storage_class(): Searches for the default storage class of a k8s cluster.
- Libjuju.py: modified add_k8s function and get_k8s_cloud_credential function was added.
- add_k8s(): Improves the way of generation Cloud and CloudCredential objects for the K8s Cloud
- get_k8s_cloud_credential(): It parses the kubeconfig to properly determine the authentication
method type that should be used for that k8s cluster.
- Unit tests: Added unit tests for all the new functions added.
- Exceptions: Make all Juju Exceptions to inherit from N2VC Exception.
Now Juju exceptions have the message attribute, that is useful for unit testing, to not only check that
an exception raised, but to check the message too.
- Move get_k8s_cloud_credential() function to n2vc/utils in order to share that code between different connectors.
Change-Id: Ife9027d80663fe95f1f3ad883cb9a3376b047d0b
Signed-off-by: David Garcia <david.garcia@canonical.com>
diff --git a/n2vc/tests/unit/test_kubectl.py b/n2vc/tests/unit/test_kubectl.py
index dfb954a..8fb0310 100644
--- a/n2vc/tests/unit/test_kubectl.py
+++ b/n2vc/tests/unit/test_kubectl.py
@@ -17,6 +17,55 @@
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": [
@@ -64,6 +113,11 @@
)
+class KubectlTestCase(TestCase):
+ def setUp(self,):
+ pass
+
+
class FakeCoreV1Api:
def list_service_for_all_namespaces(self, **kwargs):
return fake_list_services
@@ -91,3 +145,93 @@
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()