X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_n2vc_juju_conn.py;fp=n2vc%2Ftests%2Funit%2Ftest_n2vc_juju_conn.py;h=d89de3fb77d538ed144ef6b2482045d5631ce794;hp=e5e26be1d291ce3bda04cace2a51313ad84da615;hb=eb8943a887e2fb8cce0240382811f9e504f3c7fb;hpb=6331b04745fcd6d44b1b0320ca6e3e63cdebd0e8 diff --git a/n2vc/tests/unit/test_n2vc_juju_conn.py b/n2vc/tests/unit/test_n2vc_juju_conn.py index e5e26be..d89de3f 100644 --- a/n2vc/tests/unit/test_n2vc_juju_conn.py +++ b/n2vc/tests/unit/test_n2vc_juju_conn.py @@ -22,143 +22,145 @@ import asynctest 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("n2vc.libjuju.Libjuju._create_health_check_task") - @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__create_health_check_task=None, + mock_base64_to_cacert=None, + mock_get_connection=None, + mock_store=None, ): - mock__get_api_endpoints_db.return_value = ["2.2.2.2:17070"] - 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() -@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_metrics") class GetMetricssTest(N2VCJujuConnTestCase): def setUp(self): super(GetMetricssTest, self).setUp() + self.n2vc.libjuju.get_metrics = AsyncMock() - def test_success(self, mock_get_metrics): + def test_success(self): _ = self.loop.run_until_complete(self.n2vc.get_metrics("model", "application")) - mock_get_metrics.assert_called_once() + self.n2vc.libjuju.get_metrics.assert_called_once() - def test_except(self, mock_get_metrics): - mock_get_metrics.side_effect = Exception() + 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") ) - mock_get_metrics.assert_called_once() + self.n2vc.libjuju.get_metrics.assert_called_once() -@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") -@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") -@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_executed_actions") -@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_actions") -@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_application_configs") -@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") 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, - mock_get_application, - mock_get_application_configs, - mock_get_actions, - mock_get_executed_actions, - mock_get_model, - mock_get_controller, ): - self.loop.run_until_complete(self.n2vc.update_vca_status( - {"model": {"applications": {"app": {"actions": {}}}}})) - mock_get_executed_actions.assert_called_once() - mock_get_actions.assert_called_once() - mock_get_application_configs.assert_called_once() + 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, - mock_get_application, - mock_get_application_configs, - mock_get_actions, - mock_get_executed_actions, - mock_get_model, - mock_get_controller, - ): - mock_get_model.return_value = None - mock_get_executed_actions.side_effect = Exception() + 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": {}}}}})) - mock_get_executed_actions.assert_not_called() - mock_get_actions.assert_not_called_once() - mock_get_application_configs.assert_not_called_once() + 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("n2vc.libjuju.Libjuju.model_exists") @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 def test_success( - self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists + self, + mock_path, + mock_file_exists, ): - mock_model_exists.return_value = None 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", - cloud_name=self.n2vc.k8s_cloud, - credential_name=self.n2vc.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", path="/path/path/", @@ -170,74 +172,70 @@ class K8sProxyCharmsTest(N2VCJujuConnTestCase): ) self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.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_model_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, mock_model_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, mock_model_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, mock_model_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, mock_model_exists, + self, + mock_path, + mock_file_exists, ): - mock_model_exists.return_value = None 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)