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