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
  22 from n2vc.temporal_libjuju import Libjuju, ConnectionInfo
  23 from n2vc.exceptions import (
  24     JujuControllerFailedConnecting,
  25     JujuModelAlreadyExists,
  26 )
  27
  28 cacert = """-----BEGIN CERTIFICATE-----
  29 SOMECERT
  30 -----END CERTIFICATE-----"""
  31
  32
  33 @asynctest.mock.patch("n2vc.temporal_libjuju.Controller")
  34 class LibjujuTestCase(asynctest.TestCase):
  35     def setUp(
  36         self,
  37         mock_base64_to_cacert=None,
  38     ):
  39         self.loop = asyncio.get_event_loop()
  40         self.connection_info = ConnectionInfo(
  41             "1.2.3.4:17070", "user", "secret", cacert, "k8s_cloud", "k8s_credentials"
  42         )
  43         logging.disable(logging.CRITICAL)
  44         self.libjuju = Libjuju(self.connection_info)
  45
  46
  47 @asynctest.mock.patch("juju.controller.Controller.connect")
  48 class GetControllerTest(LibjujuTestCase):
  49     def setUp(self):
  50         super(GetControllerTest, self).setUp()
  51
  52     def test_get_controller(self, mock_connect):
  53         controller = self.loop.run_until_complete(self.libjuju.get_controller())
  54         self.assertIsInstance(controller, juju.controller.Controller)
  55         mock_connect.assert_called_with(
  56             endpoint=self.connection_info.endpoint,
  57             username=self.connection_info.user,
  58             password=self.connection_info.password,
  59             cacert=self.connection_info.cacert,
  60         )
  61
  62     @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
  63     def test_exception(
  64         self,
  65         mock_disconnect_controller,
  66         mock_connect,
  67     ):
  68         mock_connect.side_effect = Exception()
  69         controller = None
  70         with self.assertRaises(JujuControllerFailedConnecting):
  71             controller = self.loop.run_until_complete(self.libjuju.get_controller())
  72         self.assertIsNone(controller)
  73         mock_disconnect_controller.assert_called()
  74
  75
  76 class DisconnectTest(LibjujuTestCase):
  77     def setUp(self):
  78         super(DisconnectTest, self).setUp()
  79
  80     @asynctest.mock.patch("juju.model.Model.disconnect")
  81     def test_disconnect_model(self, mock_disconnect):
  82         self.loop.run_until_complete(self.libjuju.disconnect_model(juju.model.Model()))
  83         mock_disconnect.assert_called_once()
  84
  85     @asynctest.mock.patch("juju.controller.Controller.disconnect")
  86     def test_disconnect_controller(self, mock_disconnect):
  87         self.loop.run_until_complete(
  88             self.libjuju.disconnect_controller(juju.controller.Controller())
  89         )
  90         mock_disconnect.assert_called_once()
  91
  92
  93 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
  94 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.model_exists")
  95 @asynctest.mock.patch("juju.controller.Controller.add_model")
  96 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
  97 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_model")
  98 class AddModelTest(LibjujuTestCase):
  99     def setUp(self):
 100         super(AddModelTest, self).setUp()
 101
 102     def test_existing_model_raises_exception(
 103         self,
 104         mock_disconnect_model,
 105         mock_disconnect_controller,
 106         mock_add_model,
 107         mock_model_exists,
 108         mock_get_controller,
 109     ):
 110         mock_model_exists.return_value = True
 111         mock_get_controller.return_value = juju.controller.Controller()
 112         with self.assertRaises(JujuModelAlreadyExists):
 113             self.loop.run_until_complete(self.libjuju.add_model("existing_model"))
 114         mock_get_controller.assert_called()
 115         mock_add_model.assert_not_called()
 116         mock_disconnect_controller.assert_called()
 117
 118     def test_non_existing_model(
 119         self,
 120         mock_disconnect_model,
 121         mock_disconnect_controller,
 122         mock_add_model,
 123         mock_model_exists,
 124         mock_get_controller,
 125     ):
 126         mock_model_exists.return_value = False
 127         mock_get_controller.return_value = juju.controller.Controller()
 128         new_model_name = "nonexisting_model"
 129         self.loop.run_until_complete(self.libjuju.add_model(new_model_name))
 130         mock_add_model.assert_called_once_with(
 131             new_model_name,
 132             cloud_name=self.connection_info.cloud_name,
 133             credential_name=self.connection_info.cloud_credentials,
 134         )
 135         mock_disconnect_controller.assert_called()
 136         mock_disconnect_model.assert_called()
 137
 138
 139 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
 140 @asynctest.mock.patch("juju.controller.Controller.list_models")
 141 class ModelExistsTest(LibjujuTestCase):
 142     def setUp(self):
 143         super(ModelExistsTest, self).setUp()
 144
 145     async def test_existing_model(
 146         self,
 147         mock_list_models,
 148         mock_get_controller,
 149     ):
 150         model_name = "existing_model"
 151         mock_list_models.return_value = [model_name]
 152         self.assertTrue(
 153             await self.libjuju.model_exists(model_name, juju.controller.Controller())
 154         )
 155
 156     @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
 157     async def test_no_controller(
 158         self,
 159         mock_disconnect_controller,
 160         mock_list_models,
 161         mock_get_controller,
 162     ):
 163         model_name = "existing_model"
 164         mock_list_models.return_value = [model_name]
 165         mock_get_controller.return_value = juju.controller.Controller()
 166         self.assertTrue(await self.libjuju.model_exists(model_name))
 167         mock_disconnect_controller.assert_called_once()
 168
 169     async def test_non_existing_model(
 170         self,
 171         mock_list_models,
 172         mock_get_controller,
 173     ):
 174         mock_list_models.return_value = []
 175         self.assertFalse(
 176             await self.libjuju.model_exists(
 177                 "not_existing_model", juju.controller.Controller()
 178             )
 179         )
 180
 181
 182 @asynctest.mock.patch("juju.controller.Controller.get_model")
 183 class GetModelTest(LibjujuTestCase):
 184     def setUp(self):
 185         super(GetModelTest, self).setUp()
 186
 187     def test_get_model(
 188         self,
 189         mock_get_model,
 190     ):
 191         mock_get_model.return_value = juju.model.Model()
 192         model = self.loop.run_until_complete(
 193             self.libjuju.get_model(juju.controller.Controller(), "model")
 194         )
 195         self.assertIsInstance(model, juju.model.Model)
 196
 197
 198 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
 199 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
 200 @asynctest.mock.patch("juju.controller.Controller.list_models")
 201 class ListModelsTest(LibjujuTestCase):
 202     def setUp(self):
 203         super(ListModelsTest, self).setUp()
 204
 205     def test_containing(
 206         self,
 207         mock_list_models,
 208         mock_disconnect_controller,
 209         mock_get_controller,
 210     ):
 211         expected_list = ["existingmodel"]
 212         mock_get_controller.return_value = juju.controller.Controller()
 213         mock_list_models.return_value = expected_list
 214         models = self.loop.run_until_complete(self.libjuju.list_models())
 215
 216         mock_disconnect_controller.assert_called_once()
 217         self.assertEquals(models, expected_list)
 218
 219
 220 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_controller")
 221 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.get_model")
 222 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.model_exists")
 223 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_model")
 224 @asynctest.mock.patch("n2vc.temporal_libjuju.Libjuju.disconnect_controller")
 225 @asynctest.mock.patch("juju.controller.Controller.destroy_model")
 226 class DestroyModelTest(LibjujuTestCase):
 227     def setUp(self):
 228         super(DestroyModelTest, self).setUp()
 229
 230     def test_model_is_destroyed(
 231         self,
 232         mock_destroy,
 233         mock_disconnect_controller,
 234         mock_disconnect_model,
 235         mock_model_exists,
 236         mock_get_model,
 237         mock_get_controller,
 238     ):
 239         mock_get_controller.return_value = juju.controller.Controller()
 240         mock_model_exists.return_value = True
 241         mock_get_model.return_value = juju.model.Model()
 242         model_name = "model_to_destroy"
 243         force = True
 244
 245         self.loop.run_until_complete(self.libjuju.destroy_model(model_name, force))
 246         mock_destroy.assert_called_with(
 247             model_name, destroy_storage=True, force=force, max_wait=60
 248         )
 249         mock_disconnect_controller.assert_called()
 250         mock_disconnect_model.assert_called()
 251
 252     def test_not_existing_model(
 253         self,
 254         mock_destroy,
 255         mock_disconnect_controller,
 256         mock_disconnect_model,
 257         mock_model_exists,
 258         mock_get_model,
 259         mock_get_controller,
 260     ):
 261         mock_get_controller.return_value = juju.controller.Controller()
 262         mock_model_exists.return_value = False
 263         model_name = "model_to_destroy"
 264         force = True
 265
 266         self.loop.run_until_complete(self.libjuju.destroy_model(model_name, force))
 267         mock_destroy.assert_not_called()
 268         mock_get_model.assert_not_called()
 269         mock_disconnect_controller.assert_called()
 270         mock_disconnect_model.assert_called()
 271
 272     def test_raise_exception(
 273         self,
 274         mock_destroy,
 275         mock_disconnect_controller,
 276         mock_disconnect_model,
 277         mock_model_exists,
 278         mock_get_model,
 279         mock_get_controller,
 280     ):
 281         mock_get_controller.return_value = juju.controller.Controller()
 282         mock_model_exists.return_value = True
 283         mock_get_model.side_effect = Exception()
 284         model_name = "model_to_destroy"
 285         force = False
 286
 287         with self.assertRaises(Exception):
 288             self.loop.run_until_complete(self.libjuju.destroy_model(model_name, force))
 289         mock_destroy.assert_not_called()
 290         mock_disconnect_controller.assert_called()
 291         mock_disconnect_model.assert_called()