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 import asynctest
  19 from n2vc.n2vc_juju_conn import N2VCJujuConnector
  20 from osm_common import fslocal
  21 from n2vc.exceptions import (
  22     JujuK8sProxycharmNotSupported,
  23     N2VCBadArgumentsException,
  24     N2VCException,
  25 )
  26
  27
  28 class N2VCJujuConnTestCase(asynctest.TestCase):
  29     @asynctest.mock.patch("n2vc.libjuju.Libjuju._create_health_check_task")
  30     @asynctest.mock.patch("juju.controller.Controller.update_endpoints")
  31     @asynctest.mock.patch("juju.client.connector.Connector.connect")
  32     @asynctest.mock.patch("juju.controller.Controller.connection")
  33     @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
  34     def setUp(
  35         self,
  36         mock__get_api_endpoints_db=None,
  37         mock_connection=None,
  38         mock_connect=None,
  39         mock_update_endpoints=None,
  40         mock__create_health_check_task=None,
  41     ):
  42         mock__get_api_endpoints_db.return_value = ["2.2.2.2:17070"]
  43         loop = asyncio.get_event_loop()
  44         db = {}
  45         vca_config = {
  46             "secret": "secret",
  47             "api_proxy": "api_proxy",
  48             "cloud": "cloud",
  49             "k8s_cloud": "k8s_cloud",
  50         }
  51
  52         logging.disable(logging.CRITICAL)
  53
  54         self.n2vc = N2VCJujuConnector(
  55             db=db,
  56             fs=fslocal.FsLocal(),
  57             log=None,
  58             loop=loop,
  59             url="2.2.2.2:17070",
  60             username="admin",
  61             vca_config=vca_config,
  62             on_update_db=None,
  63         )
  64
  65
  66 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_metrics")
  67 class GetMetricssTest(N2VCJujuConnTestCase):
  68     def setUp(self):
  69         super(GetMetricssTest, self).setUp()
  70
  71     def test_success(self, mock_get_metrics):
  72         _ = self.loop.run_until_complete(self.n2vc.get_metrics("model", "application"))
  73         mock_get_metrics.assert_called_once()
  74
  75     def test_except(self, mock_get_metrics):
  76         mock_get_metrics.side_effect = Exception()
  77         with self.assertRaises(Exception):
  78             _ = self.loop.run_until_complete(
  79                 self.n2vc.get_metrics("model", "application")
  80             )
  81         mock_get_metrics.assert_called_once()
  82
  83
  84 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
  85 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
  86 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_executed_actions")
  87 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_actions")
  88 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_application_configs")
  89 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
  90 class UpdateVcaStatusTest(N2VCJujuConnTestCase):
  91     def setUp(self):
  92         super(UpdateVcaStatusTest, self).setUp()
  93
  94     def test_success(
  95         self,
  96         mock_get_application,
  97         mock_get_application_configs,
  98         mock_get_actions,
  99         mock_get_executed_actions,
 100         mock_get_model,
 101         mock_get_controller,
 102     ):
 103         self.loop.run_until_complete(self.n2vc.update_vca_status(
 104             {"model": {"applications": {"app": {"actions": {}}}}}))
 105         mock_get_executed_actions.assert_called_once()
 106         mock_get_actions.assert_called_once()
 107         mock_get_application_configs.assert_called_once()
 108
 109     def test_exception(
 110         self,
 111         mock_get_application,
 112         mock_get_application_configs,
 113         mock_get_actions,
 114         mock_get_executed_actions,
 115         mock_get_model,
 116         mock_get_controller,
 117     ):
 118         mock_get_model.return_value = None
 119         mock_get_executed_actions.side_effect = Exception()
 120         with self.assertRaises(Exception):
 121             self.loop.run_until_complete(self.n2vc.update_vca_status(
 122                 {"model": {"applications": {"app": {"actions": {}}}}}))
 123             mock_get_executed_actions.assert_not_called()
 124             mock_get_actions.assert_not_called_once()
 125             mock_get_application_configs.assert_not_called_once()
 126
 127
 128 @asynctest.mock.patch("n2vc.libjuju.Libjuju.model_exists")
 129 @asynctest.mock.patch("osm_common.fslocal.FsLocal.file_exists")
 130 @asynctest.mock.patch(
 131     "osm_common.fslocal.FsLocal.path", new_callable=asynctest.PropertyMock, create=True
 132 )
 133 @asynctest.mock.patch("n2vc.libjuju.Libjuju.deploy_charm")
 134 @asynctest.mock.patch("n2vc.libjuju.Libjuju.add_model")
 135 class K8sProxyCharmsTest(N2VCJujuConnTestCase):
 136     def setUp(self):
 137         super(K8sProxyCharmsTest, self).setUp()
 138
 139     def test_success(
 140         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists
 141     ):
 142         mock_model_exists.return_value = None
 143         mock_file_exists.return_value = True
 144         mock_path.return_value = "/path"
 145         ee_id = self.loop.run_until_complete(
 146             self.n2vc.install_k8s_proxy_charm(
 147                 "charm", "nsi-id.ns-id.vnf-id.vdu", "////path/", {},
 148             )
 149         )
 150
 151         mock_add_model.assert_called_once_with(
 152             "ns-id-k8s",
 153             cloud_name=self.n2vc.k8s_cloud,
 154             credential_name=self.n2vc.k8s_cloud
 155         )
 156         mock_deploy_charm.assert_called_once_with(
 157             model_name="ns-id-k8s",
 158             application_name="app-vnf-vnf-id-vdu-vdu",
 159             path="/path/path/",
 160             machine_id=None,
 161             db_dict={},
 162             progress_timeout=None,
 163             total_timeout=None,
 164             config=None,
 165         )
 166         self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
 167
 168     @asynctest.mock.patch(
 169         "n2vc.n2vc_juju_conn.N2VCJujuConnector.k8s_cloud",
 170         new_callable=asynctest.PropertyMock,
 171         create=True,
 172     )
 173     def test_no_k8s_cloud(
 174         self,
 175         mock_k8s_cloud,
 176         mock_add_model,
 177         mock_deploy_charm,
 178         mock_path,
 179         mock_file_exists,
 180         mock_model_exists,
 181     ):
 182         mock_k8s_cloud.return_value = None
 183         with self.assertRaises(JujuK8sProxycharmNotSupported):
 184             ee_id = self.loop.run_until_complete(
 185                 self.n2vc.install_k8s_proxy_charm(
 186                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 187                 )
 188             )
 189             self.assertIsNone(ee_id)
 190
 191     def test_no_artifact_path(
 192         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 193     ):
 194         with self.assertRaises(N2VCBadArgumentsException):
 195             ee_id = self.loop.run_until_complete(
 196                 self.n2vc.install_k8s_proxy_charm(
 197                     "charm", "nsi-id.ns-id.vnf-id.vdu", "", {},
 198                 )
 199             )
 200             self.assertIsNone(ee_id)
 201
 202     def test_no_db(
 203         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 204     ):
 205         with self.assertRaises(N2VCBadArgumentsException):
 206             ee_id = self.loop.run_until_complete(
 207                 self.n2vc.install_k8s_proxy_charm(
 208                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", None,
 209                 )
 210             )
 211             self.assertIsNone(ee_id)
 212
 213     def test_file_not_exists(
 214         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 215     ):
 216         mock_file_exists.return_value = False
 217         with self.assertRaises(N2VCBadArgumentsException):
 218             ee_id = self.loop.run_until_complete(
 219                 self.n2vc.install_k8s_proxy_charm(
 220                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 221                 )
 222             )
 223             self.assertIsNone(ee_id)
 224
 225     def test_exception(
 226         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 227     ):
 228         mock_model_exists.return_value = None
 229         mock_file_exists.return_value = True
 230         mock_path.return_value = "/path"
 231         mock_deploy_charm.side_effect = Exception()
 232         with self.assertRaises(N2VCException):
 233             ee_id = self.loop.run_until_complete(
 234                 self.n2vc.install_k8s_proxy_charm(
 235                     "charm", "nsi-id.ns-id.vnf-id.vdu", "path/", {},
 236                 )
 237             )
 238             self.assertIsNone(ee_id)