# 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
+
+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": [
)
+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):
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("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()