X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_n2vc_juju_conn.py;h=3caae030e416e0842b3459d3d3997c3276cf8d3e;hp=80c4c8775384ee7d7dee776fd4671383e9e45755;hb=0ebadd80cf20e478677b1af7b57a3faaefcc239a;hpb=0c478257d6bd8126b27d80f76d128c7cc21d0609 diff --git a/n2vc/tests/unit/test_n2vc_juju_conn.py b/n2vc/tests/unit/test_n2vc_juju_conn.py index 80c4c87..3caae03 100644 --- a/n2vc/tests/unit/test_n2vc_juju_conn.py +++ b/n2vc/tests/unit/test_n2vc_juju_conn.py @@ -15,76 +15,161 @@ import asyncio import logging +from unittest.mock import Mock +from unittest.mock import patch + + import asynctest +from n2vc.definitions import Offer, RelationEndpoint from n2vc.n2vc_juju_conn import N2VCJujuConnector from osm_common import fslocal from n2vc.exceptions import ( - JujuK8sProxycharmNotSupported, N2VCBadArgumentsException, N2VCException, ) +from n2vc.tests.unit.utils import AsyncMock +from n2vc.vca.connection_data import ConnectionData class N2VCJujuConnTestCase(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.libjuju.Libjuju._get_api_endpoints_db") + @asynctest.mock.patch("n2vc.n2vc_juju_conn.MotorStore") + @asynctest.mock.patch("n2vc.n2vc_juju_conn.get_connection") + @asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert") def setUp( self, - mock__get_api_endpoints_db=None, - mock_connection=None, - mock_connect=None, - mock_update_endpoints=None, + mock_base64_to_cacert=None, + mock_get_connection=None, + mock_store=None, ): - loop = asyncio.get_event_loop() - db = {} - vca_config = { - "secret": "secret", - "api_proxy": "api_proxy", - "cloud": "cloud", - "k8s_cloud": "k8s_cloud", - } - + self.loop = asyncio.get_event_loop() + self.db = Mock() + mock_base64_to_cacert.return_value = """ + -----BEGIN CERTIFICATE----- + SOMECERT + -----END CERTIFICATE-----""" + 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) + N2VCJujuConnector.get_public_key = Mock() self.n2vc = N2VCJujuConnector( - db=db, + db=self.db, fs=fslocal.FsLocal(), log=None, - loop=loop, - url="2.2.2.2:17070", - username="admin", - vca_config=vca_config, + loop=self.loop, on_update_db=None, ) + N2VCJujuConnector.get_public_key.assert_not_called() + self.n2vc.libjuju = Mock() + + +class GetMetricssTest(N2VCJujuConnTestCase): + def setUp(self): + super(GetMetricssTest, self).setUp() + self.n2vc.libjuju.get_metrics = AsyncMock() + + def test_success(self): + _ = self.loop.run_until_complete(self.n2vc.get_metrics("model", "application")) + self.n2vc.libjuju.get_metrics.assert_called_once() + + def test_except(self): + self.n2vc.libjuju.get_metrics.side_effect = Exception() + with self.assertRaises(Exception): + _ = self.loop.run_until_complete( + self.n2vc.get_metrics("model", "application") + ) + self.n2vc.libjuju.get_metrics.assert_called_once() + + +class UpdateVcaStatusTest(N2VCJujuConnTestCase): + def setUp(self): + super(UpdateVcaStatusTest, self).setUp() + self.n2vc.libjuju.get_controller = AsyncMock() + self.n2vc.libjuju.get_model = AsyncMock() + self.n2vc.libjuju.get_executed_actions = AsyncMock() + self.n2vc.libjuju.get_actions = AsyncMock() + self.n2vc.libjuju.get_application_configs = AsyncMock() + self.n2vc.libjuju._get_application = AsyncMock() + + def test_success( + self, + ): + self.loop.run_until_complete( + self.n2vc.update_vca_status( + {"model": {"applications": {"app": {"actions": {}}}}} + ) + ) + self.n2vc.libjuju.get_executed_actions.assert_called_once() + self.n2vc.libjuju.get_actions.assert_called_once() + self.n2vc.libjuju.get_application_configs.assert_called_once() + + def test_exception(self): + self.n2vc.libjuju.get_model.return_value = None + self.n2vc.libjuju.get_executed_actions.side_effect = Exception() + with self.assertRaises(Exception): + self.loop.run_until_complete( + self.n2vc.update_vca_status( + {"model": {"applications": {"app": {"actions": {}}}}} + ) + ) + self.n2vc.libjuju.get_executed_actions.assert_not_called() + self.n2vc.libjuju.get_actions.assert_not_called_once() + self.n2vc.libjuju.get_application_configs.assert_not_called_once() @asynctest.mock.patch("osm_common.fslocal.FsLocal.file_exists") @asynctest.mock.patch( "osm_common.fslocal.FsLocal.path", new_callable=asynctest.PropertyMock, create=True ) -@asynctest.mock.patch("n2vc.libjuju.Libjuju.deploy_charm") -@asynctest.mock.patch("n2vc.libjuju.Libjuju.add_model") class K8sProxyCharmsTest(N2VCJujuConnTestCase): def setUp(self): super(K8sProxyCharmsTest, self).setUp() + self.n2vc.libjuju.model_exists = AsyncMock() + self.n2vc.libjuju.add_model = AsyncMock() + self.n2vc.libjuju.deploy_charm = AsyncMock() + self.n2vc.libjuju.model_exists.return_value = False + @patch( + "n2vc.n2vc_juju_conn.generate_random_alfanum_string", + **{"return_value": "random"} + ) def test_success( - self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, + self, + mock_generate_random_alfanum_string, + mock_path, + mock_file_exists, ): mock_file_exists.return_value = True mock_path.return_value = "/path" ee_id = self.loop.run_until_complete( self.n2vc.install_k8s_proxy_charm( - "charm", "nsi-id.ns-id.vnf-id.vdu", "////path/", {}, + "charm", + "nsi-id.ns-id.vnf-id.vdu", + "////path/", + {}, ) ) - mock_add_model.assert_called_once_with("ns-id-k8s", "k8s_cloud") - mock_deploy_charm.assert_called_once_with( + self.n2vc.libjuju.add_model.assert_called_once() + self.n2vc.libjuju.deploy_charm.assert_called_once_with( model_name="ns-id-k8s", - application_name="app-vnf-vnf-id-vdu-vdu", + application_name="app-vnf-vnf-id-vdu-vdu-random", path="/path/path/", machine_id=None, db_dict={}, @@ -92,74 +177,117 @@ class K8sProxyCharmsTest(N2VCJujuConnTestCase): total_timeout=None, config=None, ) - self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s") + self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu-random.k8s") - @asynctest.mock.patch( - "n2vc.n2vc_juju_conn.N2VCJujuConnector.k8s_cloud", - new_callable=asynctest.PropertyMock, - create=True, - ) - def test_no_k8s_cloud( + def test_no_artifact_path( self, - mock_k8s_cloud, - mock_add_model, - mock_deploy_charm, mock_path, mock_file_exists, - ): - mock_k8s_cloud.return_value = None - with self.assertRaises(JujuK8sProxycharmNotSupported): - ee_id = self.loop.run_until_complete( - self.n2vc.install_k8s_proxy_charm( - "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {}, - ) - ) - self.assertIsNone(ee_id) - - def test_no_artifact_path( - self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, ): with self.assertRaises(N2VCBadArgumentsException): ee_id = self.loop.run_until_complete( self.n2vc.install_k8s_proxy_charm( - "charm", "nsi-id.ns-id.vnf-id.vdu", "", {}, + "charm", + "nsi-id.ns-id.vnf-id.vdu", + "", + {}, ) ) self.assertIsNone(ee_id) def test_no_db( - self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, + self, + mock_path, + mock_file_exists, ): with self.assertRaises(N2VCBadArgumentsException): ee_id = self.loop.run_until_complete( self.n2vc.install_k8s_proxy_charm( - "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", None, + "charm", + "nsi-id.ns-id.vnf-id.vdu", + "/path/", + None, ) ) self.assertIsNone(ee_id) def test_file_not_exists( - self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, + self, + mock_path, + mock_file_exists, ): mock_file_exists.return_value = False with self.assertRaises(N2VCBadArgumentsException): ee_id = self.loop.run_until_complete( self.n2vc.install_k8s_proxy_charm( - "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {}, + "charm", + "nsi-id.ns-id.vnf-id.vdu", + "/path/", + {}, ) ) self.assertIsNone(ee_id) def test_exception( - self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, + self, + mock_path, + mock_file_exists, ): mock_file_exists.return_value = True mock_path.return_value = "/path" - mock_deploy_charm.side_effect = Exception() + self.n2vc.libjuju.deploy_charm.side_effect = Exception() with self.assertRaises(N2VCException): ee_id = self.loop.run_until_complete( self.n2vc.install_k8s_proxy_charm( - "charm", "nsi-id.ns-id.vnf-id.vdu", "path/", {}, + "charm", + "nsi-id.ns-id.vnf-id.vdu", + "path/", + {}, ) ) self.assertIsNone(ee_id) + + +class AddRelationTest(N2VCJujuConnTestCase): + def setUp(self): + super(AddRelationTest, self).setUp() + self.n2vc.libjuju.add_relation = AsyncMock() + self.n2vc.libjuju.offer = AsyncMock() + self.n2vc.libjuju.get_controller = AsyncMock() + self.n2vc.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.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2) + ) + self.n2vc.libjuju.add_relation.assert_called_once_with( + model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint" + ) + self.n2vc.libjuju.offer.assert_not_called() + self.n2vc.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.n2vc.libjuju.offer.return_value = offer + self.n2vc.libjuju.consume.return_value = "saas" + self.loop.run_until_complete( + self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2) + ) + self.n2vc.libjuju.offer.assert_called_once_with(relation_endpoint_1) + self.n2vc.libjuju.consume.assert_called_once() + self.n2vc.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.n2vc.libjuju.offer.side_effect = Exception() + with self.assertRaises(N2VCException): + self.loop.run_until_complete( + self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2) + )