X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_k8s_juju_conn.py;h=e0faaf0405c231d7e563fb74d5efa5a0b8b43f0e;hb=refs%2Fchanges%2F91%2F11291%2F7;hp=e5f150b7dcb5919aea5e25671a55af1dd9749c9a;hpb=2962f3e7aba84b4584d2deac30d1c163f6441a03;p=osm%2FN2VC.git diff --git a/n2vc/tests/unit/test_k8s_juju_conn.py b/n2vc/tests/unit/test_k8s_juju_conn.py index e5f150b..e0faaf0 100644 --- a/n2vc/tests/unit/test_k8s_juju_conn.py +++ b/n2vc/tests/unit/test_k8s_juju_conn.py @@ -17,6 +17,7 @@ import asyncio import logging import asynctest from unittest.mock import Mock +from n2vc.definitions import Offer, RelationEndpoint from n2vc.k8s_juju_conn import K8sJujuConnector, RBAC_LABEL_KEY_NAME from osm_common import fslocal from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock, FakeApplication @@ -75,25 +76,22 @@ class K8sJujuConnTestCase(asynctest.TestCase): ) 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( @@ -104,12 +102,11 @@ class InitEnvTest(K8sJujuConnTestCase): 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) @@ -117,12 +114,11 @@ class InitEnvTest(K8sJujuConnTestCase): 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 @@ -134,7 +130,6 @@ class InitEnvTest(K8sJujuConnTestCase): self.assertIsNone(created) self.assertIsNone(uuid) - mock_get_default_storage_class.assert_called_once() self.k8s_juju_conn.libjuju.add_k8s.assert_called_once() @@ -183,9 +178,6 @@ class ResetTest(K8sJujuConnTestCase): 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 @@ -614,26 +606,24 @@ class GetServicesTest(K8sJujuConnTestCase): 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): @@ -664,8 +654,9 @@ class UpdateVcaStatusTest(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() @@ -674,8 +665,9 @@ class UpdateVcaStatusTest(K8sJujuConnTestCase): 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() @@ -689,15 +681,9 @@ class ScaleTest(K8sJujuConnTestCase): self._scale = 2 self.k8s_juju_conn.libjuju.scale_application = AsyncMock() - def test_success( - self - ): + 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.scale(self.kdu_name, self._scale, self.application_name) ) self.k8s_juju_conn.libjuju.scale_application.assert_called_once() @@ -706,9 +692,7 @@ class ScaleTest(K8sJujuConnTestCase): with self.assertRaises(Exception): self.loop.run_until_complete( self.k8s_juju_conn.scale( - self.kdu_name, - self._scale, - self.application_name + self.kdu_name, self._scale, self.application_name ) ) self.k8s_juju_conn.libjuju.scale_application.assert_called_once() @@ -738,3 +722,50 @@ class GetScaleCount(K8sJujuConnTestCase): ) self.assertIsNone(status) self.k8s_juju_conn.libjuju.get_model_status.assert_called_once() + + +class AddRelationTest(K8sJujuConnTestCase): + def setUp(self): + super(AddRelationTest, self).setUp() + self.k8s_juju_conn.libjuju.add_relation = AsyncMock() + self.k8s_juju_conn.libjuju.offer = AsyncMock() + self.k8s_juju_conn.libjuju.get_controller = AsyncMock() + self.k8s_juju_conn.libjuju.consume = AsyncMock() + + def test_standard_relation(self): + relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint") + relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint") + self.loop.run_until_complete( + self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2) + ) + self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with( + model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint" + ) + self.k8s_juju_conn.libjuju.offer.assert_not_called() + self.k8s_juju_conn.libjuju.consume.assert_not_called() + + def test_cmr_relation_same_controller(self): + relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint") + relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint") + offer = Offer("admin/model-1.app1") + self.k8s_juju_conn.libjuju.offer.return_value = offer + self.k8s_juju_conn.libjuju.consume.return_value = "saas" + self.loop.run_until_complete( + self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2) + ) + self.k8s_juju_conn.libjuju.offer.assert_called_once_with(relation_endpoint_1) + self.k8s_juju_conn.libjuju.consume.assert_called_once() + self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with( + "model-2", "app2:endpoint", "saas" + ) + + def test_relation_exception(self): + relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint") + relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint") + self.k8s_juju_conn.libjuju.offer.side_effect = Exception() + with self.assertRaises(Exception): + self.loop.run_until_complete( + self.k8s_juju_conn.add_relation( + relation_endpoint_1, relation_endpoint_2 + ) + )