X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_k8s_juju_conn.py;h=f4543801bbb33c3e4d3ecf994e658ddd7ee2a5a2;hp=50e827ed09ff861c15962cbcbaebefde31535c8a;hb=f8c25c9ceeeca8ac91caa62faa402a8b6dab02e1;hpb=667696ef11356f3267df58f2a81c6ecebb0e94b9;ds=inline diff --git a/n2vc/tests/unit/test_k8s_juju_conn.py b/n2vc/tests/unit/test_k8s_juju_conn.py index 50e827e..f454380 100644 --- a/n2vc/tests/unit/test_k8s_juju_conn.py +++ b/n2vc/tests/unit/test_k8s_juju_conn.py @@ -16,16 +16,15 @@ import asyncio import logging import asynctest -from n2vc.k8s_juju_conn import K8sJujuConnector +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 +from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock from n2vc.exceptions import ( MethodNotImplemented, K8sException, N2VCBadArgumentsException, ) -from unittest.mock import Mock -from .utils import AsyncMock class K8sJujuConnTestCase(asynctest.TestCase): @@ -112,8 +111,7 @@ class K8sJujuConnInitSuccessTestCase(asynctest.TestCase): mock_libjuju.assert_called_once_with( endpoint="1.1.1.1:17070", api_proxy=None, # Not needed for k8s charms - enable_os_upgrade=True, - apt_mirror=None, + model_config={}, username="user", password="secret", cacert=mock_base64_to_cacert.return_value, @@ -193,12 +191,20 @@ class K8sJujuConnectorInitFailureTestCase(asynctest.TestCase): ) +@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") - @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_default_storage_class") def test_with_cluster_uuid( self, mock_get_default_storage_class, @@ -215,8 +221,10 @@ class InitEnvTest(K8sJujuConnTestCase): mock_get_default_storage_class.assert_called_once() self.k8s_juju_conn.libjuju.add_k8s.assert_called_once() - @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_default_storage_class") - def test_with_no_cluster_uuid(self, mock_get_default_storage_class): + 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) ) @@ -226,8 +234,10 @@ class InitEnvTest(K8sJujuConnTestCase): mock_get_default_storage_class.assert_called_once() self.k8s_juju_conn.libjuju.add_k8s.assert_called_once() - @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_default_storage_class") - def test_init_env_exception(self, mock_get_default_storage_class): + def test_init_env_exception( + self, + mock_get_default_storage_class, + ): self.k8s_juju_conn.libjuju.add_k8s.side_effect = Exception() created = None uuid = None @@ -283,6 +293,15 @@ class ResetTest(K8sJujuConnTestCase): def setUp(self): super(ResetTest, self).setUp() self.k8s_juju_conn.libjuju.remove_cloud = AsyncMock() + self.k8s_juju_conn.libjuju.get_cloud_credentials = AsyncMock() + 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 test_success(self): removed = self.loop.run_until_complete(self.k8s_juju_conn.reset("uuid")) @@ -308,105 +327,102 @@ class InstallTest(K8sJujuConnTestCase): self.http_bundle = "https://example.com/bundle.yaml" self.kdu_name = "kdu_name" self.cluster_uuid = "cluster" + self.kdu_instance = "{}-{}".format(self.kdu_name, "id") self.k8s_juju_conn.libjuju.add_model = AsyncMock() self.k8s_juju_conn.libjuju.deploy = AsyncMock() def test_success_local(self, mock_chdir): - expected_kdu_instance = "{}-{}".format(self.kdu_name, "id") - kdu_instance = self.loop.run_until_complete( + self.loop.run_until_complete( self.k8s_juju_conn.install( self.cluster_uuid, self.local_bundle, + self.kdu_instance, atomic=True, kdu_name=self.kdu_name, db_dict=self.db_dict, timeout=1800, ) ) - self.assertEqual(kdu_instance, expected_kdu_instance) self.assertEqual(mock_chdir.call_count, 2) self.k8s_juju_conn.libjuju.add_model.assert_called_once_with( - model_name=expected_kdu_instance, + model_name=self.kdu_instance, cloud_name=self.cluster_uuid, credential_name="cred-{}".format(self.cluster_uuid), ) self.k8s_juju_conn.libjuju.deploy.assert_called_once_with( "local:{}".format(self.local_bundle), - model_name=expected_kdu_instance, + model_name=self.kdu_instance, wait=True, timeout=1800, ) def test_success_cs(self, mock_chdir): - expected_kdu_instance = "{}-{}".format(self.kdu_name, "id") - kdu_instance = self.loop.run_until_complete( + self.loop.run_until_complete( self.k8s_juju_conn.install( self.cluster_uuid, self.cs_bundle, + self.kdu_instance, atomic=True, kdu_name=self.kdu_name, db_dict=self.db_dict, timeout=1800, ) ) - self.assertEqual(kdu_instance, expected_kdu_instance) self.k8s_juju_conn.libjuju.add_model.assert_called_once_with( - model_name=expected_kdu_instance, + model_name=self.kdu_instance, cloud_name=self.cluster_uuid, credential_name="cred-{}".format(self.cluster_uuid), ) self.k8s_juju_conn.libjuju.deploy.assert_called_once_with( self.cs_bundle, - model_name=expected_kdu_instance, + model_name=self.kdu_instance, wait=True, timeout=1800, ) def test_success_http(self, mock_chdir): - expected_kdu_instance = "{}-{}".format(self.kdu_name, "id") - kdu_instance = self.loop.run_until_complete( + self.loop.run_until_complete( self.k8s_juju_conn.install( self.cluster_uuid, self.http_bundle, + self.kdu_instance, atomic=True, kdu_name=self.kdu_name, db_dict=self.db_dict, timeout=1800, ) ) - self.assertEqual(kdu_instance, expected_kdu_instance) self.k8s_juju_conn.libjuju.add_model.assert_called_once_with( - model_name=expected_kdu_instance, + model_name=self.kdu_instance, cloud_name=self.cluster_uuid, credential_name="cred-{}".format(self.cluster_uuid), ) self.k8s_juju_conn.libjuju.deploy.assert_called_once_with( self.http_bundle, - model_name=expected_kdu_instance, + model_name=self.kdu_instance, wait=True, timeout=1800, ) def test_success_not_kdu_name(self, mock_chdir): - expected_kdu_instance = "id" - kdu_instance = self.loop.run_until_complete( + self.loop.run_until_complete( self.k8s_juju_conn.install( self.cluster_uuid, self.cs_bundle, + self.kdu_instance, atomic=True, db_dict=self.db_dict, timeout=1800, ) ) - self.assertEqual(kdu_instance, expected_kdu_instance) self.k8s_juju_conn.libjuju.add_model.assert_called_once_with( - model_name=expected_kdu_instance, + model_name=self.kdu_instance, cloud_name=self.cluster_uuid, credential_name="cred-{}".format(self.cluster_uuid), ) self.k8s_juju_conn.libjuju.deploy.assert_called_once_with( self.cs_bundle, - model_name=expected_kdu_instance, + model_name=self.kdu_instance, wait=True, timeout=1800, ) @@ -414,10 +430,11 @@ class InstallTest(K8sJujuConnTestCase): def test_missing_db_dict(self, mock_chdir): kdu_instance = None with self.assertRaises(K8sException): - kdu_instance = self.loop.run_until_complete( + self.loop.run_until_complete( self.k8s_juju_conn.install( self.cluster_uuid, self.cs_bundle, + self.kdu_instance, atomic=True, kdu_name=self.kdu_name, timeout=1800, @@ -430,71 +447,67 @@ class InstallTest(K8sJujuConnTestCase): @asynctest.mock.patch("os.getcwd") def test_getcwd_exception(self, mock_getcwd, mock_chdir): mock_getcwd.side_effect = FileNotFoundError() - expected_kdu_instance = "{}-{}".format(self.kdu_name, "id") - kdu_instance = self.loop.run_until_complete( + self.loop.run_until_complete( self.k8s_juju_conn.install( self.cluster_uuid, self.cs_bundle, + self.kdu_instance, atomic=True, kdu_name=self.kdu_name, db_dict=self.db_dict, timeout=1800, ) ) - self.assertEqual(kdu_instance, expected_kdu_instance) self.k8s_juju_conn.libjuju.add_model.assert_called_once_with( - model_name=expected_kdu_instance, + model_name=self.kdu_instance, cloud_name=self.cluster_uuid, credential_name="cred-{}".format(self.cluster_uuid), ) self.k8s_juju_conn.libjuju.deploy.assert_called_once_with( self.cs_bundle, - model_name=expected_kdu_instance, + model_name=self.kdu_instance, wait=True, timeout=1800, ) def test_missing_bundle(self, mock_chdir): - kdu_instance = None with self.assertRaises(K8sException): - kdu_instance = self.loop.run_until_complete( + self.loop.run_until_complete( self.k8s_juju_conn.install( self.cluster_uuid, "", + self.kdu_instance, atomic=True, kdu_name=self.kdu_name, timeout=1800, db_dict=self.db_dict, ) ) - self.assertIsNone(kdu_instance) self.k8s_juju_conn.libjuju.add_model.assert_not_called() self.k8s_juju_conn.libjuju.deploy.assert_not_called() def test_missing_exception(self, mock_chdir): - expected_kdu_instance = "{}-{}".format(self.kdu_name, "id") - kdu_instance = None self.k8s_juju_conn.libjuju.deploy.side_effect = Exception() with self.assertRaises(Exception): - kdu_instance = self.loop.run_until_complete( + self.loop.run_until_complete( self.k8s_juju_conn.install( self.cluster_uuid, self.local_bundle, + self.kdu_instance, atomic=True, kdu_name=self.kdu_name, db_dict=self.db_dict, timeout=1800, ) ) - self.assertIsNone(kdu_instance) self.k8s_juju_conn.libjuju.add_model.assert_called_once_with( - model_name=expected_kdu_instance, + model_name=self.kdu_instance, cloud_name=self.cluster_uuid, credential_name="cred-{}".format(self.cluster_uuid), ) self.k8s_juju_conn.libjuju.deploy.assert_called_once_with( "local:{}".format(self.local_bundle), - model_name=expected_kdu_instance, + model_name=self.kdu_instance, wait=True, timeout=1800, ) @@ -776,3 +789,31 @@ class GetCredentialsTest(K8sJujuConnTestCase): self.k8s_juju_conn.db.get_one.assert_called_once() self.k8s_juju_conn.db.encrypt_decrypt_fields.assert_called_once() mock_safe_dump.assert_called_once() + + +class UpdateVcaStatusTest(K8sJujuConnTestCase): + def setUp(self): + super(UpdateVcaStatusTest, self).setUp() + self.vcaStatus = {"model": {"applications": {"app": {"actions": {}}}}} + self.kdu_name = "kdu_name" + self.kdu_instance = "{}-{}".format(self.kdu_name, "id") + self.k8s_juju_conn.libjuju.get_executed_actions = AsyncMock() + self.k8s_juju_conn.libjuju.get_actions = AsyncMock() + 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.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() + + def test_exception(self): + 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.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()