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