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("n2vc.libjuju.Libjuju.get_controller")
  64 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
  65 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_executed_actions")
  66 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_actions")
  67 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_application_configs")
  68 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
  69 class UpdateVcaStatusTest(N2VCJujuConnTestCase):
  70     def setUp(self):
  71         super(UpdateVcaStatusTest, self).setUp()
  72
  73     def test_success(
  74         self,
  75         mock_get_application,
  76         mock_get_application_configs,
  77         mock_get_actions,
  78         mock_get_executed_actions,
  79         mock_get_model,
  80         mock_get_controller,
  81     ):
  82         self.loop.run_until_complete(self.n2vc.update_vca_status(
  83             {"model": {"applications": {"app": {"actions": {}}}}}))
  84         mock_get_executed_actions.assert_called_once()
  85         mock_get_actions.assert_called_once()
  86         mock_get_application_configs.assert_called_once()
  87
  88     def test_exception(
  89         self,
  90         mock_get_application,
  91         mock_get_application_configs,
  92         mock_get_actions,
  93         mock_get_executed_actions,
  94         mock_get_model,
  95         mock_get_controller,
  96     ):
  97         mock_get_model.return_value = None
  98         mock_get_executed_actions.side_effect = Exception()
  99         with self.assertRaises(Exception):
 100             self.loop.run_until_complete(self.n2vc.update_vca_status(
 101                 {"model": {"applications": {"app": {"actions": {}}}}}))
 102             mock_get_executed_actions.assert_not_called()
 103             mock_get_actions.assert_not_called_once()
 104             mock_get_application_configs.assert_not_called_once()
 105
 106
 107 @asynctest.mock.patch("osm_common.fslocal.FsLocal.file_exists")
 108 @asynctest.mock.patch(
 109     "osm_common.fslocal.FsLocal.path", new_callable=asynctest.PropertyMock, create=True
 110 )
 111 @asynctest.mock.patch("n2vc.libjuju.Libjuju.deploy_charm")
 112 @asynctest.mock.patch("n2vc.libjuju.Libjuju.add_model")
 113 class K8sProxyCharmsTest(N2VCJujuConnTestCase):
 114     def setUp(self):
 115         super(K8sProxyCharmsTest, self).setUp()
 116
 117     def test_success(
 118         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
 119     ):
 120         mock_file_exists.return_value = True
 121         mock_path.return_value = "/path"
 122         ee_id = self.loop.run_until_complete(
 123             self.n2vc.install_k8s_proxy_charm(
 124                 "charm", "nsi-id.ns-id.vnf-id.vdu", "////path/", {},
 125             )
 126         )
 127
 128         mock_add_model.assert_called_once_with("ns-id-k8s", "k8s_cloud")
 129         mock_deploy_charm.assert_called_once_with(
 130             model_name="ns-id-k8s",
 131             application_name="app-vnf-vnf-id-vdu-vdu",
 132             path="/path/path/",
 133             machine_id=None,
 134             db_dict={},
 135             progress_timeout=None,
 136             total_timeout=None,
 137             config=None,
 138         )
 139         self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
 140
 141     @asynctest.mock.patch(
 142         "n2vc.n2vc_juju_conn.N2VCJujuConnector.k8s_cloud",
 143         new_callable=asynctest.PropertyMock,
 144         create=True,
 145     )
 146     def test_no_k8s_cloud(
 147         self,
 148         mock_k8s_cloud,
 149         mock_add_model,
 150         mock_deploy_charm,
 151         mock_path,
 152         mock_file_exists,
 153     ):
 154         mock_k8s_cloud.return_value = None
 155         with self.assertRaises(JujuK8sProxycharmNotSupported):
 156             ee_id = self.loop.run_until_complete(
 157                 self.n2vc.install_k8s_proxy_charm(
 158                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 159                 )
 160             )
 161             self.assertIsNone(ee_id)
 162
 163     def test_no_artifact_path(
 164         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
 165     ):
 166         with self.assertRaises(N2VCBadArgumentsException):
 167             ee_id = self.loop.run_until_complete(
 168                 self.n2vc.install_k8s_proxy_charm(
 169                     "charm", "nsi-id.ns-id.vnf-id.vdu", "", {},
 170                 )
 171             )
 172             self.assertIsNone(ee_id)
 173
 174     def test_no_db(
 175         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
 176     ):
 177         with self.assertRaises(N2VCBadArgumentsException):
 178             ee_id = self.loop.run_until_complete(
 179                 self.n2vc.install_k8s_proxy_charm(
 180                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", None,
 181                 )
 182             )
 183             self.assertIsNone(ee_id)
 184
 185     def test_file_not_exists(
 186         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
 187     ):
 188         mock_file_exists.return_value = False
 189         with self.assertRaises(N2VCBadArgumentsException):
 190             ee_id = self.loop.run_until_complete(
 191                 self.n2vc.install_k8s_proxy_charm(
 192                     "charm", "nsi-id.ns-id.vnf-id.vdu", "/path/", {},
 193                 )
 194             )
 195             self.assertIsNone(ee_id)
 196
 197     def test_exception(
 198         self, mock_add_model, mock_deploy_charm, mock_path, mock_file_exists,
 199     ):
 200         mock_file_exists.return_value = True
 201         mock_path.return_value = "/path"
 202         mock_deploy_charm.side_effect = Exception()
 203         with self.assertRaises(N2VCException):
 204             ee_id = self.loop.run_until_complete(
 205                 self.n2vc.install_k8s_proxy_charm(
 206                     "charm", "nsi-id.ns-id.vnf-id.vdu", "path/", {},
 207                 )
 208             )
 209             self.assertIsNone(ee_id)