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=e5f150b7dcb5919aea5e25671a55af1dd9749c9a;hp=50e827ed09ff861c15962cbcbaebefde31535c8a;hb=2962f3e7aba84b4584d2deac30d1c163f6441a03;hpb=667696ef11356f3267df58f2a81c6ecebb0e94b9 diff --git a/n2vc/tests/unit/test_k8s_juju_conn.py b/n2vc/tests/unit/test_k8s_juju_conn.py index 50e827e..e5f150b 100644 --- a/n2vc/tests/unit/test_k8s_juju_conn.py +++ b/n2vc/tests/unit/test_k8s_juju_conn.py @@ -16,189 +16,81 @@ 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, FakeApplication from n2vc.exceptions import ( MethodNotImplemented, K8sException, - N2VCBadArgumentsException, ) -from unittest.mock import Mock -from .utils import AsyncMock +from n2vc.vca.connection_data import ConnectionData class K8sJujuConnTestCase(asynctest.TestCase): - @asynctest.mock.patch("juju.controller.Controller.update_endpoints") - @asynctest.mock.patch("juju.client.connector.Connector.connect") - @asynctest.mock.patch("juju.controller.Controller.connection") - @asynctest.mock.patch("n2vc.k8s_juju_conn.base64_to_cacert") @asynctest.mock.patch("n2vc.k8s_juju_conn.Libjuju") + @asynctest.mock.patch("n2vc.k8s_juju_conn.MotorStore") + @asynctest.mock.patch("n2vc.k8s_juju_conn.get_connection") + @asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert") def setUp( self, - mock_libjuju=None, mock_base64_to_cacert=None, - mock_connection=None, - mock_connect=None, - mock_update_endpoints=None, + mock_get_connection=None, + mock_store=None, + mock_libjuju=None, ): self.loop = asyncio.get_event_loop() - mock_libjuju.return_value = AsyncMock() - db = Mock() - vca_config = { - "secret": "secret", - "api_proxy": "api_proxy", - "cloud": "cloud", - "k8s_cloud": "k8s_cloud", - "user": "user", - "host": "1.1.1.1", - "port": 17070, - "ca_cert": "cacert", - } - + self.db = Mock() + mock_base64_to_cacert.return_value = """ + -----BEGIN CERTIFICATE----- + SOMECERT + -----END CERTIFICATE-----""" + mock_libjuju.return_value = Mock() + mock_store.return_value = AsyncMock() + mock_vca_connection = Mock() + mock_get_connection.return_value = mock_vca_connection + mock_vca_connection.data.return_value = ConnectionData( + **{ + "endpoints": ["1.2.3.4:17070"], + "user": "user", + "secret": "secret", + "cacert": "cacert", + "pubkey": "pubkey", + "lxd-cloud": "cloud", + "lxd-credentials": "credentials", + "k8s-cloud": "k8s_cloud", + "k8s-credentials": "k8s_credentials", + "model-config": {}, + "api-proxy": "api_proxy", + } + ) logging.disable(logging.CRITICAL) self.k8s_juju_conn = K8sJujuConnector( fs=fslocal.FsLocal(), - db=db, + db=self.db, log=None, loop=self.loop, - vca_config=vca_config, - on_update_db=None, - ) - - -class K8sJujuConnInitSuccessTestCase(asynctest.TestCase): - def setUp( - self, - ): - logging.disable(logging.CRITICAL) - - @asynctest.mock.patch("juju.controller.Controller.update_endpoints") - @asynctest.mock.patch("juju.client.connector.Connector.connect") - @asynctest.mock.patch("juju.controller.Controller.connection") - @asynctest.mock.patch("n2vc.k8s_juju_conn.base64_to_cacert") - @asynctest.mock.patch("n2vc.libjuju.Libjuju.__init__") - def test_success( - self, - mock_libjuju=None, - mock_base64_to_cacert=None, - mock_connection=None, - mock_connect=None, - mock_update_endpoints=None, - ): - mock_libjuju.return_value = None - loop = asyncio.get_event_loop() - log = logging.getLogger() - db = Mock() - vca_config = { - "secret": "secret", - "cloud": "cloud", - "k8s_cloud": "k8s_cloud", - "user": "user", - "host": "1.1.1.1", - "port": 17070, - "ca_cert": "cacert", - } - K8sJujuConnector( - fs=fslocal.FsLocal(), - db=db, - log=log, - loop=self.loop, - vca_config=vca_config, on_update_db=None, ) - - 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, - username="user", - password="secret", - cacert=mock_base64_to_cacert.return_value, - loop=loop, - log=log, - db=db, - ) - - -class K8sJujuConnectorInitFailureTestCase(asynctest.TestCase): - def setUp( - self, - ): - self.loop = asyncio.get_event_loop() - logging.disable(logging.CRITICAL) - self.vca_config = { - "secret": "secret", - "api_proxy": "api_proxy", - "cloud": "cloud", - "k8s_cloud": "k8s_cloud", - "user": "user", - "host": "1.1.1.1", - "port": 17070, - "ca_cert": "cacert", - } - - def test_missing_vca_config_host(self): - db = Mock() - self.vca_config.pop("host") - with self.assertRaises(N2VCBadArgumentsException): - self.k8s_juju_conn = K8sJujuConnector( - fs=fslocal.FsLocal(), - db=db, - log=None, - loop=self.loop, - vca_config=self.vca_config, - on_update_db=None, - ) - - def test_missing_vca_config_user(self): - db = Mock() - self.vca_config.pop("user") - with self.assertRaises(N2VCBadArgumentsException): - self.k8s_juju_conn = K8sJujuConnector( - fs=fslocal.FsLocal(), - db=db, - log=None, - loop=self.loop, - vca_config=self.vca_config, - on_update_db=None, - ) - - def test_missing_vca_config_secret(self): - db = Mock() - self.vca_config.pop("secret") - with self.assertRaises(N2VCBadArgumentsException): - self.k8s_juju_conn = K8sJujuConnector( - fs=fslocal.FsLocal(), - db=db, - log=None, - loop=self.loop, - vca_config=self.vca_config, - on_update_db=None, - ) - - def test_missing_vca_config_ca_cert(self): - db = Mock() - self.vca_config.pop("ca_cert") - with self.assertRaises(N2VCBadArgumentsException): - self.k8s_juju_conn = K8sJujuConnector( - fs=fslocal.FsLocal(), - db=db, - log=None, - loop=self.loop, - vca_config=self.vca_config, - on_update_db=None, - ) + self.k8s_juju_conn._store.get_vca_id.return_value = None + self.k8s_juju_conn.libjuju = Mock() +@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 +107,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 +120,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 +179,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 +213,86 @@ 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, - cloud_name=self.cluster_uuid, - credential_name="cred-{}".format(self.cluster_uuid), - ) + self.k8s_juju_conn.libjuju.add_model.assert_called_once() 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, - cloud_name=self.cluster_uuid, - credential_name="cred-{}".format(self.cluster_uuid), - ) + self.k8s_juju_conn.libjuju.add_model.assert_called_once() 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, - cloud_name=self.cluster_uuid, - credential_name="cred-{}".format(self.cluster_uuid), - ) + self.k8s_juju_conn.libjuju.add_model.assert_called_once() 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, - cloud_name=self.cluster_uuid, - credential_name="cred-{}".format(self.cluster_uuid), - ) + self.k8s_juju_conn.libjuju.add_model.assert_called_once() 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 +300,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 +317,59 @@ 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, - cloud_name=self.cluster_uuid, - credential_name="cred-{}".format(self.cluster_uuid), - ) + self.k8s_juju_conn.libjuju.add_model.assert_called_once() 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, - cloud_name=self.cluster_uuid, - credential_name="cred-{}".format(self.cluster_uuid), - ) + self.k8s_juju_conn.libjuju.add_model.assert_called_once() 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 +651,90 @@ 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() + + +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()