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.model_exists")
  85 @asynctest.mock.patch("osm_common.fslocal.FsLocal.file_exists")
  86 @asynctest.mock.patch(
  87     "osm_common.fslocal.FsLocal.path", new_callable=asynctest.PropertyMock, create=True
  88 )
  89 @asynctest.mock.patch("n2vc.libjuju.Libjuju.deploy_charm")
  90 @asynctest.mock.patch("n2vc.libjuju.Libjuju.add_model")
  91 class K8sProxyCharmsTest(N2VCJujuConnTestCase):
  92     def setUp(self):
  93         super(K8sProxyCharmsTest, self).setUp()
  94
  95     def test_success(
  96         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists
  97     ):
  98         mock_model_exists.return_value = None
  99         mock_file_exists.return_value = True
 100         mock_path.return_value = "/path"
 101         ee_id = self.loop.run_until_complete(
 102             self.n2vc.install_k8s_proxy_charm(
 103                 "charm", "nsi-id.ns-id.vnf-id.vdu", "////path/", {},
 104             )
 105         )
 106
 107         mock_add_model.assert_called_once_with(
 108             "ns-id-k8s",
 109             cloud_name=self.n2vc.k8s_cloud,
 110             credential_name=self.n2vc.k8s_cloud
 111         )
 112         mock_deploy_charm.assert_called_once_with(
 113             model_name="ns-id-k8s",
 114             application_name="app-vnf-vnf-id-vdu-vdu",
 115             path="/path/path/",
 116             machine_id=None,
 117             db_dict={},
 118             progress_timeout=None,
 119             total_timeout=None,
 120             config=None,
 121         )
 122         self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
 123
 124     @asynctest.mock.patch(
 125         "n2vc.n2vc_juju_conn.N2VCJujuConnector.k8s_cloud",
 126         new_callable=asynctest.PropertyMock,
 127         create=True,
 128     )
 129     def test_no_k8s_cloud(
 130         self,
 131         mock_k8s_cloud,
 132         mock_add_model,
 133         mock_deploy_charm,
 134         mock_path,
 135         mock_file_exists,
 136         mock_model_exists,
 137     ):
 138         mock_k8s_cloud.return_value = None
 139         with self.assertRaises(JujuK8sProxycharmNotSupported):
 140             ee_id = self.loop.run_until_complete(
 141                 self.n2vc.install_k8s_proxy_charm(
 142                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 143                 )
 144             )
 145             self.assertIsNone(ee_id)
 146
 147     def test_no_artifact_path(
 148         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 149     ):
 150         with self.assertRaises(N2VCBadArgumentsException):
 151             ee_id = self.loop.run_until_complete(
 152                 self.n2vc.install_k8s_proxy_charm(
 153                     "charm", "nsi-id.ns-id.vnf-id.vdu", "", {},
 154                 )
 155             )
 156             self.assertIsNone(ee_id)
 157
 158     def test_no_db(
 159         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 160     ):
 161         with self.assertRaises(N2VCBadArgumentsException):
 162             ee_id = self.loop.run_until_complete(
 163                 self.n2vc.install_k8s_proxy_charm(
 164                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", None,
 165                 )
 166             )
 167             self.assertIsNone(ee_id)
 168
 169     def test_file_not_exists(
 170         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 171     ):
 172         mock_file_exists.return_value = False
 173         with self.assertRaises(N2VCBadArgumentsException):
 174             ee_id = self.loop.run_until_complete(
 175                 self.n2vc.install_k8s_proxy_charm(
 176                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 177                 )
 178             )
 179             self.assertIsNone(ee_id)
 180
 181     def test_exception(
 182         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists, mock_model_exists,
 183     ):
 184         mock_model_exists.return_value = None
 185         mock_file_exists.return_value = True
 186         mock_path.return_value = "/path"
 187         mock_deploy_charm.side_effect = Exception()
 188         with self.assertRaises(N2VCException):
 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)