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.definitions import Offer, RelationEndpoint
  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     def test_success(
 148         self,
 149         mock_path,
 150         mock_file_exists,
 151     ):
 152         mock_file_exists.return_value = True
 153         mock_path.return_value = "/path"
 154         ee_id = self.loop.run_until_complete(
 155             self.n2vc.install_k8s_proxy_charm(
 156                 "charm",
 157                 "nsi-id.ns-id.vnf-id.vdu",
 158                 "////path/",
 159                 {},
 160             )
 161         )
 162
 163         self.n2vc.libjuju.add_model.assert_called_once()
 164         self.n2vc.libjuju.deploy_charm.assert_called_once_with(
 165             model_name="ns-id-k8s",
 166             application_name="app-vnf-vnf-id-vdu-vdu",
 167             path="/path/path/",
 168             machine_id=None,
 169             db_dict={},
 170             progress_timeout=None,
 171             total_timeout=None,
 172             config=None,
 173         )
 174         self.assertEqual(ee_id, "ns-id-k8s.app-vnf-vnf-id-vdu-vdu.k8s")
 175
 176     def test_no_artifact_path(
 177         self,
 178         mock_path,
 179         mock_file_exists,
 180     ):
 181         with self.assertRaises(N2VCBadArgumentsException):
 182             ee_id = self.loop.run_until_complete(
 183                 self.n2vc.install_k8s_proxy_charm(
 184                     "charm",
 185                     "nsi-id.ns-id.vnf-id.vdu",
 186                     "",
 187                     {},
 188                 )
 189             )
 190             self.assertIsNone(ee_id)
 191
 192     def test_no_db(
 193         self,
 194         mock_path,
 195         mock_file_exists,
 196     ):
 197         with self.assertRaises(N2VCBadArgumentsException):
 198             ee_id = self.loop.run_until_complete(
 199                 self.n2vc.install_k8s_proxy_charm(
 200                     "charm",
 201                     "nsi-id.ns-id.vnf-id.vdu",
 202                     "/path/",
 203                     None,
 204                 )
 205             )
 206             self.assertIsNone(ee_id)
 207
 208     def test_file_not_exists(
 209         self,
 210         mock_path,
 211         mock_file_exists,
 212     ):
 213         mock_file_exists.return_value = False
 214         with self.assertRaises(N2VCBadArgumentsException):
 215             ee_id = self.loop.run_until_complete(
 216                 self.n2vc.install_k8s_proxy_charm(
 217                     "charm",
 218                     "nsi-id.ns-id.vnf-id.vdu",
 219                     "/path/",
 220                     {},
 221                 )
 222             )
 223             self.assertIsNone(ee_id)
 224
 225     def test_exception(
 226         self,
 227         mock_path,
 228         mock_file_exists,
 229     ):
 230         mock_file_exists.return_value = True
 231         mock_path.return_value = "/path"
 232         self.n2vc.libjuju.deploy_charm.side_effect = Exception()
 233         with self.assertRaises(N2VCException):
 234             ee_id = self.loop.run_until_complete(
 235                 self.n2vc.install_k8s_proxy_charm(
 236                     "charm",
 237                     "nsi-id.ns-id.vnf-id.vdu",
 238                     "path/",
 239                     {},
 240                 )
 241             )
 242             self.assertIsNone(ee_id)
 243
 244
 245 class AddRelationTest(N2VCJujuConnTestCase):
 246     def setUp(self):
 247         super(AddRelationTest, self).setUp()
 248         self.n2vc.libjuju.add_relation = AsyncMock()
 249         self.n2vc.libjuju.offer = AsyncMock()
 250         self.n2vc.libjuju.get_controller = AsyncMock()
 251         self.n2vc.libjuju.consume = AsyncMock()
 252
 253     def test_standard_relation(self):
 254         relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
 255         relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint")
 256         self.loop.run_until_complete(
 257             self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
 258         )
 259         self.n2vc.libjuju.add_relation.assert_called_once_with(
 260             model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint"
 261         )
 262         self.n2vc.libjuju.offer.assert_not_called()
 263         self.n2vc.libjuju.consume.assert_not_called()
 264
 265     def test_cmr_relation_same_controller(self):
 266         relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
 267         relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
 268         offer = Offer("admin/model-1.app1")
 269         self.n2vc.libjuju.offer.return_value = offer
 270         self.n2vc.libjuju.consume.return_value = "saas"
 271         self.loop.run_until_complete(
 272             self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
 273         )
 274         self.n2vc.libjuju.offer.assert_called_once_with(relation_endpoint_1)
 275         self.n2vc.libjuju.consume.assert_called_once()
 276         self.n2vc.libjuju.add_relation.assert_called_once_with(
 277             "model-2", "app2:endpoint", "saas"
 278         )
 279
 280     def test_relation_exception(self):
 281         relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
 282         relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
 283         self.n2vc.libjuju.offer.side_effect = Exception()
 284         with self.assertRaises(N2VCException):
 285             self.loop.run_until_complete(
 286                 self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
 287             )