n2vc/tests/unit/test_temporal_libjuju.py

   1 #######################################################################################
   2 # Copyright ETSI Contributors and Others.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #    http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 import asyncio
  18 import asynctest
  19 import logging
  20 import juju
  21 from unittest.mock import Mock
  22
  23 from n2vc.temporal_libjuju import Libjuju, ConnectionInfo
  24 from n2vc.exceptions import (
  25     JujuApplicationExists,
  26     JujuControllerFailedConnecting,
  27     JujuModelAlreadyExists,
  28 )
  29
  30 cacert = """-----BEGIN CERTIFICATE-----
  31 SOMECERT
  32 -----END CERTIFICATE-----"""
  33
  34
  35 @asynctest.mock.patch("n2vc.temporal_libjuju.Controller")
  36 class LibjujuTestCase(asynctest.TestCase):
  37     def setUp(
  38         self,
  39         mock_base64_to_cacert=None,
  40     ):
  41         self.loop = asyncio.get_event_loop()
  42         self.connection_info = ConnectionInfo(
  43             "1.2.3.4:17070", "user", "secret", cacert, "k8s_cloud", "k8s_credentials"
  44         )
  45         logging.disable(logging.CRITICAL)
  46         self.libjuju = Libjuju(self.connection_info)
  47
  48
  49 @asynctest.mock.patch("juju.controller.Controller.connect")
  50 class GetControllerTest(LibjujuTestCase):
  51     def setUp(self):
  52         super(GetControllerTest, self).setUp()
  53
  54     def test_get_controller(self, mock_connect):
  55         controller = self.loop.run_until_complete(self.libjuju.get_controller())
  56         self.assertIsInstance(controller, juju.controller.Controller)
  57         mock_connect.assert_called_with(
  58             endpoint=self.connection_info.endpoint,
  59             username=self.connection_info.user,
  60             password=self.connection_info.password,
  61             cacert=self.connection_info.cacert,
  62         )
  63
  64     @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
  65     def test_exception(
  66         self,
  67         mock_disconnect_controller,
  68         mock_connect,
  69     ):
  70         mock_connect.side_effect = Exception()
  71         controller = None
  72         with self.assertRaises(JujuControllerFailedConnecting):
  73             controller = self.loop.run_until_complete(self.libjuju.get_controller())
  74         self.assertIsNone(controller)
  75         mock_disconnect_controller.assert_called()
  76
  77
  78 class DisconnectTest(LibjujuTestCase):
  79     def setUp(self):
  80         super(DisconnectTest, self).setUp()
  81
  82     @asynctest.mock.patch("juju.model.Model.disconnect")
  83     def test_disconnect_model(self, mock_disconnect):
  84         self.loop.run_until_complete(self.libjuju.disconnect_model(juju.model.Model()))
  85         mock_disconnect.assert_called_once()
  86
  87     @asynctest.mock.patch("juju.controller.Controller.disconnect")
  88     def test_disconnect_controller(self, mock_disconnect):
  89         self.loop.run_until_complete(
  90             self.libjuju.disconnect_controller(juju.controller.Controller())
  91         )
  92         mock_disconnect.assert_called_once()
  93
  94
  95 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
  96 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.model_exists")
  97 @asynctest.mock.patch("juju.controller.Controller.add_model")
  98 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
  99 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_model")
 100 class AddModelTest(LibjujuTestCase):
 101     def setUp(self):
 102         super(AddModelTest, self).setUp()
 103
 104     def test_existing_model_raises_exception(
 105         self,
 106         mock_disconnect_model,
 107         mock_disconnect_controller,
 108         mock_add_model,
 109         mock_model_exists,
 110         mock_get_controller,
 111     ):
 112         mock_model_exists.return_value = True
 113         mock_get_controller.return_value = juju.controller.Controller()
 114         with self.assertRaises(JujuModelAlreadyExists):
 115             self.loop.run_until_complete(self.libjuju.add_model("existing_model"))
 116         mock_get_controller.assert_called()
 117         mock_add_model.assert_not_called()
 118         mock_disconnect_controller.assert_called()
 119
 120     def test_non_existing_model(
 121         self,
 122         mock_disconnect_model,
 123         mock_disconnect_controller,
 124         mock_add_model,
 125         mock_model_exists,
 126         mock_get_controller,
 127     ):
 128         mock_model_exists.return_value = False
 129         mock_get_controller.return_value = juju.controller.Controller()
 130         new_model_name = "nonexisting_model"
 131         self.loop.run_until_complete(self.libjuju.add_model(new_model_name))
 132         mock_add_model.assert_called_once_with(
 133             new_model_name,
 134             cloud_name=self.connection_info.cloud_name,
 135             credential_name=self.connection_info.cloud_credentials,
 136         )
 137         mock_disconnect_controller.assert_called()
 138         mock_disconnect_model.assert_called()
 139
 140
 141 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
 142 @asynctest.mock.patch("juju.controller.Controller.list_models")
 143 class ModelExistsTest(LibjujuTestCase):
 144     def setUp(self):
 145         super(ModelExistsTest, self).setUp()
 146
 147     async def test_existing_model(
 148         self,
 149         mock_list_models,
 150         mock_get_controller,
 151     ):
 152         model_name = "existing_model"
 153         mock_list_models.return_value = [model_name]
 154         self.assertTrue(
 155             await self.libjuju.model_exists(model_name, juju.controller.Controller())
 156         )
 157
 158     @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
 159     async def test_no_controller(
 160         self,
 161         mock_disconnect_controller,
 162         mock_list_models,
 163         mock_get_controller,
 164     ):
 165         model_name = "existing_model"
 166         mock_list_models.return_value = [model_name]
 167         mock_get_controller.return_value = juju.controller.Controller()
 168         self.assertTrue(await self.libjuju.model_exists(model_name))
 169         mock_disconnect_controller.assert_called_once()
 170
 171     async def test_non_existing_model(
 172         self,
 173         mock_list_models,
 174         mock_get_controller,
 175     ):
 176         mock_list_models.return_value = []
 177         self.assertFalse(
 178             await self.libjuju.model_exists(
 179                 "not_existing_model", juju.controller.Controller()
 180             )
 181         )
 182
 183
 184 @asynctest.mock.patch("juju.controller.Controller.get_model")
 185 class GetModelTest(LibjujuTestCase):
 186     def setUp(self):
 187         super(GetModelTest, self).setUp()
 188
 189     def test_get_model(
 190         self,
 191         mock_get_model,
 192     ):
 193         mock_get_model.return_value = juju.model.Model()
 194         model = self.loop.run_until_complete(
 195             self.libjuju.get_model(juju.controller.Controller(), "model")
 196         )
 197         self.assertIsInstance(model, juju.model.Model)
 198
 199
 200 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
 201 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
 202 @asynctest.mock.patch("juju.controller.Controller.list_models")
 203 class ListModelsTest(LibjujuTestCase):
 204     def setUp(self):
 205         super(ListModelsTest, self).setUp()
 206
 207     def test_containing(
 208         self,
 209         mock_list_models,
 210         mock_disconnect_controller,
 211         mock_get_controller,
 212     ):
 213         expected_list = ["existingmodel"]
 214         mock_get_controller.return_value = juju.controller.Controller()
 215         mock_list_models.return_value = expected_list
 216         models = self.loop.run_until_complete(self.libjuju.list_models())
 217
 218         mock_disconnect_controller.assert_called_once()
 219         self.assertEquals(models, expected_list)
 220
 221
 222 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
 223 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_model")
 224 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_model")
 225 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
 226 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.wait_app_deployment_completion")
 227 @asynctest.mock.patch(
 228     "juju.model.Model.applications", new_callable=asynctest.PropertyMock
 229 )
 230 @asynctest.mock.patch("juju.model.Model.deploy")
 231 class DeployCharmTest(LibjujuTestCase):
 232     def setUp(self):
 233         super(DeployCharmTest, self).setUp()
 234
 235     def test_existing_app_is_not_deployed(
 236         self,
 237         mock_deploy,
 238         mock_applications,
 239         mock_wait_app_deployment,
 240         mock_disconnect_controller,
 241         mock_disconnect_model,
 242         mock_get_model,
 243         mock_get_controller,
 244     ):
 245         mock_get_model.return_value = juju.model.Model()
 246         mock_applications.return_value = {"existing_app"}
 247
 248         application = None
 249         with self.assertRaises(JujuApplicationExists):
 250             application = self.loop.run_until_complete(
 251                 self.libjuju.deploy_charm(
 252                     "existing_app",
 253                     "path",
 254                     "model",
 255                     "machine",
 256                 )
 257             )
 258         self.assertIsNone(application)
 259
 260         mock_disconnect_controller.assert_called()
 261         mock_disconnect_model.assert_called()
 262
 263     def test_app_is_deployed(
 264         self,
 265         mock_deploy,
 266         mock_applications,
 267         mock_wait_app_deployment,
 268         mock_disconnect_controller,
 269         mock_disconnect_model,
 270         mock_get_model,
 271         mock_get_controller,
 272     ):
 273         mock_get_model.return_value = juju.model.Model()
 274         mock_deploy.return_value = Mock()
 275         self.loop.run_until_complete(
 276             self.libjuju.deploy_charm(
 277                 "app",
 278                 "path",
 279                 "model",
 280                 series="series",
 281                 num_units=2,
 282             )
 283         )
 284         mock_deploy.assert_called_once_with(
 285             entity_url="path",
 286             application_name="app",
 287             channel="stable",
 288             num_units=2,
 289             series="series",
 290             config=None,
 291         )
 292         mock_wait_app_deployment.assert_called()
 293         mock_disconnect_controller.assert_called()
 294         mock_disconnect_model.assert_called()
 295
 296
 297 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
 298 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_model")
 299 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.model_exists")
 300 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_model")
 301 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
 302 @asynctest.mock.patch("juju.controller.Controller.destroy_model")
 303 class DestroyModelTest(LibjujuTestCase):
 304     def setUp(self):
 305         super(DestroyModelTest, self).setUp()
 306
 307     def test_model_is_destroyed(
 308         self,
 309         mock_destroy,
 310         mock_disconnect_controller,
 311         mock_disconnect_model,
 312         mock_model_exists,
 313         mock_get_model,
 314         mock_get_controller,
 315     ):
 316         mock_get_controller.return_value = juju.controller.Controller()
 317         mock_model_exists.return_value = True
 318         mock_get_model.return_value = juju.model.Model()
 319         model_name = "model_to_destroy"
 320         force = True
 321
 322         self.loop.run_until_complete(self.libjuju.destroy_model(model_name, force))
 323         mock_destroy.assert_called_with(
 324             model_name, destroy_storage=True, force=force, max_wait=60
 325         )
 326         mock_disconnect_controller.assert_called()
 327         mock_disconnect_model.assert_called()
 328
 329     def test_not_existing_model(
 330         self,
 331         mock_destroy,
 332         mock_disconnect_controller,
 333         mock_disconnect_model,
 334         mock_model_exists,
 335         mock_get_model,
 336         mock_get_controller,
 337     ):
 338         mock_get_controller.return_value = juju.controller.Controller()
 339         mock_model_exists.return_value = False
 340         model_name = "model_to_destroy"
 341         force = True
 342
 343         self.loop.run_until_complete(self.libjuju.destroy_model(model_name, force))
 344         mock_destroy.assert_not_called()
 345         mock_get_model.assert_not_called()
 346         mock_disconnect_controller.assert_called()
 347         mock_disconnect_model.assert_called()
 348
 349     def test_raise_exception(
 350         self,
 351         mock_destroy,
 352         mock_disconnect_controller,
 353         mock_disconnect_model,
 354         mock_model_exists,
 355         mock_get_model,
 356         mock_get_controller,
 357     ):
 358         mock_get_controller.return_value = juju.controller.Controller()
 359         mock_model_exists.return_value = True
 360         mock_get_model.side_effect = Exception()
 361         model_name = "model_to_destroy"
 362         force = False
 363
 364         with self.assertRaises(Exception):
 365             self.loop.run_until_complete(self.libjuju.destroy_model(model_name, force))
 366         mock_destroy.assert_not_called()
 367         mock_disconnect_controller.assert_called()
 368         mock_disconnect_model.assert_called()