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