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.definitions import Offer, RelationEndpoint
  24 from n2vc.n2vc_juju_conn import N2VCJujuConnector
  25 from osm_common import fslocal
  26 from n2vc.exceptions import (
  27     N2VCBadArgumentsException,
  28     N2VCException,
  29     JujuApplicationNotFound,
  30 )
  31 from n2vc.tests.unit.utils import AsyncMock
  32 from n2vc.vca.connection_data import ConnectionData
  33
  34
  35 class N2VCJujuConnTestCase(asynctest.TestCase):
  36     @asynctest.mock.patch("n2vc.n2vc_juju_conn.MotorStore")
  37     @asynctest.mock.patch("n2vc.n2vc_juju_conn.get_connection")
  38     @asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert")
  39     def setUp(
  40         self,
  41         mock_base64_to_cacert=None,
  42         mock_get_connection=None,
  43         mock_store=None,
  44     ):
  45         self.loop = asyncio.get_event_loop()
  46         self.db = Mock()
  47         mock_base64_to_cacert.return_value = """
  48     -----BEGIN CERTIFICATE-----
  49     SOMECERT
  50     -----END CERTIFICATE-----"""
  51         mock_store.return_value = AsyncMock()
  52         mock_vca_connection = Mock()
  53         mock_get_connection.return_value = mock_vca_connection
  54         mock_vca_connection.data.return_value = ConnectionData(
  55             **{
  56                 "endpoints": ["1.2.3.4:17070"],
  57                 "user": "user",
  58                 "secret": "secret",
  59                 "cacert": "cacert",
  60                 "pubkey": "pubkey",
  61                 "lxd-cloud": "cloud",
  62                 "lxd-credentials": "credentials",
  63                 "k8s-cloud": "k8s_cloud",
  64                 "k8s-credentials": "k8s_credentials",
  65                 "model-config": {},
  66                 "api-proxy": "api_proxy",
  67             }
  68         )
  69         logging.disable(logging.CRITICAL)
  70
  71         N2VCJujuConnector.get_public_key = Mock()
  72         self.n2vc = N2VCJujuConnector(
  73             db=self.db,
  74             fs=fslocal.FsLocal(),
  75             log=None,
  76             loop=self.loop,
  77             on_update_db=None,
  78         )
  79         N2VCJujuConnector.get_public_key.assert_not_called()
  80         self.n2vc.libjuju = Mock()
  81
  82
  83 class GetMetricssTest(N2VCJujuConnTestCase):
  84     def setUp(self):
  85         super(GetMetricssTest, self).setUp()
  86         self.n2vc.libjuju.get_metrics = AsyncMock()
  87
  88     def test_success(self):
  89         _ = self.loop.run_until_complete(self.n2vc.get_metrics("model", "application"))
  90         self.n2vc.libjuju.get_metrics.assert_called_once()
  91
  92     def test_except(self):
  93         self.n2vc.libjuju.get_metrics.side_effect = Exception()
  94         with self.assertRaises(Exception):
  95             _ = self.loop.run_until_complete(
  96                 self.n2vc.get_metrics("model", "application")
  97             )
  98         self.n2vc.libjuju.get_metrics.assert_called_once()
  99
 100
 101 class UpdateVcaStatusTest(N2VCJujuConnTestCase):
 102     def setUp(self):
 103         super(UpdateVcaStatusTest, self).setUp()
 104         self.n2vc.libjuju.get_controller = AsyncMock()
 105         self.n2vc.libjuju.get_model = AsyncMock()
 106         self.n2vc.libjuju.get_executed_actions = AsyncMock()
 107         self.n2vc.libjuju.get_actions = AsyncMock()
 108         self.n2vc.libjuju.get_application_configs = AsyncMock()
 109         self.n2vc.libjuju._get_application = AsyncMock()
 110
 111     def test_success(
 112         self,
 113     ):
 114         self.loop.run_until_complete(
 115             self.n2vc.update_vca_status(
 116                 {"model": {"applications": {"app": {"actions": {}}}}}
 117             )
 118         )
 119         self.n2vc.libjuju.get_executed_actions.assert_called_once()
 120         self.n2vc.libjuju.get_actions.assert_called_once()
 121         self.n2vc.libjuju.get_application_configs.assert_called_once()
 122
 123     def test_exception(self):
 124         self.n2vc.libjuju.get_model.return_value = None
 125         self.n2vc.libjuju.get_executed_actions.side_effect = Exception()
 126         with self.assertRaises(Exception):
 127             self.loop.run_until_complete(
 128                 self.n2vc.update_vca_status(
 129                     {"model": {"applications": {"app": {"actions": {}}}}}
 130                 )
 131             )
 132             self.n2vc.libjuju.get_executed_actions.assert_not_called()
 133             self.n2vc.libjuju.get_actions.assert_not_called_once()
 134             self.n2vc.libjuju.get_application_configs.assert_not_called_once()
 135
 136
 137 @asynctest.mock.patch("osm_common.fslocal.FsLocal.file_exists")
 138 @asynctest.mock.patch(
 139     "osm_common.fslocal.FsLocal.path", new_callable=asynctest.PropertyMock, create=True
 140 )
 141 class K8sProxyCharmsTest(N2VCJujuConnTestCase):
 142     def setUp(self):
 143         super(K8sProxyCharmsTest, self).setUp()
 144         self.n2vc.libjuju.model_exists = AsyncMock()
 145         self.n2vc.libjuju.add_model = AsyncMock()
 146         self.n2vc.libjuju.deploy_charm = AsyncMock()
 147         self.n2vc.libjuju.model_exists.return_value = False
 148
 149     @patch(
 150         "n2vc.n2vc_juju_conn.generate_random_alfanum_string",
 151         **{"return_value": "random"}
 152     )
 153     def test_success(
 154         self,
 155         mock_generate_random_alfanum_string,
 156         mock_path,
 157         mock_file_exists,
 158     ):
 159         mock_file_exists.return_value = True
 160         mock_path.return_value = "/path"
 161         ee_id = self.loop.run_until_complete(
 162             self.n2vc.install_k8s_proxy_charm(
 163                 "charm",
 164                 "nsi-id.ns-id.vnf-id.vdu",
 165                 "////path/",
 166                 {},
 167             )
 168         )
 169
 170         self.n2vc.libjuju.add_model.assert_called_once()
 171         self.n2vc.libjuju.deploy_charm.assert_called_once_with(
 172             model_name="ns-id-k8s",
 173             application_name="app-vnf-vnf-id-vdu-vdu-random",
 174             path="/path/path/",
 175             machine_id=None,
 176             db_dict={},
 177             progress_timeout=None,
 178             total_timeout=None,
 179             config=None,
 180         )
 181         self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu-random.k8s")
 182
 183     def test_no_artifact_path(
 184         self,
 185         mock_path,
 186         mock_file_exists,
 187     ):
 188         with self.assertRaises(N2VCBadArgumentsException):
 189             ee_id = self.loop.run_until_complete(
 190                 self.n2vc.install_k8s_proxy_charm(
 191                     "charm",
 192                     "nsi-id.ns-id.vnf-id.vdu",
 193                     "",
 194                     {},
 195                 )
 196             )
 197             self.assertIsNone(ee_id)
 198
 199     def test_no_db(
 200         self,
 201         mock_path,
 202         mock_file_exists,
 203     ):
 204         with self.assertRaises(N2VCBadArgumentsException):
 205             ee_id = self.loop.run_until_complete(
 206                 self.n2vc.install_k8s_proxy_charm(
 207                     "charm",
 208                     "nsi-id.ns-id.vnf-id.vdu",
 209                     "/path/",
 210                     None,
 211                 )
 212             )
 213             self.assertIsNone(ee_id)
 214
 215     def test_file_not_exists(
 216         self,
 217         mock_path,
 218         mock_file_exists,
 219     ):
 220         mock_file_exists.return_value = False
 221         with self.assertRaises(N2VCBadArgumentsException):
 222             ee_id = self.loop.run_until_complete(
 223                 self.n2vc.install_k8s_proxy_charm(
 224                     "charm",
 225                     "nsi-id.ns-id.vnf-id.vdu",
 226                     "/path/",
 227                     {},
 228                 )
 229             )
 230             self.assertIsNone(ee_id)
 231
 232     def test_exception(
 233         self,
 234         mock_path,
 235         mock_file_exists,
 236     ):
 237         mock_file_exists.return_value = True
 238         mock_path.return_value = "/path"
 239         self.n2vc.libjuju.deploy_charm.side_effect = Exception()
 240         with self.assertRaises(N2VCException):
 241             ee_id = self.loop.run_until_complete(
 242                 self.n2vc.install_k8s_proxy_charm(
 243                     "charm",
 244                     "nsi-id.ns-id.vnf-id.vdu",
 245                     "path/",
 246                     {},
 247                 )
 248             )
 249             self.assertIsNone(ee_id)
 250
 251
 252 class AddRelationTest(N2VCJujuConnTestCase):
 253     def setUp(self):
 254         super(AddRelationTest, self).setUp()
 255         self.n2vc.libjuju.add_relation = AsyncMock()
 256         self.n2vc.libjuju.offer = AsyncMock()
 257         self.n2vc.libjuju.get_controller = AsyncMock()
 258         self.n2vc.libjuju.consume = AsyncMock()
 259
 260     def test_standard_relation(self):
 261         relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
 262         relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint")
 263         self.loop.run_until_complete(
 264             self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
 265         )
 266         self.n2vc.libjuju.add_relation.assert_called_once_with(
 267             model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint"
 268         )
 269         self.n2vc.libjuju.offer.assert_not_called()
 270         self.n2vc.libjuju.consume.assert_not_called()
 271
 272     def test_cmr_relation_same_controller(self):
 273         relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
 274         relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
 275         offer = Offer("admin/model-1.app1")
 276         self.n2vc.libjuju.offer.return_value = offer
 277         self.n2vc.libjuju.consume.return_value = "saas"
 278         self.loop.run_until_complete(
 279             self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
 280         )
 281         self.n2vc.libjuju.offer.assert_called_once_with(relation_endpoint_1)
 282         self.n2vc.libjuju.consume.assert_called_once()
 283         self.n2vc.libjuju.add_relation.assert_called_once_with(
 284             "model-2", "app2:endpoint", "saas"
 285         )
 286
 287     def test_relation_exception(self):
 288         relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
 289         relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
 290         self.n2vc.libjuju.offer.side_effect = Exception()
 291         with self.assertRaises(N2VCException):
 292             self.loop.run_until_complete(
 293                 self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
 294             )
 295
 296
 297 class UpgradeCharmTest(N2VCJujuConnTestCase):
 298     def setUp(self):
 299         super(UpgradeCharmTest, self).setUp()
 300         self.n2vc._get_libjuju = AsyncMock(return_value=self.n2vc.libjuju)
 301         N2VCJujuConnector._get_ee_id_components = Mock()
 302         self.n2vc.libjuju.upgrade_charm = AsyncMock()
 303
 304     def test_empty_ee_id(self):
 305         with self.assertRaises(N2VCBadArgumentsException):
 306             self.loop.run_until_complete(
 307                 self.n2vc.upgrade_charm(
 308                     "", "/sample_charm_path", "sample_charm_id", "native-charm", None
 309                 )
 310             )
 311         self.n2vc._get_libjuju.assert_called()
 312         self.n2vc._get_ee_id_components.assert_not_called()
 313         self.n2vc.libjuju.upgrade_charm.assert_not_called()
 314
 315     def test_wrong_ee_id(self):
 316         N2VCJujuConnector._get_ee_id_components.side_effect = Exception
 317         with self.assertRaises(N2VCBadArgumentsException):
 318             self.loop.run_until_complete(
 319                 self.n2vc.upgrade_charm(
 320                     "ns-id-k8s.app-vnf-vnf-id-vdu-vdu-random.k8s",
 321                     "/sample_charm_path",
 322                     "sample_charm_id",
 323                     "native-charm",
 324                     500,
 325                 )
 326             )
 327         self.n2vc._get_libjuju.assert_called()
 328         self.n2vc._get_ee_id_components.assert_called()
 329         self.n2vc.libjuju.upgrade_charm.assert_not_called()
 330
 331     def test_charm_upgrade_succeded(self):
 332         N2VCJujuConnector._get_ee_id_components.return_value = (
 333             "sample_model",
 334             "sample_app",
 335             "sample_machine_id",
 336         )
 337         self.loop.run_until_complete(
 338             self.n2vc.upgrade_charm(
 339                 "ns-id-k8s.app-vnf-vnf-id-vdu-vdu-random.k8s",
 340                 "/sample_charm_path",
 341                 "sample_charm_id",
 342                 "native-charm",
 343                 500,
 344             )
 345         )
 346         self.n2vc._get_libjuju.assert_called()
 347         self.n2vc._get_ee_id_components.assert_called()
 348         self.n2vc.libjuju.upgrade_charm.assert_called_with(
 349             application_name="sample_app",
 350             path="/sample_charm_path",
 351             model_name="sample_model",
 352             total_timeout=500,
 353         )
 354
 355     def test_charm_upgrade_failed(self):
 356         N2VCJujuConnector._get_ee_id_components.return_value = (
 357             "sample_model",
 358             "sample_app",
 359             "sample_machine_id",
 360         )
 361         self.n2vc.libjuju.upgrade_charm.side_effect = JujuApplicationNotFound
 362         with self.assertRaises(N2VCException):
 363             self.loop.run_until_complete(
 364                 self.n2vc.upgrade_charm(
 365                     "ns-id-k8s.app-vnf-vnf-id-vdu-vdu-random.k8s",
 366                     "/sample_charm_path",
 367                     "sample_charm_id",
 368                     "native-charm",
 369                     None,
 370                 )
 371             )
 372         self.n2vc._get_libjuju.assert_called()
 373         self.n2vc._get_ee_id_components.assert_called()
 374         self.n2vc.libjuju.upgrade_charm.assert_called_with(
 375             application_name="sample_app",
 376             path="/sample_charm_path",
 377             model_name="sample_model",
 378             total_timeout=None,
 379         )