from unittest.mock import Mock
from n2vc.k8s_juju_conn import K8sJujuConnector, RBAC_LABEL_KEY_NAME
from osm_common import fslocal
-from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock
+from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock, FakeApplication
from n2vc.exceptions import (
MethodNotImplemented,
K8sException,
)
self.k8s_juju_conn._store.get_vca_id.return_value = None
self.k8s_juju_conn.libjuju = Mock()
+ # Mock Kubectl
+ self.kubectl = Mock()
+ self.kubectl.get_secret_data = AsyncMock()
+ self.kubectl.get_secret_data.return_value = ("token", "cacert")
+ self.kubectl.get_services.return_value = [{}]
+ self.k8s_juju_conn._get_kubectl = Mock()
+ self.k8s_juju_conn._get_kubectl.return_value = self.kubectl
-@asynctest.mock.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
class InitEnvTest(K8sJujuConnTestCase):
def setUp(self):
super(InitEnvTest, self).setUp()
self.k8s_juju_conn.libjuju.add_k8s = AsyncMock()
- self.k8s_juju_conn._create_cluster_role = Mock()
- self.k8s_juju_conn._create_service_account = Mock()
- self.k8s_juju_conn._create_cluster_role_binding = Mock()
- self.k8s_juju_conn._delete_cluster_role = Mock()
- self.k8s_juju_conn._delete_service_account = Mock()
- self.k8s_juju_conn._delete_cluster_role_binding = Mock()
- self.k8s_juju_conn._get_secret_data = AsyncMock()
- self.k8s_juju_conn._get_secret_data.return_value = ("token", "cacert")
def test_with_cluster_uuid(
self,
- mock_get_default_storage_class,
):
reuse_cluster_uuid = "uuid"
uuid, created = self.loop.run_until_complete(
self.assertTrue(created)
self.assertEqual(uuid, reuse_cluster_uuid)
- mock_get_default_storage_class.assert_called_once()
+ self.kubectl.get_default_storage_class.assert_called_once()
self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
def test_with_no_cluster_uuid(
self,
- mock_get_default_storage_class,
):
uuid, created = self.loop.run_until_complete(
self.k8s_juju_conn.init_env(k8s_creds=kubeconfig)
self.assertTrue(created)
self.assertTrue(isinstance(uuid, str))
- mock_get_default_storage_class.assert_called_once()
+ self.kubectl.get_default_storage_class.assert_called_once()
self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
def test_init_env_exception(
self,
- mock_get_default_storage_class,
):
self.k8s_juju_conn.libjuju.add_k8s.side_effect = Exception()
created = None
self.assertIsNone(created)
self.assertIsNone(uuid)
- mock_get_default_storage_class.assert_called_once()
self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
cloud_creds = Mock()
cloud_creds.result = {"attrs": {RBAC_LABEL_KEY_NAME: "asd"}}
self.k8s_juju_conn.libjuju.get_cloud_credentials.return_value = [cloud_creds]
- self.k8s_juju_conn._delete_cluster_role_binding = Mock()
- self.k8s_juju_conn._delete_service_account = Mock()
- self.k8s_juju_conn._delete_cluster_role = Mock()
self.k8s_juju_conn.get_credentials = Mock()
self.k8s_juju_conn.get_credentials.return_value = kubeconfig
def setUp(self):
super(GetServicesTest, self).setUp()
- @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_services")
@asynctest.mock.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
- def test_success(self, mock_get_credentials, mock_get_services):
+ def test_success(self, mock_get_credentials):
mock_get_credentials.return_value = kubeconfig
self.loop.run_until_complete(self.k8s_juju_conn.get_services("", "", ""))
mock_get_credentials.assert_called_once()
- mock_get_services.assert_called_once()
+ self.kubectl.get_services.assert_called_once()
class GetServiceTest(K8sJujuConnTestCase):
def setUp(self):
super(GetServiceTest, self).setUp()
- @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_services")
@asynctest.mock.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
- def test_success(self, mock_get_credentials, mock_get_services):
+ def test_success(self, mock_get_credentials):
mock_get_credentials.return_value = kubeconfig
self.loop.run_until_complete(self.k8s_juju_conn.get_service("", "", ""))
mock_get_credentials.assert_called_once()
- mock_get_services.assert_called_once()
+ self.kubectl.get_services.assert_called_once()
class GetCredentialsTest(K8sJujuConnTestCase):
self.k8s_juju_conn.libjuju.get_application_configs = AsyncMock()
def test_success(self):
- self.loop.run_until_complete(self.k8s_juju_conn.update_vca_status(
- self.vcaStatus, self.kdu_instance))
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.update_vca_status(self.vcaStatus, self.kdu_instance)
+ )
self.k8s_juju_conn.libjuju.get_executed_actions.assert_called_once()
self.k8s_juju_conn.libjuju.get_actions.assert_called_once()
self.k8s_juju_conn.libjuju.get_application_configs.assert_called_once()
self.k8s_juju_conn.libjuju.get_model.return_value = None
self.k8s_juju_conn.libjuju.get_executed_actions.side_effect = Exception()
with self.assertRaises(Exception):
- self.loop.run_until_complete(self.k8s_juju_conn.update_vca_status(
- self.vcaStatus, self.kdu_instance))
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.update_vca_status(self.vcaStatus, self.kdu_instance)
+ )
self.k8s_juju_conn.libjuju.get_executed_actions.assert_not_called()
self.k8s_juju_conn.libjuju.get_actions.assert_not_called_once()
self.k8s_juju_conn.libjuju.get_application_configs.assert_not_called_once()
+
+
+class ScaleTest(K8sJujuConnTestCase):
+ def setUp(self):
+ super(ScaleTest, self).setUp()
+ self.application_name = "app"
+ self.kdu_name = "kdu-instance"
+ self._scale = 2
+ self.k8s_juju_conn.libjuju.scale_application = AsyncMock()
+
+ def test_success(self):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.scale(self.kdu_name, self._scale, self.application_name)
+ )
+ self.k8s_juju_conn.libjuju.scale_application.assert_called_once()
+
+ def test_exception(self):
+ self.k8s_juju_conn.libjuju.scale_application.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.scale(
+ self.kdu_name, self._scale, self.application_name
+ )
+ )
+ self.k8s_juju_conn.libjuju.scale_application.assert_called_once()
+
+
+class GetScaleCount(K8sJujuConnTestCase):
+ def setUp(self):
+ super(GetScaleCount, self).setUp()
+ self.k8s_juju_conn.libjuju.get_model_status = AsyncMock()
+
+ def test_success(self):
+ applications = {"app": FakeApplication()}
+ model = FakeModel(applications=applications)
+ self.k8s_juju_conn.libjuju.get_model_status.return_value = model
+ status = self.loop.run_until_complete(
+ self.k8s_juju_conn.get_scale_count("app", "kdu_instance")
+ )
+ self.assertEqual(status, 2)
+ self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
+
+ def test_exception(self):
+ self.k8s_juju_conn.libjuju.get_model_status.side_effect = Exception()
+ status = None
+ with self.assertRaises(Exception):
+ status = self.loop.run_until_complete(
+ self.k8s_juju_conn.status_kdu("app", "kdu_instance")
+ )
+ self.assertIsNone(status)
+ self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()