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("juju.controller.Controller.update_endpoints")
  30     @asynctest.mock.patch("juju.client.connector.Connector.connect")
  31     @asynctest.mock.patch("juju.controller.Controller.connection")
  32     @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
  33     def setUp(
  34         self,
  35         mock__get_api_endpoints_db=None,
  36         mock_connection=None,
  37         mock_connect=None,
  38         mock_update_endpoints=None,
  39     ):
  40         loop = asyncio.get_event_loop()
  41         db = {}
  42         vca_config = {
  43             "secret": "secret",
  44             "api_proxy": "api_proxy",
  45             "cloud": "cloud",
  46             "k8s_cloud": "k8s_cloud",
  47         }
  48
  49         logging.disable(logging.CRITICAL)
  50
  51         self.n2vc = N2VCJujuConnector(
  52             db=db,
  53             fs=fslocal.FsLocal(),
  54             log=None,
  55             loop=loop,
  56             url="2.2.2.2:17070",
  57             username="admin",
  58             vca_config=vca_config,
  59             on_update_db=None,
  60         )
  61
  62
  63 @asynctest.mock.patch("osm_common.fslocal.FsLocal.file_exists")
  64 @asynctest.mock.patch(
  65     "osm_common.fslocal.FsLocal.path", new_callable=asynctest.PropertyMock, create=True
  66 )
  67 @asynctest.mock.patch("n2vc.libjuju.Libjuju.deploy_charm")
  68 @asynctest.mock.patch("n2vc.libjuju.Libjuju.add_model")
  69 class K8sProxyCharmsTest(N2VCJujuConnTestCase):
  70     def setUp(self):
  71         super(K8sProxyCharmsTest, self).setUp()
  72
  73     def test_success(
  74         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
  75     ):
  76         mock_file_exists.return_value = True
  77         mock_path.return_value = "/path"
  78         ee_id = self.loop.run_until_complete(
  79             self.n2vc.install_k8s_proxy_charm(
  80                 "charm", "nsi-id.ns-id.vnf-id.vdu", "////path/", {},
  81             )
  82         )
  83
  84         mock_add_model.assert_called_once_with("ns-id-k8s", "k8s_cloud")
  85         mock_deploy_charm.assert_called_once_with(
  86             model_name="ns-id-k8s",
  87             application_name="app-vnf-vnf-id-vdu-vdu",
  88             path="/path/path/",
  89             machine_id=None,
  90             db_dict={},
  91             progress_timeout=None,
  92             total_timeout=None,
  93             config=None,
  94         )
  95         self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
  96
  97     @asynctest.mock.patch(
  98         "n2vc.n2vc_juju_conn.N2VCJujuConnector.k8s_cloud",
  99         new_callable=asynctest.PropertyMock,
 100         create=True,
 101     )
 102     def test_no_k8s_cloud(
 103         self,
 104         mock_k8s_cloud,
 105         mock_add_model,
 106         mock_deploy_charm,
 107         mock_path,
 108         mock_file_exists,
 109     ):
 110         mock_k8s_cloud.return_value = None
 111         with self.assertRaises(JujuK8sProxycharmNotSupported):
 112             ee_id = self.loop.run_until_complete(
 113                 self.n2vc.install_k8s_proxy_charm(
 114                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 115                 )
 116             )
 117             self.assertIsNone(ee_id)
 118
 119     def test_no_artifact_path(
 120         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
 121     ):
 122         with self.assertRaises(N2VCBadArgumentsException):
 123             ee_id = self.loop.run_until_complete(
 124                 self.n2vc.install_k8s_proxy_charm(
 125                     "charm", "nsi-id.ns-id.vnf-id.vdu", "", {},
 126                 )
 127             )
 128             self.assertIsNone(ee_id)
 129
 130     def test_no_db(
 131         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
 132     ):
 133         with self.assertRaises(N2VCBadArgumentsException):
 134             ee_id = self.loop.run_until_complete(
 135                 self.n2vc.install_k8s_proxy_charm(
 136                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", None,
 137                 )
 138             )
 139             self.assertIsNone(ee_id)
 140
 141     def test_file_not_exists(
 142         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
 143     ):
 144         mock_file_exists.return_value = False
 145         with self.assertRaises(N2VCBadArgumentsException):
 146             ee_id = self.loop.run_until_complete(
 147                 self.n2vc.install_k8s_proxy_charm(
 148                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 149                 )
 150             )
 151             self.assertIsNone(ee_id)
 152
 153     def test_exception(
 154         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
 155     ):
 156         mock_file_exists.return_value = True
 157         mock_path.return_value = "/path"
 158         mock_deploy_charm.side_effect = Exception()
 159         with self.assertRaises(N2VCException):
 160             ee_id = self.loop.run_until_complete(
 161                 self.n2vc.install_k8s_proxy_charm(
 162                     "charm", "nsi-id.ns-id.vnf-id.vdu", "path/", {},
 163                 )
 164             )
 165             self.assertIsNone(ee_id)