n2vc/tests/unit/test_n2vc_juju_conn.py

   1 # Copyright 2020 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15
  16 import asyncio
  17 import logging
  18 from unittest.mock import Mock
  19
  20
  21 import asynctest
  22 from n2vc.n2vc_juju_conn import N2VCJujuConnector
  23 from osm_common import fslocal
  24 from n2vc.exceptions import (
  25     JujuK8sProxycharmNotSupported,
  26     N2VCBadArgumentsException,
  27     N2VCException,
  28 )
  29
  30
  31 class N2VCJujuConnTestCase(asynctest.TestCase):
  32     @asynctest.mock.patch("n2vc.libjuju.Libjuju._create_health_check_task")
  33     @asynctest.mock.patch("juju.controller.Controller.update_endpoints")
  34     @asynctest.mock.patch("juju.client.connector.Connector.connect")
  35     @asynctest.mock.patch("juju.controller.Controller.connection")
  36     @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
  37     def setUp(
  38         self,
  39         mock__get_api_endpoints_db=None,
  40         mock_connection=None,
  41         mock_connect=None,
  42         mock_update_endpoints=None,
  43         mock__create_health_check_task=None,
  44     ):
  45         mock__get_api_endpoints_db.return_value = ["2.2.2.2:17070"]
  46         loop = asyncio.get_event_loop()
  47         db = {}
  48         vca_config = {
  49             "secret": "secret",
  50             "api_proxy": "api_proxy",
  51             "cloud": "cloud",
  52             "k8s_cloud": "k8s_cloud",
  53         }
  54
  55         logging.disable(logging.CRITICAL)
  56
  57         N2VCJujuConnector.get_public_key = Mock()
  58         self.n2vc = N2VCJujuConnector(
  59             db=db,
  60             fs=fslocal.FsLocal(),
  61             log=None,
  62             loop=loop,
  63             url="2.2.2.2:17070",
  64             username="admin",
  65             vca_config=vca_config,
  66             on_update_db=None,
  67         )
  68         N2VCJujuConnector.get_public_key.assert_not_called()
  69
  70
  71 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_metrics")
  72 class GetMetricssTest(N2VCJujuConnTestCase):
  73     def setUp(self):
  74         super(GetMetricssTest, self).setUp()
  75
  76     def test_success(self, mock_get_metrics):
  77         _ = self.loop.run_until_complete(self.n2vc.get_metrics("model", "application"))
  78         mock_get_metrics.assert_called_once()
  79
  80     def test_except(self, mock_get_metrics):
  81         mock_get_metrics.side_effect = Exception()
  82         with self.assertRaises(Exception):
  83             _ = self.loop.run_until_complete(
  84                 self.n2vc.get_metrics("model", "application")
  85             )
  86         mock_get_metrics.assert_called_once()
  87
  88
  89 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
  90 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
  91 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_executed_actions")
  92 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_actions")
  93 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_application_configs")
  94 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
  95 class UpdateVcaStatusTest(N2VCJujuConnTestCase):
  96     def setUp(self):
  97         super(UpdateVcaStatusTest, self).setUp()
  98
  99     def test_success(
 100         self,
 101         mock_get_application,
 102         mock_get_application_configs,
 103         mock_get_actions,
 104         mock_get_executed_actions,
 105         mock_get_model,
 106         mock_get_controller,
 107     ):
 108         self.loop.run_until_complete(self.n2vc.update_vca_status(
 109             {"model": {"applications": {"app": {"actions": {}}}}}))
 110         mock_get_executed_actions.assert_called_once()
 111         mock_get_actions.assert_called_once()
 112         mock_get_application_configs.assert_called_once()
 113
 114     def test_exception(
 115         self,
 116         mock_get_application,
 117         mock_get_application_configs,
 118         mock_get_actions,
 119         mock_get_executed_actions,
 120         mock_get_model,
 121         mock_get_controller,
 122     ):
 123         mock_get_model.return_value = None
 124         mock_get_executed_actions.side_effect = Exception()
 125         with self.assertRaises(Exception):
 126             self.loop.run_until_complete(self.n2vc.update_vca_status(
 127                 {"model": {"applications": {"app": {"actions": {}}}}}))
 128             mock_get_executed_actions.assert_not_called()
 129             mock_get_actions.assert_not_called_once()
 130             mock_get_application_configs.assert_not_called_once()
 131
 132
 133 @asynctest.mock.patch("n2vc.libjuju.Libjuju.model_exists")
 134 @asynctest.mock.patch("osm_common.fslocal.FsLocal.file_exists")
 135 @asynctest.mock.patch(
 136     "osm_common.fslocal.FsLocal.path", new_callable=asynctest.PropertyMock, create=True
 137 )
 138 @asynctest.mock.patch("n2vc.libjuju.Libjuju.deploy_charm")
 139 @asynctest.mock.patch("n2vc.libjuju.Libjuju.add_model")
 140 class K8sProxyCharmsTest(N2VCJujuConnTestCase):
 141     def setUp(self):
 142         super(K8sProxyCharmsTest, self).setUp()
 143
 144     def test_success(
 145         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists
 146     ):
 147         mock_model_exists.return_value = None
 148         mock_file_exists.return_value = True
 149         mock_path.return_value = "/path"
 150         ee_id = self.loop.run_until_complete(
 151             self.n2vc.install_k8s_proxy_charm(
 152                 "charm", "nsi-id.ns-id.vnf-id.vdu", "////path/", {},
 153             )
 154         )
 155
 156         mock_add_model.assert_called_once_with(
 157             "ns-id-k8s",
 158             cloud_name=self.n2vc.k8s_cloud,
 159             credential_name=self.n2vc.k8s_cloud
 160         )
 161         mock_deploy_charm.assert_called_once_with(
 162             model_name="ns-id-k8s",
 163             application_name="app-vnf-vnf-id-vdu-vdu",
 164             path="/path/path/",
 165             machine_id=None,
 166             db_dict={},
 167             progress_timeout=None,
 168             total_timeout=None,
 169             config=None,
 170         )
 171         self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
 172
 173     @asynctest.mock.patch(
 174         "n2vc.n2vc_juju_conn.N2VCJujuConnector.k8s_cloud",
 175         new_callable=asynctest.PropertyMock,
 176         create=True,
 177     )
 178     def test_no_k8s_cloud(
 179         self,
 180         mock_k8s_cloud,
 181         mock_add_model,
 182         mock_deploy_charm,
 183         mock_path,
 184         mock_file_exists,
 185         mock_model_exists,
 186     ):
 187         mock_k8s_cloud.return_value = None
 188         with self.assertRaises(JujuK8sProxycharmNotSupported):
 189             ee_id = self.loop.run_until_complete(
 190                 self.n2vc.install_k8s_proxy_charm(
 191                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 192                 )
 193             )
 194             self.assertIsNone(ee_id)
 195
 196     def test_no_artifact_path(
 197         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 198     ):
 199         with self.assertRaises(N2VCBadArgumentsException):
 200             ee_id = self.loop.run_until_complete(
 201                 self.n2vc.install_k8s_proxy_charm(
 202                     "charm", "nsi-id.ns-id.vnf-id.vdu", "", {},
 203                 )
 204             )
 205             self.assertIsNone(ee_id)
 206
 207     def test_no_db(
 208         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 209     ):
 210         with self.assertRaises(N2VCBadArgumentsException):
 211             ee_id = self.loop.run_until_complete(
 212                 self.n2vc.install_k8s_proxy_charm(
 213                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", None,
 214                 )
 215             )
 216             self.assertIsNone(ee_id)
 217
 218     def test_file_not_exists(
 219         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 220     ):
 221         mock_file_exists.return_value = False
 222         with self.assertRaises(N2VCBadArgumentsException):
 223             ee_id = self.loop.run_until_complete(
 224                 self.n2vc.install_k8s_proxy_charm(
 225                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 226                 )
 227             )
 228             self.assertIsNone(ee_id)
 229
 230     def test_exception(
 231         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 232     ):
 233         mock_model_exists.return_value = None
 234         mock_file_exists.return_value = True
 235         mock_path.return_value = "/path"
 236         mock_deploy_charm.side_effect = Exception()
 237         with self.assertRaises(N2VCException):
 238             ee_id = self.loop.run_until_complete(
 239                 self.n2vc.install_k8s_proxy_charm(
 240                     "charm", "nsi-id.ns-id.vnf-id.vdu", "path/", {},
 241                 )
 242             )
 243             self.assertIsNone(ee_id)